一、核心流程必须先搞清楚普通 LLM 调用是一问一答Tool Calling 是两轮第一轮你 → 模型附带工具列表 模型 → 你我要调用 get_weather参数是 city北京 你执行工具拿到结果 第二轮你 → 模型把工具结果发回去 模型 → 你北京今天晴22度适合外出模型不会自己执行工具。它只是告诉你我想调用哪个工具、传什么参数真正执行是你的代码结果再喂回给模型。二、最小完整示例pythonfrom openai import OpenAI from dotenv import load_dotenv import json load_dotenv() client OpenAI() # ── 第一步定义工具列表 ────────────────────────────────────── tools [ { type: function, function: { name: get_weather, description: ( 获取指定城市的当前天气。 适用于用户询问天气、出行建议、穿衣建议。 返回温度、天气状况、湿度。 ), parameters: { type: object, properties: { city: { type: string, description: 城市名例如北京、上海、广州 }, unit: { type: string, enum: [celsius, fahrenheit], description: 温度单位默认 celsius } }, required: [city], additionalProperties: False } } } ] # ── 第二步真正执行工具的函数你来实现──────────────────── def get_weather(city: str, unit: str celsius) - dict: # 实际项目里这里调用天气 API # 现在先用假数据演示 return { city: city, temperature: 22 if unit celsius else 72, condition: 晴天, humidity: 45%, unit: unit, } # ── 第三步工具注册表名字 → 函数的映射─────────────────── TOOL_REGISTRY { get_weather: get_weather, } # ── 第四步完整的两轮调用 ──────────────────────────────────── def chat_with_tools(user_message: str) - str: messages [{role: user, content: user_message}] # 第一轮让模型决定要不要调工具 response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) choice response.choices[0] # 模型直接回答没有调工具 if choice.finish_reason stop: return choice.message.content # 模型要调工具 if choice.finish_reason tool_calls: # 把模型的我要调工具这条消息加入历史 messages.append(choice.message) # 执行每一个工具调用可能同时调多个 for tool_call in choice.message.tool_calls: func_name tool_call.function.name func_args json.loads(tool_call.function.arguments) print(f[执行工具] {func_name}({func_args})) result TOOL_REGISTRY[func_name](**func_args) # 把工具结果加入历史 messages.append({ role: tool, tool_call_id: tool_call.id, # 必须和 tool_call.id 对应 content: json.dumps(result, ensure_asciiFalse), }) # 第二轮模型看到工具结果生成最终回答 final_response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) return final_response.choices[0].message.content return 未知的 finish_reason # 测试 print(chat_with_tools(北京今天天气怎么样要带伞吗)) # [执行工具] get_weather({city: 北京}) # 北京今天晴天气温22°C湿度45%不需要带伞适合外出。三、工具 Schema 详解Schema 是模型理解工具的唯一依据写得好不好直接影响调用准确率。字段说明python{ type: function, # 固定写 function function: { name: search_kb, # 函数名用 snake_case # 模型靠这个名字找到对应函数 description: ..., # ★ 最关键的字段 # 模型靠这段描述决定要不要调这个工具 # 写清楚做什么、适用场景、不适用场景 parameters: { type: object, # 固定写 object properties: { param_name: { type: string, # 类型 description: ..., # 参数含义给例子更好 enum: [a, b], # 枚举限定可选值 default: medium, # 默认值仅说明用不强制 } }, required: [必填参数名], # 必填字段列表 additionalProperties: False # 不允许额外字段生产必加 } } }好描述 vs 差描述对比python# ✗ 差模糊模型不知道什么时候该调 { name: search, description: 搜索文档, parameters: { type: object, properties: { query: {type: string} }, required: [query] } } # ✓ 好说清楚适用场景、不适用场景、返回内容 { name: search_knowledge_base, description: ( 在企业内部知识库中搜索相关文档。 适用于查询公司政策、规章制度、产品说明、技术文档。 不适用于实时数据查询用 query_database 代替。 返回最相关的文档片段及来源文件名。 ), parameters: { type: object, properties: { query: { type: string, description: 自然语言搜索词例如年假政策 或 报销流程 }, top_k: { type: integer, description: 返回结果数量1到10之间默认3, default: 3 } }, required: [query], additionalProperties: False } }用 Pydantic 自动生成 Schema工程化写法pythonfrom pydantic import BaseModel, Field from typing import Optional class SearchKBParams(BaseModel): query: str Field(description自然语言搜索词例如年假政策) top_k: int Field(default3, description返回数量1-10之间, ge1, le10) class QueryDBParams(BaseModel): table: str Field(description表名tasks/users/documents) filters: dict Field(default{}, description过滤条件例如{status: done}) limit: int Field(default10, description返回行数上限, ge1, le100) def make_tool(name: str, description: str, params_class: type[BaseModel]) - dict: 从 Pydantic 模型自动生成工具定义 schema params_class.model_json_schema() return { type: function, function: { name: name, description: description, parameters: { type: object, properties: schema.get(properties, {}), required: schema.get(required, []), additionalProperties: False, }, }, } # 使用 tools [ make_tool( search_knowledge_base, 在企业知识库中搜索文档。适用于政策、规程、技术文档查询。, SearchKBParams, ), make_tool( query_database, 查询业务数据库。适用于获取任务、用户、订单等实时数据。, QueryDBParams, ), ]四、多工具并行调用模型可以在一次响应里同时调多个工具要处理好pythondef run_with_multiple_tools(user_message: str) - str: messages [{role: user, content: user_message}] response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) choice response.choices[0] messages.append(choice.message) if choice.finish_reason tool_calls: # 可能同时调多个工具全部执行完再进入第二轮 for tool_call in choice.message.tool_calls: name tool_call.function.name args json.loads(tool_call.function.arguments) if name not in TOOL_REGISTRY: result {error: f未知工具: {name}} else: try: result TOOL_REGISTRY[name](**args) except Exception as e: result {error: str(e)} messages.append({ role: tool, tool_call_id: tool_call.id, content: json.dumps(result, ensure_asciiFalse), }) # 所有工具结果都加进去再调一次 final client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) return final.choices[0].message.content return choice.message.content # 测试 # 查一下北京和上海今天的天气哪个更适合出行 # 模型会同时调用两次 get_weather分别查北京和上海五、异步版本FastAPI 里用这个pythonfrom openai import AsyncOpenAI import asyncio async_client AsyncOpenAI() async def execute_tool(name: str, args: dict) - dict: 统一的工具执行器处理异步工具 ASYNC_TOOLS { search_knowledge_base: search_knowledge_base, # async 函数 query_database: query_database, # async 函数 get_weather: get_weather, # 同步函数 } func ASYNC_TOOLS.get(name) if not func: return {error: f未知工具: {name}} try: # 同步函数用 asyncio.to_thread 包一下 if asyncio.iscoroutinefunction(func): return await func(**args) else: return await asyncio.to_thread(func, **args) except Exception as e: return {error: f工具执行失败: {str(e)}} async def agent_loop(user_message: str) - str: Agent 主循环支持多轮工具调用 messages [{role: user, content: user_message}] max_rounds 10 # 防死循环 for _ in range(max_rounds): response await async_client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) choice response.choices[0] messages.append(choice.message) # 模型决定直接回答 if choice.finish_reason stop: return choice.message.content # 模型要调工具 if choice.finish_reason tool_calls: # 并发执行所有工具asyncio.gather 同时跑 tool_results await asyncio.gather(*[ execute_tool( tc.function.name, json.loads(tc.function.arguments) ) for tc in choice.message.tool_calls ]) # 把结果加入历史 for tc, result in zip(choice.message.tool_calls, tool_results): messages.append({ role: tool, tool_call_id: tc.id, content: json.dumps(result, ensure_asciiFalse), }) # 继续循环让模型处理结果 return 已达到最大工具调用次数 # 在 FastAPI 里 app.post(/agent) async def agent_endpoint(message: str): result await agent_loop(message) return {answer: result}六、实战给 Task Manager 加工具调用把阶段1的 Task Manager 和工具调用串起来——用自然语言操作任务python# app/services/agent.py tools [ { type: function, function: { name: list_tasks, description: 查询当前用户的任务列表支持按状态过滤。适用于查看任务、统计进度。, parameters: { type: object, properties: { status: { type: string, enum: [todo, in_progress, done], description: 按状态过滤不传则返回全部 } }, required: [], additionalProperties: False } } }, { type: function, function: { name: create_task, description: 创建新任务。适用于用户说添加任务、新建任务、我要做某件事。, parameters: { type: object, properties: { title: {type: string, description: 任务标题}, priority: {type: string, enum: [low, medium, high]}, due_date: {type: string, description: 截止日期 YYYY-MM-DD没有则不传} }, required: [title], additionalProperties: False } } }, { type: function, function: { name: complete_task, description: 将指定任务标记为完成。适用于用户说完成了、做完了、标记为已完成。, parameters: { type: object, properties: { task_title: {type: string, description: 要完成的任务标题模糊匹配} }, required: [task_title], additionalProperties: False } } }, ] async def task_agent(user_message: str, user_id: int, db) - str: 自然语言操作任务的 Agent # 工具实现闭包捕获 user_id 和 db async def list_tasks(status: str None): rows await db.fetch( SELECT id, title, status, priority FROM tasks WHERE user_id$1 ( AND status$2 if status else ), *([user_id, status] if status else [user_id]) ) return [dict(r) for r in rows] async def create_task(title: str, priority: str medium, due_date: str None): row await db.fetchrow( INSERT INTO tasks (user_id,title,priority,due_date) VALUES ($1,$2,$3,$4) RETURNING id,title,priority, user_id, title, priority, due_date ) return dict(row) async def complete_task(task_title: str): row await db.fetchrow( UPDATE tasks SET statusdone WHERE user_id$1 AND title ILIKE $2 RETURNING id,title, user_id, f%{task_title}% ) return dict(row) if row else {error: 未找到匹配的任务} registry { list_tasks: list_tasks, create_task: create_task, complete_task: complete_task, } messages [ {role: system, content: f你是任务管理助手帮用户管理他们的任务列表。今天是 {datetime.now().date()}。}, {role: user, content: user_message}, ] for _ in range(5): response await async_client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolstools, ) choice response.choices[0] messages.append(choice.message) if choice.finish_reason stop: return choice.message.content if choice.finish_reason tool_calls: for tc in choice.message.tool_calls: args json.loads(tc.function.arguments) result await registry[tc.function.name](**args) messages.append({ role: tool, tool_call_id: tc.id, content: json.dumps(result, ensure_asciiFalse), }) return 处理超时 # 路由 router.post(/agent, summary自然语言操作任务) async def task_agent_endpoint( message: str, user_id: int Depends(_get_user_id), dbDepends(get_db), ): result await task_agent(message, user_id, db) return {answer: result} # 测试 # 我今天完成了哪些任务还有什么没做 # 帮我添加一个高优先级任务下周五前完成API文档 # 把写单元测试那个任务标记为完成七、常见坑tool_call.function.arguments是字符串不是 dict— 必须json.loads()才能用直接当 dict 用会报错。多个工具调用时tool_call_id必须一一对应— 每个工具结果消息的tool_call_id必须和触发它的tool_call.id完全一致否则模型会混乱。工具失败要返回有意义的错误信息— 不要抛出异常让调用中断而是返回{error: 原因}让模型能理解并告知用户。description 是调优的核心— 模型调错工具 80% 是因为 description 写得不够清晰。改 description 通常比改代码更有效。掌握了 Tool Calling下一步是把多个工具组合进LangGraph让模型自主决定调用顺序和是否需要多步推理——也就是路线图里的阶段4 Agent