AI Agent最小循环与可靠系统工程实践:从LangGraph到FastAPI服务部署
别被“AI Agent”这个名词唬住。最近这半年我几乎天天在跟“Agent”这个概念打交道市面上聊 Agent 的文章多到刷不完但大部分要么停在给大模型发个“你会做什么”的提示词要么直接上 LangGraph、CrewAI 这种框架跑个 Demo 出来就算完事。真正到了要上生产、要扛并发、要处理真实业务场景的时候很多人会发现问题根本不在“模型聪明不聪明”而在“你把它当玩具还是当零件”。这篇文章就把我自己的实践经验摊开讲。先让你理解 Agent 的“最小循环”到底是什么然后一步步从单次调用、工具接入、上下文管理走到多轮控制、结构化输出、可观测性最后用 FastAPI LangChain LangGraph 搭一个能“下地干活”的服务骨架。适合人群很明确有 Python 基础、能调大模型接口、但还没搞懂“如何让 Agent 真正可靠”的开发者。看完以后你应该能自己评估框架选型、理解并发瓶颈在哪也知道为什么“最小循环”比“选哪个模型”更重要。1. AI Agent 到底在干什么先拆出那个“最小循环”1.1 Agent 不是“调一次 API”而是“模型 工具 循环”三位一体很多人第一次接触 Agent 的时候容易把它理解成“起一个 system prompt然后扔给大模型让它继续输出”。这其实还停留在 Chatbot 的思路上。Agent 的本质区别在于它拥有“行动能力”——它能调用工具、看到工具返回的结果、并根据结果决定下一步做什么直到达到目标为止。这个模型在日常里很像你去餐厅点菜服务员大模型先问你想吃什么用户输入然后跑到后厨下单调用工具后厨出菜后把菜端到你面前工具结果返回你尝了一口觉得差点意思模型再次分析服务员再跑一趟加个辣再次调用工具。这一整天跑下来真正有价值的不是“说话”本身而是“行动 → 观察 → 决策”这个结构。我把这套结构叫作“最小循环”大模型产生一段结构化输出通常是一个 Action 描述系统解析这段输出并执行对应工具工具结果回到模型上下文里模型基于新状态继续输出。这个循环一旦转起来Agent 才算真的“活”了。1.2 手写一个最小循环不依赖任何 Agent 框架要搞懂 Agent我强烈建议你先自己用原生代码写一遍最小循环别一上来就套框架。框架帮你解决了 80% 的问题但那 80% 恰恰是你不理解的坑。一个最简单的结构长这样import json from openai import OpenAI client OpenAI() TOOLS [ { type: function, function: { name: calculator, description: 执行加减乘除四则运算, parameters: { type: object, properties: { expression: {type: string, description: 例如 12、3*4} }, required: [expression] } } } ] def run_calculator(expression: str) - str: # 简单场景这里用 eval 演示生产环境绝对不要用 eval return str(eval(expression)) def agent_loop(user_input: str, max_steps: int 5): messages [{role: user, content: user_input}] for step in range(max_steps): response client.chat.completions.create( modelgpt-4o-mini, messagesmessages, toolsTOOLS, tool_choiceauto ) message response.choices[0].message messages.append(message.model_dump()) if not message.tool_calls: return message.content for tool_call in message.tool_calls: if tool_call.function.name calculator: result run_calculator(tool_call.function.arguments) messages.append({ role: tool, tool_call_id: tool_call.id, content: result }) return 达到最大迭代次数仍然没有完成 print(agent_loop(请问 (23 5) * 2 等于多少))这段代码看着朴素但你仔细琢磨这个循环里的每一步模型输出可能带着 tool_calls我们把它原样放回 messages执行完工具之后又把 tool 角色的输出放回去下一轮模型就能看到工具结果继续判断。1.3 为什么这个“循环”比选什么模型更关键我把最小循环比作发动机的曲轴选模型就像选汽油标号。你加了 98 号的油发动机装错了车子照样抖。反过来只要循环结构够清晰、够稳健哪怕用个中等规模的模型也能把任务完成得七七八八。这个循环里最容易翻车的有三个点messages 历史不能丢一旦把工具调用记录和结果漏掉模型会“失忆”导致同一件事反复做。工具执行要健壮工具本身报错了也要把错误信息当成工具结果传回给模型让模型自己决定下一步。必须设置最大轮次否则模型可能在错误的路上无限狂奔既烧钱又拖垮服务。从最小循环出发后面所有“可靠系统”的概念本质上都是在这个循环上做加法给它加记忆、加校验、加并发控制、加人工审批。框架做的事情是这个你自己做也是这个。2. 工具协议与上下文管理从“会转”到“转得顺”2.1 工具的本质是 JSON Schema不是“写个函数”前三章我们先把最小循环跑通现在要给这个循环加一个工程化能力工具接入。很多初学者把工具理解成“写一个 Python 函数然后让模型调用”这句话对了一半。模型不会直接调用你的 Python 函数它只会“表达调用意图”真正执行的是你的业务代码。关键在于模型表达调用意图的时候你需要给它一份它看得懂的“菜单”这份菜单就是 JSON Schema。继续拿餐厅打比方模型不是走进后厨自己炒菜而是对着菜单勾选“我要一份宫保鸡丁少辣”后厨再根据订单去做。一个工具定义通常包含四部分name工具名模型用它来区分要调用谁description描述这个工具能做什么、什么时候用、注意什么parameters一个 JSON Schema描述参数结构required哪些参数必填description 非常关键我见过很多团队工具定义写得很潦草结果模型乱调用。你写 description 时一定要说清楚“什么时候该用这个工具”、“参数怎么填”、“有哪些坑”。比如一个查询天气的工具description 可以写“根据城市名获取当前天气仅支持中国主要城市输入格式为北京而不是beijing”。2.2 上下文管理窗口裁剪与记忆分层Agent 只要跑多轮上下文膨胀就是绕不开的问题。每次把工具的结果往 messages 里塞塞到一定量上下文窗口就满了。窗口满了模型连原始用户意图都可能想不起来更别提完成任务。最简单的策略是“滑动窗口裁剪”只保留最近的 N 条消息。复杂一点的会用“摘要记忆”每跑完一个阶段就把关键结论压缩成一段摘要替代原有的大段对话。再复杂一点就是向量检索记忆把重要事实存到向量库里需要时拉出来。这些方案选哪种取决于你的业务复杂度。我自己的经验是先做滑动窗口 关键上下文固定。所谓关键上下文固定就是把用户的核心指令、业务约束条件单独放在一个“永远不被裁剪”的区块里窗口挤压时只丢工具调用的细节。messages [] FIXED_CONTEXT [ {role: system, content: 你是订单处理助手最终目标是把订单状态推进到已完成}, {role: user, content: 本次会话的目标是处理编号 A1001 的退款申请约束单笔退款不得超过 2000 元}, ] recent_history rolling_window(conversation_messages, max_len20) messages FIXED_CONTEXT recent_history这样做的好处是模型永远不会丢掉目标顶多丢掉“某个中间步骤的原始输出”而这些原始输出往往可以通过重新跑工具拿回来。2.3 工具执行结果不一定要“全文塞回”很多工具的返回结果又长又杂比如搜索接口返回 80 条链接和大段网页摘要。这种时候你需要把工具结果做一层“提炼”再回填给模型。我常用的方案是加一个“reducer”层工具执行完拿到原始结果先经过摘要、截断、结构化字段抽取再把精简后的内容放进上下文。比如搜索工具我一般只保留 5 条结果每条只保留标题和 50 字摘要。模型不需要看 80 条它只需要从中做出下一步决策。这就是为什么说 Agent 工程的核心是“信息管理”不是提示词。你给模型的信息不是越多越好而是越“恰到好处”越好。上下文里塞满噪音模型的决策质量会肉眼可见地下降。3. 可靠系统的第一课控制、校验与可观测性3.1 给 Agent 装“手刹”最大轮次、终止条件与人工审批Agent 循环最让人头疼的问题是什么失控。模型会想出各种你没想到的方式去调用工具花掉大把 token最后给你一个错误答案。要让它可靠就必须在“模型自由发挥”和“业务规则约束”之间搭一道护栏。首先是最基本的“最大轮次”。我在最小循环里设置了 max_steps5到了生产环境这个值可能需要按照任务类型分别配置比如“单轮信息查询”最多跑 3 步“多步骤数据分析”最多跑 10 步。超过步数就直接抛出一个可控异常不要让用户在那干等。其次是终止条件。Agent 不应该只在“模型说没有 tool_calls”的时候结束你应该定义“什么状态叫完成”。比如一个订单处理 Agent只有状态机标记为“已完成”或“已退款”才算真正完成一个内容生成 Agent只有通过了文本长度校验、敏感词校验才算完成。再就是人工审批。凡是对外会产生真实影响的动作发消息、转账、删除数据、下单建议插入“确认点”。实现方式可以是让 Agent 跑到某个节点时暂停把待执行动作和影响范围发给人工等人工确认后再用“继续执行”消息喂回模型。3.2 结构化输出让模型说“机器听得懂的话”很多人用 Agent 失败不是模型不够聪明而是结果格式太乱。比如你让模型输出“待处理工单列表”模型可能给你一段自然语言也可能给你 Markdown 表格还可能干脆把 JSON 包在 json 代码块里。生产系统根本没法解析这种输出。解决办法就是“结构化输出优先”。OpenAI 和一些主流框架都支持 response_format 或结构化生成。只要你的最终结果能用 JSON 表达就强制模型输出 JSON并且用 JSON Schema 校验。我在生产环境里的做法是所有 Agent 最终结论必须走结构化工序定义 JSON Schema 时属性名用英文描述写清楚字段含义输出之后过一层 pydantic 校验校验失败则把错误信息反馈给模型重新生成最多重试 2 次重试仍然失败就降级为人工兜底绝不在系统里留下脏数据这一套流程看似多了一步消耗了一些 token但能省掉下游无数个解析 Bug。3.3 可观测性没有 trace 的 Agent 等于盲盒AI Agent 和传统接口最大的区别在于它的执行路径是动态的。传统接口你看到报错就知道哪行代码出了问题Agent 出错了你看着终端愣住了——它调了哪个工具为什么调这个工具是模型理解错了还是工具结果太烂这个时候没有日志、没有 trace你根本无法定位。从第一版 Agent 开始我就在每个节点埋了三层观测数据输入层记录当前轮次的 messages 摘要、模型配置、上下文 token 数决策层记录模型响应全文、tool_calls 内容、选择该工具的原因推测通过 prompt 里的要求模型附上一句话判断依据执行层记录工具名、入参、出参、耗时、是否异常每一条 Agent 会话都会生成一个 trace_id贯穿始终。前端发生问题时拿 trace_id 出来就能回放整个决策过程。这套做法建议每个人都学不论你用 LangGraph 还是自己手撸框架。没有 trace 的 Agent 应用出问题的时候你是真的只能在“重试”和“看天”之间二选一。4. 架构选型观察从 LangGraph 到 Spring AI、Rust4.1 主流 Agent 架构模式到底怎么选聊完了底层逻辑再回头看市面上的框架和架构就清晰很多。当前主流架构大概分四类单 Agent 直连一个模型实例 若干工具适合任务边界清晰、工具数量少的场景最小循环跑得飞快。路由编排型先由“路由模型”判断请求类型再分发给不同子 Agent每种子 Agent 负责一个领域适合客服、工单分类这类多领域混合场景。流水线型多个 Agent 按固定顺序执行前一个的输出进后一个的输入比如先分析、后生成、再质检。图状态机型用节点 边的图结构描述整个流程支持条件分支、并行执行、循环收敛LangGraph 就是这个思路的代表。你说哪种最好没有最好只有适合。你要是只想做一个“联网搜索总结”工具单 Agent 直连半小时就上线你要做一个支持退款、改地址、查物流的客服 Agent图状态机型会更合适你要做多语言、多个业务线的复杂流程编排路由 图结合也许才是答案。4.2 框架横向对比LangGraph、Spring AI、Rust 生态我们再看三个在当前社区里讨论度很高的方向。第一个是 LangGraph。今年我已经把好几个项目从纯手写循环迁到了 LangGraph 上。LangGraph 的优势在于“状态图 持久化检查点 人工中断”这几个能力恰好戳中可靠系统的痛点。你可以把工具调用、模型推理设计成图上的节点然后通过边来控制跳转、循环和并行。它的学习曲线有点陡但一旦理解了 StateGraph 和节点函数签名写起来非常顺手。第二个是 Spring AI。Java 生态的同学问我要不要上 Agent我的建议是如果团队已经重度使用 Spring BootSpring AI 确实能让你在不引入太多新语言的情况下做 Agent 应用因为它的设计理念是“移植 Spring 的经验到 AI 应用里”提供了类似 PromptTemplate、ModelClient 这些抽象。但 Java 生态在 Agent 编排这块的成熟度还是不如 Python 生态如果是纯创新项目、团队有 Python 能力我会优先选后者。第三个是 Rust 方向。用 Rust 写 Agent 的收益在于性能和内存安全适合高并发、低延迟的“基础设施级”Agent 服务。但现实是大模型生态的 SDK、框架、工具链大部分以 Python/JS 为主Rust 团队往往要自己造很多轮子。我的观点很直白Rust 适合做“Agent 的网关层、调度层、推理代理层”不值得用它来写业务逻辑型 Agent。4.3 那个所有人都在问的问题AI Agent 怎么扛并发Agent 扛并发这件事很多人一开始就搞错了方向。他们以为给 FastAPI 加个 async 就完事了实际上 Agent 请求的耗时是“模型推理耗时 工具调用耗时 多轮循环耗时”往往比普通接口慢一个数量级。一个 Agent 请求可能跑 5 秒普通接口 200 毫秒你计算并发能力的时候不能拿假设 200ms 的思维去算。我总结的有效手段有这么几个把 Agent 请求设计成“任务式”而非“长连接式”。客户端提交任务拿到 task_id服务端异步执行客户端通过轮询或 WebSocket 等结果。这样可以把长任务丢给后台队列HTTP 线程不被拖死。在 Agent 服务前面加限流层按照模型 RPM、工具服务 QPS 双维度限流。模型 RPM 是很硬的瓶颈超了你就直接被服务商限速甚至封号。把 Agent 服务拆成“无状态”的。会话状态放到 Redis、数据库等外部存储里同一用户的多个请求打到任意节点都能继续跑。对工具调用做超时控制和熔断。某个工具挂了不能整个 Agent 跟着挂一定要快速失败并把它当成一种工具结果反馈给模型自己决定。简而言之Agent 服务要和普通 Web 服务同样做“背压”。你不可能靠一个进程无限开线程来硬顶并发最终还是回到分布式系统的老路上来队列、状态外部化、限流、重试、幂等。5. 实战FastAPI LangChain LangGraph 搭一个可用服务5.1 项目结构设计HTTP 层、状态图、工具层三件套如果你已经决定要落地一个 Agent 服务我推荐一个我反复用的结构。它不是唯一解但非常清晰拆成三个层面HTTP 接入层FastAPI 管路由、鉴权、限流、参数校验它不关心 Agent 内部怎么跑。Agent 编排层用 LangGraph 定义状态图把节点、边、条件跳转写清楚。工具执行层所有外部能力封装成统一接口每个工具都接收一个结构化参数返回一个字符串结果。这样分层的好处是HTTP 层和工具层都能做单元测试中间编排层只管“流程”。哪一层出问题都好修不会出现一个巨大 controller 里既写路由又调模型又执行爬虫的事。5.2 核心代码带状态图的 LangGraph Agent下面给一个精简但可跑通的示例功能是“查天气 算费用”的小 Agent。模型先判断用户意图如果涉及天气就调 get_weather如果涉及费用就调 calculator然后给出最终答复。from typing import TypedDict from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.tools import tool class AgentState(TypedDict): messages: list final_answer: str tool def get_weather(city: str) - str: 获取指定城市的天气信息输入城市名如北京 # 这里替换成真正的天气服务 return f{city} 晴25℃ tool def calculator(expression: str) - str: 计算四则运算输入如 12 return str(eval(expression)) tools [get_weather, calculator] model ChatOpenAI(modelgpt-4o-mini, temperature0) model_with_tools model.bind_tools(tools) def agent_node(state: AgentState) - AgentState: result model_with_tools.invoke(state[messages]) return {messages: state[messages] [result]} def tool_node(state: AgentState) - AgentState: last_message state[messages][-1] tool_calls getattr(last_message, tool_calls, []) results [] for call in tool_calls: tool {get_weather: get_weather, calculator: calculator}[call[name]] results.append(tool.invoke(call[args])) return {messages: state[messages] results} def router(state: AgentState): last_message state[messages][-1] if getattr(last_message, tool_calls, None): return tools return done graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tools, tool_node) graph.set_entry_point(agent) graph.add_conditional_edges(agent, router, {tools: tools, done: END}) graph.add_edge(tools, agent) app graph.compile()这个代码里关键点在 router 函数。它根据模型输出里有没有 tool_calls 来决定下一步是继续调工具还是直接收尾。整个状态图的结构就两条路径agent → tools → agent或者 agent → END。真实业务里你可以在 tools 和 agent 之间再塞一个“人工确认”节点、一个“结果校验”节点。这都比在纯手写循环里加 if-else 要清晰得多。5.3 FastAPI 接入与部署要点FastAPI 部分其实很简单把编译好的 app 包进一个 AgentService然后在路由里调用from fastapi import FastAPI, BackgroundTasks from pydantic import BaseModel app FastAPI() class TaskRequest(BaseModel): session_id: str user_message: str class TaskResponse(BaseModel): task_id: str app.post(/agent/task, response_modelTaskResponse) async def create_task(req: TaskRequest, background_tasks: BackgroundTasks): task_id generate_id() background_tasks.add_task(run_agent_in_background, req.session_id, req.user_message, task_id) return TaskResponse(task_idtask_id)实际部署时我一般不会直接依赖 FastAPI 的 BackgroundTasks因为进程重启任务就丢了。我会优先接一个真正的任务队列比如 Redis Stream 或 Celery/RQ。Web 层只负责把请求写入队列Worker 进程消费队列跑 Agent结果写回 Redis客户端轮询获取。扛并发从来不是一个端口的事而是一个“入口 → 队列 → Worker 池 → 结果存储”的事。Agent 任务是天然适合消息队列的因为它的执行时间不确定而且可以容忍一定延时。5.4 部署后第一件事先设置“黄金用例集”服务上线后我强烈建议你立刻维护一套“黄金用例集”。每一条用例包含用户输入、期望工具调用序列、期望最终输出。比如输入“北京天气怎么样” → 期望调用 get_weather参数 city北京输入“帮我算 (23)*4” → 期望调用 calculator最终输出 20每修一个 Agent 问题每改一次 prompt每换一次模型版本都把这套用例集跑一遍。Agent 是概率系统你不能靠一次手动测试就放心发布。这个习惯帮我在生产环境里拦住了无数个“这次看起来没问题但下个用户就崩了”的回归。6. 常见问题与排查实录6.1 高频问题速查表我整理了这段时间最常见的一些问题以及对应的排查思路你可以直接当参考。现象可能原因排查思路Agent 重复调用同一个工具工具结果没有正确返回上下文或模型认为还没拿到答案检查 tool 角色消息是否被完整塞回 messages确认工具结果是否清晰给出了结论工具调用格式频繁解析失败模型输出 JSON 不合法或被 markdown 包裹使用结构化输出强制 JSON加一层 pydantic 校验失败后重试Agent 跑着跑着“失忆”上下文窗口被裁掉了关键信息引入固定上下文区把核心目标放进去对历史做摘要而不是直接截断某个工具超时整个 Agent 卡死工具缺少超时控制给每个工具调用加 asyncio.wait_for超时后返回“工具超时”作为结果让模型继续处理高并发时模型接口报 429模型 RPM 限流在 Agent 服务内部加模型调用限流器做指数退避重试最终答案格式千奇百怪没有强制最终输出走结构化增加 result_validator 节点让系统在最终输出前校验结构不通过就重新生成Agent 走了不该走的工具工具 description 不明确模型理解偏差优化工具描述限制工具的触发条件必要时在白名单里去掉不合适的工具上下文爆满导致费用飙升没有做 token 监控和裁剪在每个节点统计 token 用量设置上限后触顶降级或重启会话这些是概率最高的几个坑每一个我都真金白银地踩过。尤其是工具调用格式解析失败在最初一个月几乎天天遇到后来用强制结构化输出 校验重试一次就解决了。6.2 我踩过的最深一个坑把 Agent 当普通函数调用有段时间我图省事直接在主业务进程里同步调用 Agent心想“反正就那点并发”。结果上线当天就被打爆了——几个 Agent 请求把数据库连接池耗尽模型接口也触发了限流。后来我把 Agent 执行改成异步任务队列请求发出去立刻返回 task_id在后台稳步消费这才算真正把并发问题解决掉。这也让我明白一个道理Agent 代码本身怎么写往往不是系统瓶颈你把它放在什么“生命周期”里跑才是决定系统稳定性的关键。6.3 给你一套排查 Agent 问题的“三板斧”每次出问题我都按这个顺序排查。第一件事查 trace看模型在哪个节点做了哪个决策工具调用和结果是不是对得上。第二件事复现——拿同样的输入跑一次但如果问题依赖状态就要把整段会话重放一遍。第三件事做“最小化定位”把工具从 10 个减到 2 个把 prompt 缩到最短如果问题消失就一个个加回来。这三板斧几乎能解决我遇到的 90% 的 Agent 问题。Agent 比传统代码难调就在于状态不可控但只要你把它当系统而不是当“魔法”问题基本都是可解、可复现、可预防的。7. 最后想说的几点个人体会写了这么多年业务代码Agent 是第一次让我觉得“软件工程的基本原则”和“概率模型的不可预测性”需要同时握在手里。你既要用传统工程的手段去约束它又要接受它有时候会出现你完全无法解释的行为。我个人最推崇的做法是小而可靠优先。与其做一个“啥都会”的大 Agent不如拆成一堆“只会一件事”的子 Agent再用流程把它们串起来。每个子 Agent 的职责边界越清晰整个系统就越不容易崩。另外一个小技巧在开发阶段就把 trace 和日志打好这会让你调试时省掉大量时间。我见过太多团队Agent 跑起来狂喜上线一周后被线上问题按在地上摩擦回头连“上次这个请求到底调了哪个工具”都查不到。从最小循环到可靠系统这条路不算短但每一步都可以用工程手段走扎实。搞清楚状态、控制住循环、记录下决策、设计好降级Agent 就能从一个“惊艳的 Demo”变成一个“稳如老狗的线上服务”。