ARTICLE · INTELLIGENCE

战地情报 · 详情页

来自尧图项目组的一线实战观察与深度解析

从Agent七要素到工程决策:FastAPI+LangGraph落地实践

从Agent七要素到工程决策:FastAPI+LangGraph落地实践 聊 AI Agent很多人第一反应是“一个能自己调用工具的聊天机器人”。这个类比没毛病但一到工程落地你会被一连串问题锤懵模型选多大合适工具调用失败了怎么补救上下文一长 Token 就爆用户一多并发就乱这些都不是背几个概念能解决的。所以我今天换个拆法不按框架讲原理而是先把一个 Agent 系统拆成七个必要组成部分再从这七个要素里拎出七个真正需要拍板的工程决策点最后落到一套基于 FastAPI LangGraph 的最小实现上。这篇文章适合两类人看一类是准备把 Agent 从 Demo 推向生产的工程师另一类是正在做技术选型、但被 LangChain、Spring AI、Rust 这些生态词绕晕的技术负责人。我尽量用做项目的方式把问题讲透你可以当它是自己踩坑两年后的复盘而不是一层层往上叠概念的教程。1. 先搭骨架Agent 七要素到底指什么1.1 用“人”来类比七件事缺一不可一个 Agent 能独立完成某个任务本质上和一个人处理工作没什么区别。你想清楚一个人要做成一件事需要什么Agent 的骨架也就出来了。目标这次任务到底要让 Agent 完成什么边界在哪里。模型负责推理和决策的“大脑”也就是 LLM。上下文与提示模型看到的背景信息、规则、约束统称 Prompt。工具Agent 能访问的外部能力比如查天气、发消息、调数据库。记忆暂存用户说过什么、Agent 做过什么以及需要长期记住的知识。规划把大目标拆成小步骤决定先做什么后做什么。执行与反馈真正去调用工具把结果送回模型形成一个闭环。我见过不少团队说自己在做 Agent结果拆开一看只有模型、提示词和工具三项。这种充其量是“带工具调用的对话机器人”因为当任务稍微复杂一点没有记忆、没有规划、没有反馈闭环系统就会走一步忘一步遇到错误也无法自我修正。以做菜来打比方目标是“四十分钟做出三菜一汤”模型是厨子提示词是厨房规矩工具是锅碗瓢盆和灶台记忆是“盐刚才放过了”的临时记录规划是先煮汤还是先炒菜执行和反馈则是试吃后调整火候。缺了任何一环这顿饭大概率没法顺利做完。这就是七要素的工程意义它不是概念清单是运行闭环。1.2 为什么不是六个也不是八个你可以说“安全”算不算第八要素或者“数据”算不算。但在工程实现上我更愿意把安全当成一个横切约束而不是独立组件数据则分散在记忆、工具、上下文里。七要素的边界不是一个绝对真理而是一个方便拆解和排障的心智模型。更重要的是少一个要素系统会立刻退化到某种已知形态没有目标和规划Agent 变成“你问一句它答一句”的 Chatbot没有工具它只是个会写作文、但无法改变世界的大模型没有记忆每次对话都像失忆用户必须重复上下文没有反馈闭环工具报错它就停在原地甚至把错误当成正确答案。这样看七要素更像是一张“功能完整性检查单”。你拿到任何 Agent 项目逐项去问“这一项谁负责”问不出来工程上迟早要出事。1.3 七要素不是 LangGraph 概念是团队责任分工很多新手看 LangChain/LangGraph 里一堆概念会误以为学会了框架就等于学会了 Agent。其实框架只是把七个要素用代码包装了一下真正的难点在于每个要素背后都要有人做决策。比如“目标”不是简单写一句 system prompt而是要定义输入 schema、校验方式、任务边界“记忆”不是 Redis 里存 chat history而是要考虑多久清空、要不要做摘要、向量库怎么更新。所以我习惯把七要素当成一份责任清单开需求会时逐个讨论目标谁来解析用户请求解析不了怎么告知用户模型用哪个模型准确率和成本怎么平衡上下文Prompt 有多长哪些信息永远不能被覆盖工具工具注册表怎么维护权限边界在哪记忆哪些记忆进短期窗口哪些进长期存储规划单步决策还是多步拆解终止条件是什么反馈工具失败后是重试、反思还是找人工把这份清单过一遍你就会发现后续所有决策点其实都是在这七个框架上长出来的。2. 七要素对应七个决策点工程实现真正的分岔路2.1 决策点一模型选型不是追最强而是追“够用且可控”七要素里最容易被卡住的是“模型”。很多人一上来直接上最强的大模型结果成本爆炸、延迟感人也有团队为了省钱用小的开源模型结果工具调用格式每天都出错。我的建议是先想清楚你的任务里模型到底承担多少推理负担。我通常按四个维度打分工具调用准确性、上下文长度、推理速度、综合成本。如果任务只是“从邮件里抽取结构化信息然后写进数据库”那么中等模型配合强 schema 约束就够如果 Agent 需要长时间自主规划、频繁调用工具、中途修正策略那就得让更强模型上场。工程上还要考虑“API 稳定性”和“私有化要求”。能公网调用 OpenAPI 自然省事但有些项目必须部署开源模型到私服。这时候别只看推理分数还要看它是否稳定支持 Function Calling因为弱模型的工具调用会让后面所有环节都难受。最近总有人问我“用 Rust 写 Agent 是不是性能更强”。我的回复是Rust 生态确实有 Agent 框架也适合追求极致吞吐的组件但大模型的推理接口才是瓶颈语言带来的性能提升在大多数业务场景根本感知不到。除非你的团队全是 Rust 工程师否则没必要为了技术热度换赛道。Python 生态在 LangGraph、LangChain、FastAPI 上的成熟度短期内仍是做 Agent 工程最省路的选择。2.2 决策点二目标怎么“写”给模型系统提示词不只是角色扮演第二个决策点是“目标”这个要素的落地方式。你以为写个 system prompt 就够了结果模型经常跑偏。原因很简单你没有定义清楚边界。我在实际项目里会把 system prompt 拆成四块角色与能力范围你是谁能做什么不能做什么任务拆解规则拿到用户请求后先做什么后做什么工具使用边界哪些场景必须调用工具哪些场景禁止调用输出格式约束给模型的回答套上固定框架。举个例子如果有人想用 Agent 管理小红书自动发消息我不会让它“自动发”我会在目标定义里写清楚“Agent 只能生成草稿并提交给用户人工确认未经二次确认不得执行发布动作。” 这不是限制是在保护系统因为一旦模型失控对外发出错误消息的后果远比丢掉一点自动化效率严重。同样的逻辑适用于任何高风险场景。有人问“个人用 Agent 做期货交易可以吗”我的回答是代码层面当然可以接行情和交易接口但真正难的是把“目标”转化成严格的约束什么条件下允许下单、最大仓位是多少、亏损多少必须停手、是否必须人工确认。这些决策如果全交给模型那就是在赌博。你也可以用人工审批节点来做硬约束后面我会讲到 LangGraph 的 interrupt 机制。2.3 决策点三工具调用Function Calling 和 MCP 怎么选七要素里的“工具”落到工程里第一个问题是模型怎么知道要调用哪个工具、参数怎么填。现在主流方案有三种原生 Function Calling模型厂商在 API 层支持结构化的工具调用模型会输出一个 JSON 格式的调用声明。这是最稳的方案因为它是模型对齐后直接生成的而不是让模型把 JSON 写进对话文本里。自绘 JSON 解析有些模型不支持原生 Function Calling你得规定它输出“我想调用工具 xxx参数是 yyy”然后再从文本里解析。这种方式很脆模型一旦换个措辞解析就崩。MCPModel Context Protocol它不是替代 Function Calling而是把工具暴露方式标准化。以往每接一个外部系统就要写一套自定义工具接入MCP 提供统一的工具发现和调用协议。如果你的业务里外部工具很多、更新很频繁我会优先考虑 MCP如果工具就两三个直接用 Function Calling 够省心。真正让工具调用稳定核心不在选协议而在工具描述和参数设计。我给过不少人一个建议每个工具的描述要写清楚“什么时候用、什么时候不用”参数能少就少枚举值尽量给死。比如一个“发送消息”工具不要给一个自由文本的 target 字段而是给枚举类型可选值为“微信、短信、邮件”模型就不容易填错。还要注意工具数量。一次性把 50 个工具堆给大模型它的选择准确率会明显下降。工程上可以分组先给 Agent 一个“工具选择器”让它决定调用哪一组工具再进入具体的组。这其实就是在用“规划”要素来降低“工具”要素的复杂度。2.4 决策点四记忆和 Token 预算先弄懂 Token 是什么“AI Agent token 是什么意思”这个问题几乎每个新手都会问。Token 可以理解为模型处理文本的最小单位一个中文词语可能对应一到两个 token英文则是若干字母凑一个 token。大模型 API 按 Token 计费模型也有上下文窗口也就是一次能处理的最大 Token 数。记忆设计本质上就是 Token 预算管理。你不可能把用户所有历史都塞进窗口所以要把记忆分成三类记忆类型存储位置典型实现解决什么问题短期上下文模型窗口内最近的 N 轮消息保持当前对话连贯长期记忆外部存储Redis / 数据库记住用户偏好和事实语义记忆向量库每次检索 Top K 相关记忆从海量历史里捞关键信息工程决策上我最关注的是“窗口资源分配”。如果模型窗口是 128K你不能把大半都用来塞历史。一般我会留出足够空间给系统提示词、工具描述和工具返回结果因为这些是当前决策真正依赖的信息。历史对话可以滚动窗口只留最近几轮更早的内容做摘要或者交给向量库等到需要时再检索回来。长期记忆还要考虑写入时机。不要每次对话都把所有事实写入长期记忆那样会产生大量脏数据。我常用的策略是先让模型判断“这句话是否值得记”值得才写入。这就相当于给记忆加了一个前置过滤器虽然多消耗了一次模型调用但长期收益明显。2.5 决策点五规划策略ReAct、Plan-and-Execute 还是图状态机“规划”是 Agent 和大模型聊天最不一样的要素。聊天模型一次回复就结束Agent 则要决定下一步干什么。工程上最常见的三种策略ReAct模型在“思考 - 行动 - 观察”之间循环每一步先想一下然后调用工具拿到结果再继续想。它胜在灵活适合探索性任务败在费 Token且容易绕圈。Plan-and-Execute模型先把整个任务拆成若干步骤再一步步执行。它省去重复思考的开销适合步骤相对固定的任务比如“先查库存 - 再算价格 - 最后下单”。缺点是一旦中途实际情况偏离计划它可能执迷不悟。图状态机用 LangGraph 这类框架把流程画成有向图节点是 Agent 或工具边是条件跳转。它的优势是工程可控能精确表达“什么时候必须人工确认”“什么时候重试”“什么时候终止”特别适合生产系统。怎么选我的经验是先把手头任务的执行路径、异常分支想清楚。如果 80% 的情况路径是固定的就用图状态机把主干和异常分支显式定义出来剩下 20% 的开放情况让图中的某个 Agent 节点用 ReAct 模式处理。换句话说图状态机管宏观流程ReAct 管微观决策。像扣子这类低代码平台本质就是把图状态机可视化、配置化适合快速验证流程。但它真正限制人的不是流程画布而是异常处理、权限控制、观测体系这些工程细节。所以我的习惯是先用扣子搭个 Demo 给业务看真正上线再回到代码里用 LangGraph 重写一遍。2.6 决策点六失败恢复重试、反思还是人工兜底工程实现和论文 Demo 最大的区别在于对失败的态度。Agent 的每一步都可能失败工具超时、返回格式不对、模型拒绝回答、上下文被垃圾信息污染。你必须在设计阶段就想好兜底策略。我的兜底顺序通常是有限次重试同一动作最多重试两次每次带上错误信息。让 Agent 反思把前一次失败的过程和结果作为 observation 塞回去要求模型解释失败原因并修正方案。切换路径如果反思两次仍失败就不再让 Agent 自动绕了切换到备用流程或人工处理。必有的人工审批节点凡是涉及对外发布、资金操作、删除数据的动作必须拦截住交给人工点确认才有下一步。使用 LangGraph 时这个“人工兜底”非常值得用它的 interrupt 机制来做。Agent 执行到某个节点发现需要人工确认就把执行暂停挂起状态等人点“通过”或“拒绝”再继续。这比让 Agent 自己决策“要不要发”安全得多。我见过太多项目翻车不是因为模型不够聪明而是没有设置最大迭代次数。模型会在一件事失败后反复用不同方式尝试看似在反思实则是在烧 Token 和调用次数。后来我在所有 Agent 流程里都加了一个硬性迭代上限比如“最多调用工具 15 次”达到上限直接终止并告诉用户“任务过于复杂请联系人工”。2.7 决策点七并发、可观测性与上线节奏最后这个决策点也是最近“AI Agent 怎么扛并发”这个问题问得越来越多的原因。很多人以为 Agent 就是一个 API 接口用 FastAPI 包一层然后加几个线程就能扛住用户量。但实际上 Agent 天然是多步、有状态、长耗时的工作负载它和普通 HTTP 请求处理模型完全不一样。你说接口收到一个请求模型要跑几轮、工具要调几次、外部系统要响应多久完全不可控。如果每个请求占用一个 Worker 进程把进程卡住等模型返回那 QPS 稍微一高整个服务就瘫了。我的建议是把 Agent 跑成异步任务队列而不是同步请求处理。FastAPI 只负责接收请求把任务 ID 和数据丢进队列比如 Redis Arq然后立刻返回“任务已受理”。后面的 Agent 流程由独立 Worker 异步执行执行过程中的状态全部写到 Redis 或数据库里。前端或者调用方轮询任务状态接口拿到结果。并发场景下还有一个隐蔽的坑状态串了。如果你把整个 Agent 的对话状态存在一个全局变量或者内存对象里两个用户同时触发任务后一个请求就会覆盖前一个请求的数据。解决方法是所有状态都放到按任务 ID 隔离的存储里绝对不要依赖进程内单例。可观测性则是另一个容易忽视的问题。普通接口可以看 HTTP 状态码Agent 流程怎么监控我之前踩过的坑是不知道问题出在模型调用上还是工具调用上。后来我把每一步的输入输出、Token 消耗、耗时都通过结构化日志打印出来再接入 LangSmith 或 OpenTelemetry 做追踪。只有这样当用户说“Agent 没做对”你才能定位是模型犯傻、工具报错还是流程跳错边。上线节奏上我强烈建议不要一上来就开放所有工具。可以先灰度一部分低风险工具比如“查天气”“搜索文档”让线上流量和观测数据帮你暴露问题等流程稳定了再逐步开放“发消息”“写文件”这类有副作用的能力。3. 实操一套最小实现FastAPI LangGraph 的 Agent 骨架3.1 技术组合为什么选 FastAPI 和 LangGraph前面的七个决策点讲得再多落不了地也是空的。这里我给出自己常用的最小可运行方案FastAPI LangGraph Redis。FastAPI 负责对外 API简单直接原生支持异步很适合做任务接收层。LangGraph 负责把 Agent 的规划、工具调用、人工确认、失败恢复编排成一张图而不是隐藏在无限循环里。Redis 负责状态暂存和队列。这套组合不是我拍脑袋选的它有状态隔离、有断点恢复、有可观测接口后期扩展成分布式 Worker 也方便。如果你所在的企业技术栈是 Java也可以关注 Spring AI 为此类 Agent 工程提供的集成能力本质上同样是构建任务编排和工具调用选择什么语言取决于团队但设计思路是通用的。我不推荐一上来就用 Rust 重写除非你已经确认 Python 是性能瓶颈。3.2 定义状态、节点和条件路由在 LangGraph 里第一步是定义状态。我通常用一个 dict 结构from typing import TypedDict, List, Optional class AgentState(TypedDict): task_id: str user_input: str messages: List[dict] # 模型对话历史 tool_outputs: List[dict] # 工具执行结果 current_step: str # 当前节点名称 need_human: bool # 是否需要人工审批 finished: bool这个状态就是整个 Agent 的“工作台”。所有节点读它、改它LangGraph 会负责把状态传给下一个节点。然后定义节点。最简版本至少要有两个节点一个 Agent 节点负责调用大模型做决策一个 Tool 节点负责执行工具from langgraph.graph import StateGraph, END def agent_node(state: AgentState) - AgentState: # 根据当前状态构造 messages调用大模型获取下一步动作 response call_llm(state[messages], available_tools) # 把模型回复追加到 messages state[messages].append({role: assistant, content: response.content}) state[need_human] response.need_human_confirmation # 自定义逻辑 return state def tool_node(state: AgentState) - AgentState: # 遍历模型要求调用的工具逐个执行 for call in state[tool_calls]: result execute_tool(call[name], call[args]) state[tool_outputs].append({name: call[name], result: result}) return state这里我没有写 call_llm 和 execute_tool 的具体实现因为它们跟你的业务强相关。真正关键的是条件路由也就是决定流程下一步走向def router(state: AgentState): if state[need_human]: return human_approval # 停到人工审批节点 if state[tool_calls]: return tool # 有工具要调用走 Tool 节点 return END # 没有更多动作结束 graph StateGraph(AgentState) graph.add_node(agent, agent_node) graph.add_node(tool, tool_node) graph.add_node(human_approval, human_approval_node) graph.add_edge(agent, tool) graph.add_edge(tool, agent) graph.add_conditional_edges(agent, router, {tool: tool, human_approval: human_approval, END: END}) graph.set_entry_point(agent) app graph.compile()这个图就是一个最简 Agent 闭环Agent 每次决策如果要工具就去执行工具拿到结果再回到 Agent 继续决策如果遇到人工审批或完成就退出循环。你可以在“agent”和“tool”之间循环也可以按业务需要插入更多中间节点。3.3 让工具调用稳定的小技巧实现工具节点时我一般会强迫自己遵守几个规则否则线上迟早出事。规则一永远校验模型回传的参数。模型再强也会偶尔输出缺失字段或错误类型。要用 Pydantic Schema 或者其他 JSON Schema 校验器把你的工具入参卡死。校验不通过不是直接执行而是当成“工具调用失败”反馈给模型让模型重新生成。规则二给每个工具套超时。外部接口可能 10 秒才返回Agent 系统等不起。我用一个统一的函数执行字典每个工具注册时声明自己的超时时间。超时后返回“工具调用超时”让模型决定重试还是换方案。规则三工具错误要进入反馈不是中断。很多人写代码时工具一把 try/except 直接吞掉异常Agent 看到的就是“工具执行成功”这非常危险。正确的做法是把异常信息封装成一个 observation 返回给模型比如“调用数据库失败连接超时”这样模型才有可能在下一步修正决策。def execute_tool(name: str, args: dict): tool TOOL_REGISTRY.get(name) if not tool: return {ok: False, error: fTool {name} not found} try: result tool.fn(**args) return {ok: True, result: result} except Exception as exc: return {ok: False, error: str(exc)}这条代码虽然朴实但它保证了反馈闭环不中断。你后续所有的自我纠错、重试机制都依赖这个“失败结果正确返回”的通道。3.4 并发控制把 Agent 跑成有状态工作流最小实现能跑通之后你迟早要面临并发。我现在的标准做法是 FastAPI 只做任务 API真正执行 Agent 的 Worker 另起进程或者部署成独立服务。# FastAPI 入口示例简化 from fastapi import FastAPI, BackgroundTasks import uuid app FastAPI() app.post(/agent/tasks) async def create_task(payload: dict, background_tasks: BackgroundTasks): task_id str(uuid.uuid4()) # 把 task_id 和 payload 丢进 Redis 队列或者用 BackgroundTasks 直接调度 background_tasks.add_task(run_agent_async, task_id, payload) return {task_id: task_id, status: accepted} app.get(/agent/tasks/{task_id}) async def get_task(task_id: str): state redis_client.get(fagent:{task_id}) return {task_id: task_id, state: state}轻量场景下 FastAPI 自带的 BackgroundTasks 就够用了一旦任务量上来我会换成 Celery 或 Arq这样 Worker 可以横向扩展。这里要特别强调Agent 的执行过程是异步长任务所以 API 层不要同步等待 Agent 跑完再返回。客户端通过 task_id 轮询或者后面用 WebSocket / SSE 推送都是比“同步等待”更稳的设计。状态隔离我前面提过实际操作时我会把 LangGraph 的 StateGraph compile 之后配合一个 checkpointer 参数让每一步状态自动持久化到 Redis。这样即使 Worker 中途宕机任务也能从最近的快照恢复不至于整个流程推倒重来。4. 常见问题与排查实录4.1 Token 超限窗口不够还是记忆策略不对很多团队的 Agent 一跑久就报“maximum context length exceeded”。第一反应是换更大窗口的模型但更常见的病根是记忆策略太粗暴把用户一整天的会话全部塞进 messages甚至把一堆工具返回结果原样保留在上下文里。我的排查顺序先打印 Token 构成看哪部分占大头。如果工具结果占了大头就在工具节点里做结果裁剪只保留关键字段长文本交给摘要节点压缩。如果是历史对话占大头就改成滚动窗口加摘要记忆。如果系统提示词已经很长了再把不常变的工具描述移到侧存储需要时动态加载。最后才考虑升级模型窗口。4.2 工具调用返回非法 JSON模型还能抢救一下用不支持 Function Calling 的小模型时最容易出现模型把 JSON 写在 Markdown 代码块里、或者字段名被翻译成中文之类的“灵异事件”。遇到这种情况先不要急着换模型试试以下操作收紧 system prompt明确给出 JSON 示例并强调“不要输出任何解释只输出 JSON”在解析失败时把“你刚才的输出无法解析原因是……”作为反馈发回模型让它自行修正改用厂商提供的结构化输出能力比如在请求参数里指定 response_format实在不行再加一层小的校验模型专门负责把模型输出转换成合法 JSON。这个过程中最忌讳的是解析失败就直接抛异常结束任务不试任何补救。因为模型具有很强的自我纠错能力只要把错误信息告诉它大多数情况下它能自己修回来。4.3 任务卡死Agent 反复调用工具停不下来我之前排查过一个案例Agent 为了“预订会议室”反复查了五次日历每次都说“没有合适时间”但既不结束也不换方案。典型问题是没有设置终止条件。解决办法有三层设置最大工具调用次数达到上限强制终止在系统提示词里写明“如果连续两次相同工具返回相同失败结果必须停止并告知用户”主动触发终止在条件路由里加入“步骤数”判断第 N 步仍未完成就走人工兜底节点。这里要养成一个习惯所有 Agent 循环都必须有显式的退出条件而且这个退出条件不依赖模型自觉而是由代码硬性控制。4.4 并发下状态串了同一个用户的请求相互覆盖如果上线后你发现用户 A 的操作结果跑到了用户 B 的任务里十有八九是状态存到了进程级变量。排查重点把所有状态都移到 Redis 或数据库并且以 task_id 为 KeyLangGraph 的 checkpointer 也一定要按 task_id 隔离如果同一个用户同时发起多个任务还要再区分是“同一会话的并发”还是“不同会话的并发”。更稳妥的方法是在创建任务时生成全局唯一的 task_id而不是用 user_id 当 Key。不管是日志、状态、队列、回调全部带着 task_id排查问题时才能顺着链路走。4.5 效果评测单点跑通不等于能上线最后聊一个大家不爱提但必须做的事评测。我见过太多 Agent 项目开发时说“效果不错”一上线用户乱输入就崩。原因是测试只覆盖了三五条 happy path。我的做法是建一个评测集至少 50 个问题覆盖标准任务、模糊指令、非法输入、工具失败、边界情况。每轮改动后跑一遍评测集记录几个核心指标指标含义观测方式任务完成率多少请求按预期完成终结为 END 且无人工介入工具调用准确率模型选择的工具与参数是否合理对比正确答案分析日志平均 Token 消耗每次任务烧多少 TokenLangSmith / 日志统计平均时延任务从受理到完成耗时队列 状态时间戳别追求一次性做到 95% 完成率你先要保证每次改动后指标可对比否则后面模型一升级、Prompt 一修改你都不知道效果是变好了还是变差了。5. 落地一点体会这套七要素加七个决策点的拆法我自己在实际项目中反复用。最开始我只是拿它当排查问题的干任务后来发现它还能帮团队对焦谁是负责人、哪些点还没想清楚、哪些点不需要过多纠结。尤其是工具调用和人工确认这两个地方是最容易花冤枉钱、走回头路的。最后分享一个小技巧把 Agent 的 Prompt、工具描述和评测集全部纳入代码仓库像管代码一样管它们。每次大模型升级或者 Prompt 调优都顺手跑一遍评测集防止某些“看起来 OK 的改动”带来工具调用格式的隐性变化。工程上的 Agent 好不好用很多时候不靠灵光一现靠的就是这套笨功夫。
RELATED READING

延伸阅读

更多一线实战笔记与深度复盘,助您持续精进