FastAPI Agent 函数调用实战:我让 AI 学会了“自己动手查天气“
这就是典型的“空脑子”问题——大模型如果没有手就只能靠记忆瞎掰。今天咱们就给 Agent 装上双手让它能自己查天气、算数字、搜资料。全程用 FastAPI Pydantic 函数调用不套重型框架手动实现一个清晰可控的工具执行模式。 本文能帮你解决的把只能“纯聊天”的 Agent 升级成 能调用真实函数干活的智能助手并且建立一套结构化的工具定义、选择、执行和回退机制。你会得到一个可扩展的工具链骨架往后加新功能就像插积木一样顺手。 核心脉络 案例引子一个瞎编天气的 Agent 有多坑 工具定义的艺术用 Pydantic 给函数写“说明书” 意图解析与工具选择LLM 吐出调用指令后怎么做 异步执行器 错误降级超时、参数错、网络断都不慌 实战端点改造从 /chat 到 /agent只多几十行代码 第一部分空脑子 Agent 的翻车现场上篇我们把 Ollama 接进了 FastAPI一个 /chat 端点走天下。但如果“你问他厦门明天什么天气他回得有鼻子有眼但一查根本没那回事。”这太正常了因为 模型没有调用外部工具的能力就像把厨师锁在没有食材的厨房里只能给你画菜。解决办法就是 函数调用Function Calling告诉模型嘿你没这本事就别硬编了遇到这类问题直接告诉我要调哪个工具我帮你去跑。 第二部分核心原理——把函数变成模型能看懂的菜单整个思路特别像给大厨配一本标准化菜谱 菜谱工具定义 → 用 Pydantic 模型描述函数名、作用、参数类型 大厨看单意图解析 → LLM 读到用户消息返回想调用的函数名和参数 真正炒菜执行器 → 我们的代码接收指令去调用真实的天气 API 或计算函数️ 上菜回传结果 → 把工具返回的数据再喂给模型生成最终的自然语言回答好咱们先来定义工具。在 FastAPI 里我习惯用一个 Pydantic BaseModel 来描述每个工具的输入参数。比如天气查询from pydantic import BaseModelclass WeatherParams(BaseModel):city: strdate: str “today”然后把工具元信息统一注册到一个列表里。这里注意尽量别把所有函数都塞进一个散装 dict参数一多就乱套调试到凌晨三点才发现是因为忘传了 required 字段。下面给出两种定义方式作为对比参考。另外 特别提醒工具元信息注册格式一定要规范否则大模型识别不到你的工具函数。再就是 确保你拉取的模型支持工具函数调用前面我用的 qwen3:0.6b 结果就是不调用工具换成 qwen3.5:0.8b 立马识别到工具了那个心酸呀后面多轮对话时又出现第二轮和第一轮回复相同内容的情况一通折腾最后找到原因是小模型做不了轮对话至少要是 7b 以上吃了电脑配置不够的亏TOOLS [{“type”: “function”,“function”: {“name”: “get_weather”,“description”: “查询指定城市天气”,“parameters”: WeatherParams.model_json_schema()}},{“type”: “function”,“function”: {“name”: “calculate”,“description”: “执行数学计算”,“parameters”: {“type”: “object”,“properties”: {“expression”: {“type”: “string”, “description”: “数学表达式”}},“required”: [“expression”]}}}]⚡ 第三部分实战——让 /agent 端点“长出双手”接下来咱们改造之前的 /chat变成 /agent。核心变化是把用户的请求先发给大模型但这次要多传一个 tools 参数告诉它你有这些家伙可以用。模型会返回一个类似 {“name”:“get_weather”,“arguments”:{“city”:“杭州”}} 的指令。我在 main.py 里新增了一个异步函数专门处理这件事async def execute_tool(tool_name: str, arguments: dict):if tool_name “get_weather”:# 这里接你们实际的天气 API我暂时用模拟数据city arguments.get(“city”, “未知”)return f{city}今天晴朗26.8°C适合写代码elif tool_name “calculate”:expr arguments.get(“expression”, “”)try:result eval(expr) # 仅演示生产务必沙箱return str(result)except Exception:return “计算出错请检查表达式”else:return “这个工具我还没学会”你可能会问“怎么把工具执行结果再送回模型”这里有一个我踩了无数坑的细节一定要在第二轮请求里把之前的消息和工具结果都拼进去组成完整的对话历史不然模型会失忆以为你在自言自语。端点的写法大致是这样router.post(“/agent”)async def agent_endpoint(req: ChatRequest, config: Settings Depends(get_settings)):# 第一轮发给 LLM带上 toolsfirst_resp await call_llm(req.message, toolsTOOLS, configconfig)logger.debug(f第一轮回复\n{first_resp })# 如果没有 tool_calls直接返回内容 tool_call first_resp.get(tool_calls, [None])[0] if not tool_call: return ChatResponse(replyfirst_resp[content]) # ---- 第二轮执行工具并拼接消息 ---- # 1. 把第一轮的 assistant 消息加入历史 messages [ {role: system, content: 你是一个能调用工具的助手必要时使用工具否则直接回答。当有工具返回结果时请直接基于结果回答用户不要再次调用工具也不要忽略结果}, {role: user, content: req.message}, first_resp # assistant 消息包含 tool_calls ] # 2. 执行每个工具调用并将结果作为 tool 消息追加 for tc in first_resp[tool_calls]: tool_name tc[function][name] # arguments 返回的可能是字串也可能已经解析为对象了作个判断 raw_args tc[function][arguments] if isinstance(raw_args, str): arguments json.loads(raw_args) else: arguments raw_args # 已经是 dict 了直接用 result await execute_tool(tool_name, arguments) # 你的工具执行器 messages.append({ role: tool, tool_call_id: tc[id], # 必须和上面对齐 content: result # 工具返回的字符串 }) # 3. 把完整历史再发给模型让它总结成人话不用带 tools 了 final_resp await call_llm(messagesmessages, configconfig) logger.debug(f第二轮回复\n{first_resp }) return ChatResponse(replyfinal_resp[content])再说个容易翻车的点工具执行那一步一定要加超时控制和 try-except。之前有次接的天气 API 抽风整个 /agent 接口跟着一起卡死前端直接白屏。后来强制设了 httpx 的超时并且加了一条黄金规则——工具调用失败时绝不抛异常而是把错误信息当作工具结果返回给模型让它自己圆场。这样用户至少能收到一句“天气服务暂时不可用”而不是一个冷冰冰的 500。大模型访问 call_llm() 也要做出重构核心思路 用 httpx.AsyncClient 调 Ollama 的 /api/chat 端点注意是 /api/chat支持 tools 参数不是之前的 /api/generate。在请求体里带上 messages 和 tools再处理返回的 message 字段。async def call_llm(user_message: str None, # 方便快捷调用messages: list[dict] None, # 传完整历史tools: list[dict] None,config: Settings None,system_prompt: str “你是一个能调用工具的助手必要时使用工具否则直接回答。”):if messages is None:messages [{“role”: “system”, “content”: system_prompt}]if user_message:messages.append({“role”: “user”, “content”: user_message})# 也可以选择动态添加 system prompt但要避免重复这里简化处理logger.debug(f对话消息\n{ messages })payload { model: config.model_name, messages: messages, stream: False } if tools: payload[tools] tools async with httpx.AsyncClient() as client: resp await client.post( f{config.ollama_base_url}/api/chat, jsonpayload, timeout30.0 ) resp.raise_for_status() data resp.json() return data[message] # 包含 content 和可能的 tool_calls⚠️ 第四部分进阶与避坑指南⭕ 工具定义尽量用 Pydantic 的 schema() 生成别手写 JSON那种参数名拼错的疼我挨过太多次。⭕ 执行器里 eval 只用来演示生产环境请用安全的表达式解析库或者直接写死允许的操作。⭕ 如果 LLM 不支持原生 tool_calls也可以自己在 system prompt 里要求它返回特定格式的 JSON然后手动解析效果一样稳。⭕ 后续可以把工具执行器改成插件式注册用装饰器收集加新工具就跟往工具箱里丢一把扳手一样简单。