尧图网页设计 
1. 内容整体设计与思路拆解1.1 为什么是“七要素”先看懂 Agent 的骨架聊 AI Agent很多人上来就扎进 LangChain、LangGraph 的代码里折腾半天却说不清楚一个 Agent 到底由什么组成。我早期也是这样写了一堆工具函数、接了好几个大模型接口回头一看代码像一锅粥加一个需求就要动全局。后来我把 Agent 项目拆开审视发现无论多复杂的智能体骨架其实就七个部分我把它们称为“七要素”。这七要素我用一句话概括一个角色带着目标基于记忆和规划调用工具依据反馈不断迭代直到交出结果整个过程都在安全边界内运行。展开说是这样目标角色Agent 是什么身份、要完成什么任务、任务的成功标准是什么。没有这层设定后续所有 Prompt 和编排都会失去方向。规划分解把大目标拆成可执行的小步骤。这里涉及任务拆解的粒度、步骤之间的依赖关系以及是否需要动态调整计划。记忆状态Agent 在执行过程中需要记住什么。短期记忆当前对话上下文和长期记忆跨会话的历史经验、用户偏好属于完全不同的实现路径。工具技能Agent 能调用哪些外部能力比如搜索、计算、数据库查询、HTTP 请求。工具的输入输出 schema 设计直接决定模型能不能正确调用它。反馈驱动Agent 如何感知自身执行结果的好坏。这里的反馈既包括工具返回的结构化数据也包括环境状态变化甚至是用户的中途纠偏。安全边界什么能做、什么不能做哪些操作需要人工确认哪些资源不能触碰。这一层往往被初学者忽略但在生产环境中恰恰是最先爆雷的地方。运行循环以上六要素如何被组织成一个循环执行的流程是顺序执行、条件分支、还是支持动态回退的图结构。我在实际项目中用过一个比方七要素里的前两个目标角色、规划分解是 Agent 的“大脑皮层”负责想清楚做什么中间三个记忆状态、工具技能、反馈驱动是“四肢感官”负责动手干最后两个安全边界、运行循环是“自律神经”负责让整个系统平稳运转、不失控。1.2 为什么还要谈“七个决策点”理念到工程之间的距离光有七要素只是把 Agent 的解剖结构看清楚了。真到落地开发你会发现每个要素背后都藏着一个或多个需要你来拍板的技术决策这些决策直接决定项目能不能扛住真实业务场景。比如“记忆状态”这个要素看似简单但你要决定上下文窗口满了怎么办用向量数据库还是 Redis 存历史要不要做记忆压缩不同选择的成本、延迟、效果差异非常大。再比如“运行循环”你要决定用简单的 ReAct 循环、Plan-and-Execute 模式还是引入 LangGraph 这类图编排框架决策的复杂度直接上升一个量级。我把工程实现中最常见、最影响成败的七个决策点提炼了出来对应关系大致是这样七要素七决策点目标角色技术栈选型框架和语言的取舍运行循环流程控制循环逻辑和状态转移怎么设计记忆状态记忆实现长期记忆用什么存储短期记忆怎么管理反馈驱动状态管理执行进度和中间结果要不要持久化运行循环并发策略多个任务同时进来怎么扛住压力反馈驱动观测体系日志、追踪、错误恢复怎么落地安全边界安全审计权限隔离和敏感操作把关这七个决策点是我在做了多个 Agent 项目后总结出来的“高频决策清单”。每做一个新项目我都会先把这七个问题想清楚再动手写代码。这篇文章的后半部分会逐个讲解它们背后的原理和取舍。提示七要素是静态的结构拆解七个决策点是动态的工程选择。前者回答“Agent 是什么”后者回答“Agent 怎么做出来、怎么做好”。两者结合才算是真正把 Agent 从概念变成了可运行的系统。2. 系统架构与工具链选型解析2.1 四种主流架构模式哪种更适合你的场景在动手写代码之前架构选型是绕不开的第一步。我经历过从“裸调大模型”到“图编排框架”的完整演进过程这里直接说结论目前生产环境中真正能打的 Agent 架构主流有四类。单轮直调模式把用户请求和工具定义一次性丢给大模型让模型返回完整的工具调用序列。实现成本最低适合工具数量少、流程固定的内部工具。缺点是上下文容易越撑越大一旦超过模型窗口就得做截断而且模型一旦中途“走神”整个链路就断了。ReAct 循环模式核心是“思考-行动-观察”的循环。模型每一步先推理下一步该做什么调用工具拿到观察结果再继续推理。这个模式在工具调用场景下表现相当稳定也是目前 Agent 项目里最常见的默认架构。缺点是循环次数不好控制可能陷入重复调用同一个工具的“死胡同”。Plan-and-Execute 模式先让模型一次性生成完整计划再按计划逐步执行。优点是计划和执行解耦计划可以单独展示给用户确认可解释性很强。缺点是对计划质量要求很高一旦计划有问题执行阶段再想调整就比较笨拙。不过现在很多实现会引入“计划反思”在执行过程中动态修订后续步骤弥补了早期版本的僵硬。图编排模式以 LangGraph 为代表把 Agent 的流程显式建模为一张有向图节点是“大模型推理”或“工具调用”边是状态转移条件。这是现在我主力使用的架构因为它把 ReAct 的循环、Plan-and-Execute 的分支、甚至多 Agent 协作都能统一表达在同一张图里调试和扩展都方便。我个人的建议是如果你是第一次做 Agent 项目、或者业务场景是单轮工具调用直接选 ReAct 模式起步用 LangChain 的create_react_agent就能跑通闭环如果场景涉及多步骤编排、条件分支、人工审批或者你需要对执行过程做精细控制直接上 LangGraph。不要一上来就自己写状态机和循环等跑通了最小闭环再考虑自研编排层。2.2 我为什么选择 Rust FastAPI LangGraph 的组合做技术选型时很多人的习惯是“哪个火用哪个”但工程上我更看重团队的熟悉程度、生态成熟度和部署运维成本。这里分享一套我在生产环境里稳定运行了半年的技术组合以及背后的取舍逻辑。Rust 写工具层和并发网关Agent 系统里最耗费资源的是工具调用层尤其是涉及网络 IO、数据解析、并发聚合的场景。Rust 在内存安全、无 GC 暂停、高并发吞吐上有天然优势适合承担 Agent 的工具执行网关。我不建议用 Rust 去写全部业务逻辑因为 Rust 的开发效率相比 Python 还是有差距但把高频调用的工具层用 Rust 下沉收益非常明显。FastAPI 做接口层Python 生态在 AI 领域的统治地位短期内无法撼动LangChain、LangGraph 的核心逻辑以及各种模型的 SDK都是 Python 优先。FastAPI 的异步支持、自动生成 OpenAPI 文档、Pydantic 数据校验让接口层写起来非常顺手。我一般把 FastAPI 放在 Rust 网关之后作为业务编排层对外暴露的入口。LangGraph 管流程编排前面说了图编排是复杂 Agent 架构的最优解。LangGraph 的节点、边、状态、检查点机制已经把大部分流程控制问题抽象得很好了。而且它支持异步执行、支持状态持久化和 FastAPI 的 async 模型能很自然地融合。这套组合的部署形态我画过一张草图用户请求 → Nginx → Rust 网关限流、鉴权、转发→ FastAPI 业务层 → LangGraph 编排层 → 工具执行层Rust 沙箱进程池→ 外部系统和模型 API。每一层只做一件事职责边界非常清晰出了问题也能快速定位。当然这套组合不是唯一解。如果你的团队全是 Java 背景用 Spring AI 也能搭出不错的 Agent 服务如果你追求极致性能甚至可以用 Rust 写完整的 Agent 编排逻辑。我的核心观点是选型要匹配团队能力和业务阶段而不是追逐技术热点。3. 七个决策点的深度拆解与实操3.1 决策点一技术栈选型如何影响后续六个决策许多技术书籍把选型放在最后讲但我在实际项目里发现选型几乎是第一个要拍板的事因为它会连锁决定后续所有环节的实现方式。这就像一个房子打地基地基风格决定了上面能盖什么结构的楼。以我前面推荐的组合为例来做几个连锁推演选 Rust 做工具层意味着工具调用要以子进程或服务调用的方式暴露给 Python 业务层。Rust 子进程的好处是完全隔离Python 层的崩溃不会波及工具进程坏处是进程间的通信开销和序列化成本。我第一次做的时候走了弯路让 Python 直接调 Rust 编译的.so动态库虽然性能很好但一旦 Rust 侧 panic 整个 Python 进程也跟着出问题。后来改成独立的 Rust 服务 HTTP 调用虽然单次调用增加了 0.5ms 左右的延迟但系统稳定性直线上升。选 LangGraph 做编排意味着你的流程天然是“状态机”形态。LangGraph 的节点函数需要接收一个 State 对象并在执行完毕后返回 State 的新片段。这个模式在你后面做持久化、断点续跑时会省很多事因为 LangGraph 的检查点机制已经把状态快照做好了。选 FastAPI 做接口层意味着你的 Agent 天然是异步模型。这里的坑是LangGraph 的一些节点函数如果用了同步阻塞操作会把事件循环卡住。我后面的并发章节会展开讲这里先记住一个原则Async 边界要划在编排层工具层不要让 Python 碰阻塞 IO。3.2 决策点二流程控制是 ReAct 还是图编排流程控制是 Agent 项目里最容易“先简单后炸锅”的地方。一开始你只需要让模型调一两个工具用 ReAct 循环就绰绰有余但等业务复杂到需要多轮人工审批、条件分支、异常回退时ReAct 的 while 循环会变得根本无法维护。我的建议是无论项目初期多简单都用图编排的方式建模。哪怕是只有“意图识别 → 工具调用 → 生成回复”三个节点的图也比硬编码的循环更容易扩展。LangGraph 里定义节点核心就是写一个函数from langgraph.graph import StateGraph, START, END from typing import TypedDict, Annotated class AgentState(TypedDict): messages: list current_tool: str tool_result: str retry_count: int def call_model(state: AgentState) - AgentState: # 调用大模型决定下一步动作 response llm_with_tools.invoke(state[messages]) return {messages: state[messages] [response]} def execute_tool(state: AgentState) - AgentState: # 执行工具调用把结果写回状态 result call_tool(state[current_tool], state[messages][-1]) return {tool_result: result, retry_count: state[retry_count] 1} def should_continue(state: AgentState) - str: # 根据状态决定走哪条边 if tool_calls in state[messages][-1].additional_kwargs: return execute_tool return end graph StateGraph(AgentState) graph.add_node(call_model, call_model) graph.add_node(execute_tool, execute_tool) graph.add_edge(START, call_model) graph.add_conditional_edges(call_model, should_continue, { execute_tool: execute_tool, end: END, }) graph.add_edge(execute_tool, call_model)这段代码的核心逻辑是模型节点决定要不要调工具要调就走进工具节点工具执行完回到模型节点由模型判断是否继续。should_continue这个函数就是图里的“决策边”比起在 ReAct 循环里硬编码条件这种写法在增加分支时只需要加一条边不需要动循环主体。流量控制层面图编排还有个 ReAct 做不好的事情人工审批节点。很多动作比如发送邮件、执行删除操作不能由模型直接决定需要在图中设置一个human_review节点将流程挂起等用户确认后再继续。LangGraph 的interrupt_before参数正是为这个场景设计的graph.compile(checkpointercheckpointer, interrupt_before[human_review])这个能力在财务审批、内容发布等强管控场景里是刚需。3.3 决策点三记忆实现的长短期之分记忆是我认为 Agent 项目里“看起来简单、做起来最坑”的模块。很多初学者把记忆等同于“把历史消息塞进 Prompt 里”这在对话轮次少的时候确实够用但一旦用户和 Agent 的交互超过 20 轮或涉及多个会话之间的关联这种粗暴做法立刻崩溃。先捋清楚长期记忆和短期记忆在工程上的天然差异短期记忆指当前会话内的上下文。实现上一般用消息列表随着对话推进不断追加到接近模型上下文窗口限制时需要做截断或压缩。这里的关键决策点是“截哪些”最早的消息最长的工具返回还是用摘要替代长期记忆指跨会话的用户偏好、业务事实、历史结论。实现上通常需要外部存储常见方案是向量数据库存语义向量、按相似度召回或 KV 存储存结构化偏好精确读写。我踩过的坑是在长期记忆上直接用了向量检索结果用户偏好这种带“明确键值”的信息被向量化后召回率很不稳定。后来我把长期记忆拆成两层结构化事实层用 Redis Hash 按用户维度存储比如“用户的收货地址”“用户所在城市”这种信息不需要语义检索按 key 读就能拿到非结构化经验层用向量库比如“用户上周问过一个关于退货规则的问题”这类信息需要语义匹配。这样做的准确率比单用向量库高了不止一个档次。短期记忆的截断策略也需要提前想清楚。我在生产环境用的策略是“摘要 窗口”混合每一轮对话结束后用一个小模型比如 gpt-4o-mini 级别把前面所有内容压缩成一段摘要存在状态里真正发给主模型的上下文只保留最后 6 轮完整消息加上这一段摘要。这个方案在成本和效果之间平衡得很好。3.4 决策点四状态管理的持久化与恢复状态管理是 Agent 工程跟普通 Web 项目最大的不同点。普通接口是无状态的或者状态只在 HTTP 会话层面但 Agent 的执行过程是长链路、多步骤的中间任何一个环节出错都可能导致整个任务“白干”。所以把 Agent 的执行状态落盘是实现可靠恢复的前提。LangGraph 的检查点机制帮我省了很多事。它的原理是每个节点执行完后把当前状态快照存到持久化存储里。如果流程中断可以从最近的检查点恢复而不是从头开始。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(checks.sqlite) graph graph.compile(checkpointermemory) # 执行时传入 thread_idLangGraph 会把同一条执行链路的状态绑定到这个 ID 上 config {configurable: {thread_id: user-123-task-456}} result graph.invoke({messages: [user_input]}, configconfig)这里有一个容易被忽略的细节检查点存的不只是消息内容还包括每个节点已执行的状态、当前待执行的节点、以及用户自定义状态里存的临时数据。这意味着你不但能做崩溃恢复还能实现“中断-恢复-继续”的交互模式比如用户在 Agent 执行到中途时要求“暂停一下我改个参数”这在电商订单流程、多步审批场景中非常实用。状态管理上另一个值得说的是不要把大对象塞进状态里。我见过有人把一整个 PDF 解析结果塞进 State 里结果检查点文件瞬间膨胀到几百 MB光序列化就卡好久。正确做法是状态里只存文件路径或数据库 ID需要内容时再去读。3.5 决策点五并发策略与性能优化实战Agent 系统跟普通接口的并发模型有一个本质差异单个 Agent 任务的耗时可能是分钟级甚至小时级的而普通接口通常在毫秒级返回。这意味着你的并发控制不能简单的“加个线程池”而是要深入理解任务的性质。会话内并发和会话间并发是完全不同的两个维度会话内并发同一个 Agent 任务里多个工具调用能不能并行执行比如你要同时查天气、查航班、查酒店这三个调用互相独立串行执行会白白浪费几秒甚至几十秒。LangGraph 里可以设计成“扇出-扇入”结构一个节点同时发起多个工具调用等待全部返回后聚合。会话间并发系统同时有多少个 Agent 任务在跑这里要做的是“异步化 排队 限流”。FastAPI 层面把任务直接交给后台 task 去跑用 Redis 做消息队列用独立的 worker 进程消费任务避免请求线程被长时间占用。我在生产环境里做压测时发现一个性能杀手Python 的 GIL 在大模型调用返回后、解析 token 流时会导致事件循环卡顿。这个问题在同步调用大模型 SDK 时尤其明显。后来我把大模型 SDK 的调用全部换成了异步客户端并且规定 FastAPI 层和 LangGraph 层都禁止使用同步阻塞的第三方库。扛并发的另一个关键点是把“计算密集”和“IO 密集”分离。大模型调用天然是 IO 密集等网络返回计算密集的部分主要在工具层的数据处理后处理。让 Rust 工具网关承担计算密集的部分Python 编排层专注异步调度整个系统跑下来 CPU 占用率和 P95 延迟都有明显改善。注意给并发策略做压测时不要只关注 QPS更要关注 E2E 延迟分布从请求进入到任务完成的整体耗时。Agent 任务的 E2E 延迟会因工具调用次数、模型响应长度波动巨大压测报告里至少要区分 P50、P95、P99 三个档位否则很容易误判系统容量。3.6 决策点六可观测性与故障排查体系Agent 系统的可观测性比普通 Web 系统要难做得多。原因在于一个 Agent 任务的执行轨迹是由模型动态决定的不是代码里写死的链路。你没法提前知道模型会调用哪些工具、走哪条分支。所以日志和追踪必须能完整还原“模型每一步在想什么、做了什么、拿到了什么结果”。我从头到尾坚持三个层面的观测覆盖Trace 层面每次完整的 Agent 任务生成一个 trace ID贯穿 FastAPI 入口、LangGraph 节点、工具调用、外部 API 请求全链路。同时在 LangGraph 的每个节点函数里注入 trace ID 到日志上下文确保所有日志可以按任务维度聚合。状态快照层面利用 LangGraph 的检查点机制在关键节点执行前后把状态打印出来或存起来。调试时可以一帧一帧回放 Agent 的执行过程定位“模型在哪一步产生了错误决策”。Token 与成本层面大模型调用是花钱的一个 Agent 任务可能内部调了十几次模型成本必须可观测。我在每次模型调用后统计输入 token、输出 token、缓存命中情况写入监控看板。上线初期我在这个环节踩过一个大坑Agent 在一个循环里反复重试同一个工具单次任务消费了几十万 token账单直接飙到预期之外。故障排查上最重要的一个心法是“不要让模型真的一点一点试错”。线上环境出现过 Agent 连续调用同一个工具三次、拿到同样的错误结果、仍然不肯停止的现象。后来我在每个工具节点的入参里加了retry_count字段把这个数字注入到模型的上下文里让模型“知道”自己已经失败多少次了并要求它失败两次后必须换一种方案。这种“外部状态注入”的方法比单纯在 Prompt 里写“请谨慎调用工具”有用得多。3.7 决策点七安全审计与权限隔离的底线思维安全审计是很多 Agent 项目的“事后补丁”但我强烈建议把它作为架构层面的内建能力。Agent 跟普通 API 的最大区别在于模型的决策是概率性的它可能某一次调用出现“越权冲动”而且概率不可归零。所以安全边界不能依赖模型自律必须在代码层面做强制约束。我在实际项目中执行的安全策略分成四层工具权限白名单Agent 能调用的工具集合是配置化的不是从模型输出里直接反序列化出函数名就调用。解析模型返回的tool_calls后先校验工具名是否在白名单里不在就直接拒绝。这个校验甚至可以基于一个合法函数的注册表避免模型“凭空捏造”工具名。参数强制校验工具入参要用严格的 JSON Schema 校验尤其是数字范围、枚举值、路径穿越这类问题。我曾经遇到模型生成{delete_file: ../../etc/passwd}这种参数虽然工具内部有路径校验没造成实际问题但想想都后怕。参数校验层务必由代码完成不依赖模型。敏感操作二次授权发送真实短信、扣款、删除数据、公开内容等操作在设计工具 schema 时就要标注为“高风险操作”。流程图上高风险操作必须经过human_review节点由用户确认后才能执行。这个约束要写死在编译后的图结构里运行时不可绕过。审计日志留痕模型的每一次决策、每一次工具调用、每一次参数修正都要记录。不只是为了排查问题更是为了答案“可追溯”。用户问“为什么扣款成功但系统没提示”你要能翻出当时的全链路 trace精准回答是哪一步出了错。另外还要考虑多租户场景下的数据隔离。如果你的 Agent 服务同时给多个租户使用工具调用涉及的数据读取必须强制注入租户 ID并在数据库查询层做强制过滤防止一个租户的 Agent 工具被模型错误地传参后读到另一个租户的数据。这种隔离要在 DB 层和 ORM 层做死不能只靠业务代码自觉。4. 完整实操过程从零搭建一个带并发能力和观测体系的生产级 Agent4.1 项目初始化与目录结构设计这部分我以一个真实的简化项目为例手把手带你跑通整个流程。项目需求是做一个“客服工单自动分类 处理建议生成”的 Agent输入是用户工单文本输出是分类标签、处理建议、以及是否升级到人工的判断。第一步建目录结构。我习惯按“接口层 / 编排层 / 工具层 / 存储层”四个维度拆分agent_project/ ├── api/ # FastAPI 接口层 │ ├── main.py # 入口注册路由 │ └── routes/ │ └── ticket_agent.py # 工单处理路由 ├── orchestrator/ # LangGraph 编排层 │ ├── graph.py # 构建图 │ ├── nodes.py # 节点函数 │ └── state.py # 状态定义 ├── tools/ # 工具层 │ ├── registry.py # 工具注册与白名单 │ ├── ticket_tools.py # 工单相关工具分类、查历史、升级 │ └── rust_gateway/ # Rust 实现的工具网关 ├── storage/ # 存储层 │ ├── memory_store.py # 长期记忆存取 │ └── checkpoint_db.py # 检查点数据库 ├── config/ │ ├── settings.py # 全局配置 │ └── prompts.yaml # 各类 Prompt 模板 └── tests/ └── test_agent_flow.py这个目录的关键是把“工具层”单独拆出来。工具是 Agent 与外部世界的唯一触点工具层混乱会导致安全审计和权限控制都很难下手。4.2 用 LangGraph 定义带分支与人工审批的 Agent 流程我设计这个工单 Agent 为五节点图意图识别 → 信息补充 → 工具执行 → 决策输出 → 人工审批。其中“人工审批”节点只在系统判断“工单需要升级处理”时才进入。节点定义的核心代码如下from langgraph.graph import StateGraph, START, END from typing import Literal class TicketState(TypedDict): ticket_text: str category: str suggestion: str need_human: bool human_confirmed: bool messages: Annotated[list, operator.add] trace_id: str def classify_node(state: TicketState) - TicketState: # 用模型判断工单分类 prompt load_prompt(classify_prompt).format(ticketstate[ticket_text]) resp llm.invoke(prompt) return {category: resp[category], need_human: resp[need_human]} def tool_node(state: TicketState) - TicketState: # 根据分类调用对应的处理工具 if state[category] refund: result call_refund_tool(state[ticket_text]) elif state[category] technical: result call_tech_help_tool(state[ticket_text]) else: result {suggestion: 无法自动处理转人工} return {suggestion: result[suggestion], need_human: True} def route_after_classify(state: TicketState) - Literal[tool_node, decision_node]: # 某些分类不调用工具直接生成建议 return tool_node if state[category] in (refund, technical) else decision_node def decision_node(state: TicketState) - TicketState: # 根据工具结果决定是否要人工审批 return {need_human: True} if state.get(need_human) else {need_human: False} def human_review_node(state: TicketState) - TicketState: # 挂起等待用户确认确认后继续 return {human_confirmed: True} builder StateGraph(TicketState) builder.add_node(classify_node, classify_node) builder.add_node(tool_node, tool_node) builder.add_node(decision_node, decision_node) builder.add_node(human_review_node, human_review_node) builder.add_edge(START, classify_node) builder.add_conditional_edges(classify_node, route_after_classify) builder.add_edge(tool_node, decision_node) builder.add_conditional_edges(decision_node, lambda s: human_review_node if s[need_human] else END, {human_review_node: human_review_node, END: END}) builder.add_edge(human_review_node, END) graph builder.compile(checkpointerSqliteSaver.from_conn_string(state.db), interrupt_before[human_review_node])这段代码里的几个关键细节值得展开interrupt_before[human_review_node]是让图在进入人工审批节点前自动挂起此时graph.invoke返回值中会带上__interrupt__字段。前端拿这个字段提示用户“需要确认”用户点击确认后用新的输入再次invoke图会自动从挂起点继续执行而不是从头跑。route_after_classify返回的字面量类型和条件边的映射字典保证了路由的正确性。LangGraph 会根据返回值去找对应的目标节点映射字典里没写的目标会直接报错这比裸写 if-else 安全得多。State 里的trace_id在节点间流转这样所有日志可以通过它关联到同一次任务。4.3 工具层的封装与 Rust 网关对接工具层是 Agent 最容易“脏乱差”的地方。我的做法是所有工具都用统一的装饰器注册并在注册时声明工具名、参数 schema、风险等级、是否需要人工授权。from tools.registry import register_tool from pydantic import BaseModel class RefundCheckParams(BaseModel): ticket_id: str user_id: str amount: float Field(gt0, lt100000, description退款金额必须大于0且小于10万) register_tool( namerefund_check, risk_levelhigh, # 高风险操作需人工授权 requires_humanauto, # 自动进入 human_review 流程 ) def refund_check(params: RefundCheckParams) - dict: # 这里对接 Rust 网关通过 HTTP 调用实际业务系统 gateway_url settings.RUST_GATEWAY_URL resp httpx.post(f{gateway_url}/refund/validate, jsonparams.model_dump()) resp.raise_for_status() return resp.json()Rust 网关的职责是接管所有外部系统交互这样业务系统不需要暴露大量 API 给 Python 层。我用 Rust 的axum框架写了几个核心接口每个接口都做了严格的参数校验use axum::{extract::State, Json}; async fn refund_validate( State(pool): StateDbPool, Json(params): JsonRefundValidateReq, ) - ResultJsonRefundValidateResp, AppError { let exists check_ticket_exists(pool, params.ticket_id).await?; if !exists { return Err(AppError::NotFound(ticket not found.into())); } let risk compute_refund_risk(params.user_id, params.amount)?; Ok(Json(RefundValidateResp { risk_score: risk })) }对比 Python 实现同样的接口逻辑在 Rust 里能做到更高的单位吞吐而且 panic 会被 axum 捕获转成 500 响应不会拖垮整个进程。如果你的团队 Rust 经验不足用 Go 做这个网关层也是个不错的选择核心思想是一样的把有状态、耗资源的外部交互全部下沉到独立进程中尽量少让 Python 编排层直接面对网络风暴。4.4 并发控制从同步阻塞到异步编排的改造实录我在早期版本里遇到过一个经典事故FastAPI 接口收到 20 个并发请求每个请求都在 LangGraph 里调用同一个大模型 SDK结果 Python 的事件循环被同步阻塞卡住后续所有请求排队等待接口 P95 延迟从 800ms 直接飙升到 6 秒。排查后发现根源在于 LangGraph 的节点函数里用了同步版的大模型客户端。改造方案分两步第一步把节点函数全部改成async大模型调用改用异步客户端async def classify_node(state: TicketState) - TicketState: async with async_llm_client.acquire() as client: resp await client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], ) return {category: resp.choices[0].message.content}第二步LangGraph 的ainvoke本身就是异步入口FastAPI 路由直接 await 它app.post(/ticket-agent) async def handle_ticket(request: Request): body await request.json() config {configurable: {thread_id: body.get(trace_id)}} result await graph.ainvoke({ticket_text: body[text]}, configconfig) return {result: result}改造后压测效果明显在 50 并发下P95 延迟稳定在 1.2 秒左右CPU 占用也降了下来。并发控制里还有一个容易忽略的点避免多个线程或任务共享一个可变对象。比如把大模型客户端实例化在模块顶层那它就是被所有请求共享的。如果 SDK 内部有线程不安全的状态在高并发下会出现莫名的数据错误。解决方法是把客户端做成“每请求一个实例”或者使用连接池并确保 SDK 的线程安全声明。4.5 可观测性落地日志、追踪、成本监控一次讲透我在可观测性上的落地经验是不要等技术债积累到没法排查时才补一开始就要把三大件装上。日志方面我统一使用structlog结构化日志把 trace ID、节点名、工具名、耗时作为固定字段输出。举一个生产环境里的真实日志片段{ trace_id: conv-8f3a-20240512-001, node: tool_node, tool: refund_check, status: ok, duration_ms: 312, risk_score: 0.42, timestamp: 2024-05-12T10:23:45.123Z }这样的日志按 trace_id 聚合后可以复盘整个任务的完整时间线从入口到每一个工具的耗时分布、模型决策点、最终结果一目了然。追踪方面我在 FastAPI 中间件里注入 OpenTelemetry 的 span把 LangGraph 的执行轨迹透传到 Jaeger。LangGraph 支持自定义回调我写了一个简单的回调类在每个节点执行前后记录 spanclass TracingCallbackHandler: def on_node_start(self, node_name, state, *args, **kwargs): with tracer.start_as_current_span(fnode:{node_name}) as span: span.set_attribute(state.keys, list(state.keys())) span.set_attribute(trace_id, state.get(trace_id, )) def on_node_end(self, node_name, state, *args, **kwargs): with tracer.start_as_current_span(fnode_end:{node_name}) as span: span.set_attribute(state.output, json.dumps(state, ensure_asciiFalse, defaultstr))成本监控方面我在模型调用层包了一层统计装饰器每次调用后记录模型名、输入输出 token 数、价格、总成本定时汇总到 Prometheus。这里要特别关注 Agent 任务的“单任务成本”因为它跟普通 API 的单次调用成本完全不同。我在生产环境里给每个 Agent 任务设了成本阈值比如单任务模型费用上限 0.1 元超过阈值就强制中断避免模型在异常循环里产生额外费用。4.6 安全审计的代码实现权限隔离和二次授权怎么落地安全审计在前面的决策讲解里提到了设计层面这里给一套可运行的代码实现。我用一个中间件来控制所有进入 LangGraph 的工具调用确保任何工具在执行前都过了三道关卡from tools.registry import TOOL_REGISTRY from fastapi import HTTPException def safe_execute_tool(state: dict, tool_name: str, params: dict) - dict: # 关卡一白名单校验 registered TOOL_REGISTRY.get(tool_name) if not registered or not registered.enabled: raise HTTPException(status_code403, detailftool {tool_name} not allowed) # 关卡二Schema 参数校验 validated registered.schema_model.model_validate(params) # 关卡三风险等级处理 if registered.risk_level high: # 必须经过人工审批节点否则拒绝执行 if not state.get(human_confirmed, False): raise HTTPException(status_code400, detailhigh-risk operation needs human confirmation) # 记录审计日志 audit_logger.info(tool_executing, tool_nametool_name, paramsvalidated.model_dump(), trace_idstate.get(trace_id)) return registered.func(validated)这个safe_execute_tool是所有工具节点的统一入口。你在 LangGraph 的节点函数里调用工具时不要直接调业务函数而是走这个统一入口。这样安全策略的修改只需要改一处所有工具统一生效。关于人工审批我强烈建议把它做成“异步双向确认”Agent 挂起后系统通过消息队列或 WebSocket 通知前端前端展示工具名称和参数用户确认或否决后再通过另一个接口继续执行图。而不是简单地“等待同一个 HTTP 请求一直挂着”。原因很现实真实业务里人工确认可能就是几分钟HTTP 连接早超时了。用异步确认用户体验和系统稳定性都更好。5. 常见问题与排查技巧实录5.1 模型在工具调用时反复出错怎么缓解这是所有 Agent 项目最先遇到的坎。模型要么不按 schema 输出工具调用要么输出了但参数格式错误要么拿了工具结果却不继续往下走。排查思路按顺序来先确认 Prompt 里的工具描述是否足够具体。工具描述里要写清楚“这个工具是做什么的、输入什么、输出什么、在什么情况下调用”。我见过模型反复调用错误工具的原因居然是工具描述里的示例太模糊模型根本没理解两个相似工具的区别。再确认工具入参的 schema 设计是否简单直接。字段尽量用基本类型不要用深层嵌套的 JSON 对象。模型的 JSON 生成能力对嵌套结构很不友好。如果模型总是忘记带上某些必填参数可以把这些参数从“需要模型生成”改为“从外部状态注入”。比如用户 ID、trace ID 这类字段不要出现在模型的工具入参里而是节点执行时从 State 里取。容错层面我给工具节点加了一个“解析失败则重新生成”的循环机制最多重试三次把每次失败的原因拼接回 Prompt 让模型修正。这个策略在实际项目中能把工具调用的解析成功率从 80% 提升到 97% 左右。5.2 并发一高就报错或超时问题多半出在这三个地方第一看事件循环是否被阻塞。我之前已经讲过Python 异步环境里只要有一个同步阻塞调用整个进程都会卡住。检查方法很简单在 FastAPI 接口里加一个/health探针路由专门记录事件循环的响应延迟如果探针延迟异常起伏说明有阻塞。第二看外部依赖的客户端是否有连接池上限。大模型 SDK 默认的连接数可能只有 10并发一高就排队超时。解决方法是调大连接池、启用 HTTP/2 多路复用、或者加一层 Redis 缓存减少重复请求。第三看数据库连接池是否被打满。LangGraph 的检查点机制会在每个节点执行后写一次数据库如果你的任务有很多节点高频写入会很快打满连接池。解决方法是检查点库单独用一个连接池并设置合理的连接上限。另外Agent 任务本身是长耗时任务不建议直接在 HTTP 请求里同步等待完成。更好的做法是接口只返回task_id后台异步跑 Agent 任务执行完成后通过 WebSocket 或轮询结果接口告知前端。这样并发压力从“同时处理 N 个同步长连接”变成“同时处理 N 个异步任务”系统容量完全不同。5.3 长期记忆不准确优先检查这三点长期记忆召回不准很多时候不是向量库的问题而是数据组织方式的问题。第一检查存储的粒度。我上面说过用户偏好这类确定事实应该用结构化存储不要塞进向量库。向量召回适用于“模糊语义匹配”的场景如果你用向量库查精确事实召回不准是必然的。第二检查索引构建时有没有做切块。一篇很长的历史对话日志如果不切块直接向量化模型检索时只能命中整篇的近似向量精度会很差。正确做法是按语义段落或固定长度切块为每个块独立建索引。第三检查召回后的上下文拼接顺序。长期记忆召回的内容和当次会话内容的拼接顺序会影响模型的理解。我习惯把“用户基本信息”“历史偏好”“相关历史片段”按优先级放到系统提示词的固定位置而不是混在对话消息里。5.4 线上事故排查清单我每次发版前后必看一遍这份清单是我在踩了多次坑之后整理出来的每次发版前和出事故后都会逐条过一遍模型调用的超时时间是否设置合理。我见过大模型响应慢导致 Agent 节点一个接一个超时最后整条任务失败的场景。超时设置要比模型 P99 延迟高 20% 左右。工具调用的失败重试是否有限制。无限重试会让 Agent 陷入死循环增加成本也拖垮性能。所有重试必须设置最大次数。检查点存储是否有过期清理机制。如果不清理检查点数据会积压数据库越来越大恢复速度越来越慢。我设置了 7 天 TTL超过就自动删除。高风险操作的二次授权是否真的生效。测试时容易图省事直接绕过但线上必须确保。成本阈值是否配置到了每次模型调用。新模型上线时单次任务成本可能会因为模型更贵而翻倍成本阈值要跟着重新评估。6. 从“能做出来”到“能扛住”我的经验与建议6.1 先跑通最小闭环再追求架构演进我见过太多团队在第一个 Agent 项目上就追求“完美架构”结果三个月过去连一个能跑的闭环都没有。我的建议是第一版哪怕用最土的 ReAct 循环只要能跑通“用户提问 → 模型决策 → 工具调用 → 返回结果”的最小闭环就值得庆祝。最小闭环跑通之后再按七个决策点的优先级逐步演进。我自己的优先级排序大致是安全审计最先做这是底线其次是可观测性没有观测就没法优化再然后是并发策略业务量上来之前不用做最后是长期记忆和流程控制的高级特性。6.2 永远假设模型会犯低级错误这是我在 Agent 项目里学到的最重要的一课。模型不是确定性的程序它在 99% 的情况下表现正常但那 1% 的低级错误多了一个空格、把数字写反、凭空捏造工具名一旦发生在生产环境可能就是事故。所以所有关键环节都要用代码兜底不要相信模型的“自觉”。工具调用白名单、参数 schema 校验、结果二次校验、人工审批、成本阈值——这些机制不是为了限制模型的能力而是为了保护系统的确定性边界。在这个边界之内让模型自由发挥效果会好得多。6.3 给新手的三条实操建议第一多用 LangGraph 而不是自己写状态机这能让你更快理解 Agent 的节点和边模型。第二日志要结构化从一开始就养成带 trace ID 记录的习惯后面排查问题的收益是百倍级别的。第三先手工测试单条 Agent 任务的完整链路再上压测不要在没验证功能正确性时就开始调性能。我最后想分享一个项目里的体会Agent 工程跟传统后端开发最大的不同是你要跟一个“不可控的执行者”协作。传统的代码是你写什么就执行什么但 Agent 是模型在决定执行路径。所以Agent 工程的核心能力不是写代码本身而是设计边界和兜底机制的能力——让模型在边界内自由发挥在边界外永远别让它越线。这个能力只能靠实际做项目慢慢积累七要素帮你建立结构认知七个决策点帮你找到发力方向剩下的就是在真实项目里一遍遍调试、踩坑、优化的过程了。 AI Agent开发七要素与七大决策点：从架构设计到工程实践
1. 内容整体设计与思路拆解1.1 为什么是“七要素”先看懂 Agent 的骨架聊 AI Agent很多人上来就扎进 LangChain、LangGraph 的代码里折腾半天却说不清楚一个 Agent 到底由什么组成。我早期也是这样写了一堆工具函数、接了好几个大模型接口回头一看代码像一锅粥加一个需求就要动全局。后来我把 Agent 项目拆开审视发现无论多复杂的智能体骨架其实就七个部分我把它们称为“七要素”。这七要素我用一句话概括一个角色带着目标基于记忆和规划调用工具依据反馈不断迭代直到交出结果整个过程都在安全边界内运行。展开说是这样目标角色Agent 是什么身份、要完成什么任务、任务的成功标准是什么。没有这层设定后续所有 Prompt 和编排都会失去方向。规划分解把大目标拆成可执行的小步骤。这里涉及任务拆解的粒度、步骤之间的依赖关系以及是否需要动态调整计划。记忆状态Agent 在执行过程中需要记住什么。短期记忆当前对话上下文和长期记忆跨会话的历史经验、用户偏好属于完全不同的实现路径。工具技能Agent 能调用哪些外部能力比如搜索、计算、数据库查询、HTTP 请求。工具的输入输出 schema 设计直接决定模型能不能正确调用它。反馈驱动Agent 如何感知自身执行结果的好坏。这里的反馈既包括工具返回的结构化数据也包括环境状态变化甚至是用户的中途纠偏。安全边界什么能做、什么不能做哪些操作需要人工确认哪些资源不能触碰。这一层往往被初学者忽略但在生产环境中恰恰是最先爆雷的地方。运行循环以上六要素如何被组织成一个循环执行的流程是顺序执行、条件分支、还是支持动态回退的图结构。我在实际项目中用过一个比方七要素里的前两个目标角色、规划分解是 Agent 的“大脑皮层”负责想清楚做什么中间三个记忆状态、工具技能、反馈驱动是“四肢感官”负责动手干最后两个安全边界、运行循环是“自律神经”负责让整个系统平稳运转、不失控。1.2 为什么还要谈“七个决策点”理念到工程之间的距离光有七要素只是把 Agent 的解剖结构看清楚了。真到落地开发你会发现每个要素背后都藏着一个或多个需要你来拍板的技术决策这些决策直接决定项目能不能扛住真实业务场景。比如“记忆状态”这个要素看似简单但你要决定上下文窗口满了怎么办用向量数据库还是 Redis 存历史要不要做记忆压缩不同选择的成本、延迟、效果差异非常大。再比如“运行循环”你要决定用简单的 ReAct 循环、Plan-and-Execute 模式还是引入 LangGraph 这类图编排框架决策的复杂度直接上升一个量级。我把工程实现中最常见、最影响成败的七个决策点提炼了出来对应关系大致是这样七要素七决策点目标角色技术栈选型框架和语言的取舍运行循环流程控制循环逻辑和状态转移怎么设计记忆状态记忆实现长期记忆用什么存储短期记忆怎么管理反馈驱动状态管理执行进度和中间结果要不要持久化运行循环并发策略多个任务同时进来怎么扛住压力反馈驱动观测体系日志、追踪、错误恢复怎么落地安全边界安全审计权限隔离和敏感操作把关这七个决策点是我在做了多个 Agent 项目后总结出来的“高频决策清单”。每做一个新项目我都会先把这七个问题想清楚再动手写代码。这篇文章的后半部分会逐个讲解它们背后的原理和取舍。提示七要素是静态的结构拆解七个决策点是动态的工程选择。前者回答“Agent 是什么”后者回答“Agent 怎么做出来、怎么做好”。两者结合才算是真正把 Agent 从概念变成了可运行的系统。2. 系统架构与工具链选型解析2.1 四种主流架构模式哪种更适合你的场景在动手写代码之前架构选型是绕不开的第一步。我经历过从“裸调大模型”到“图编排框架”的完整演进过程这里直接说结论目前生产环境中真正能打的 Agent 架构主流有四类。单轮直调模式把用户请求和工具定义一次性丢给大模型让模型返回完整的工具调用序列。实现成本最低适合工具数量少、流程固定的内部工具。缺点是上下文容易越撑越大一旦超过模型窗口就得做截断而且模型一旦中途“走神”整个链路就断了。ReAct 循环模式核心是“思考-行动-观察”的循环。模型每一步先推理下一步该做什么调用工具拿到观察结果再继续推理。这个模式在工具调用场景下表现相当稳定也是目前 Agent 项目里最常见的默认架构。缺点是循环次数不好控制可能陷入重复调用同一个工具的“死胡同”。Plan-and-Execute 模式先让模型一次性生成完整计划再按计划逐步执行。优点是计划和执行解耦计划可以单独展示给用户确认可解释性很强。缺点是对计划质量要求很高一旦计划有问题执行阶段再想调整就比较笨拙。不过现在很多实现会引入“计划反思”在执行过程中动态修订后续步骤弥补了早期版本的僵硬。图编排模式以 LangGraph 为代表把 Agent 的流程显式建模为一张有向图节点是“大模型推理”或“工具调用”边是状态转移条件。这是现在我主力使用的架构因为它把 ReAct 的循环、Plan-and-Execute 的分支、甚至多 Agent 协作都能统一表达在同一张图里调试和扩展都方便。我个人的建议是如果你是第一次做 Agent 项目、或者业务场景是单轮工具调用直接选 ReAct 模式起步用 LangChain 的create_react_agent就能跑通闭环如果场景涉及多步骤编排、条件分支、人工审批或者你需要对执行过程做精细控制直接上 LangGraph。不要一上来就自己写状态机和循环等跑通了最小闭环再考虑自研编排层。2.2 我为什么选择 Rust FastAPI LangGraph 的组合做技术选型时很多人的习惯是“哪个火用哪个”但工程上我更看重团队的熟悉程度、生态成熟度和部署运维成本。这里分享一套我在生产环境里稳定运行了半年的技术组合以及背后的取舍逻辑。Rust 写工具层和并发网关Agent 系统里最耗费资源的是工具调用层尤其是涉及网络 IO、数据解析、并发聚合的场景。Rust 在内存安全、无 GC 暂停、高并发吞吐上有天然优势适合承担 Agent 的工具执行网关。我不建议用 Rust 去写全部业务逻辑因为 Rust 的开发效率相比 Python 还是有差距但把高频调用的工具层用 Rust 下沉收益非常明显。FastAPI 做接口层Python 生态在 AI 领域的统治地位短期内无法撼动LangChain、LangGraph 的核心逻辑以及各种模型的 SDK都是 Python 优先。FastAPI 的异步支持、自动生成 OpenAPI 文档、Pydantic 数据校验让接口层写起来非常顺手。我一般把 FastAPI 放在 Rust 网关之后作为业务编排层对外暴露的入口。LangGraph 管流程编排前面说了图编排是复杂 Agent 架构的最优解。LangGraph 的节点、边、状态、检查点机制已经把大部分流程控制问题抽象得很好了。而且它支持异步执行、支持状态持久化和 FastAPI 的 async 模型能很自然地融合。这套组合的部署形态我画过一张草图用户请求 → Nginx → Rust 网关限流、鉴权、转发→ FastAPI 业务层 → LangGraph 编排层 → 工具执行层Rust 沙箱进程池→ 外部系统和模型 API。每一层只做一件事职责边界非常清晰出了问题也能快速定位。当然这套组合不是唯一解。如果你的团队全是 Java 背景用 Spring AI 也能搭出不错的 Agent 服务如果你追求极致性能甚至可以用 Rust 写完整的 Agent 编排逻辑。我的核心观点是选型要匹配团队能力和业务阶段而不是追逐技术热点。3. 七个决策点的深度拆解与实操3.1 决策点一技术栈选型如何影响后续六个决策许多技术书籍把选型放在最后讲但我在实际项目里发现选型几乎是第一个要拍板的事因为它会连锁决定后续所有环节的实现方式。这就像一个房子打地基地基风格决定了上面能盖什么结构的楼。以我前面推荐的组合为例来做几个连锁推演选 Rust 做工具层意味着工具调用要以子进程或服务调用的方式暴露给 Python 业务层。Rust 子进程的好处是完全隔离Python 层的崩溃不会波及工具进程坏处是进程间的通信开销和序列化成本。我第一次做的时候走了弯路让 Python 直接调 Rust 编译的.so动态库虽然性能很好但一旦 Rust 侧 panic 整个 Python 进程也跟着出问题。后来改成独立的 Rust 服务 HTTP 调用虽然单次调用增加了 0.5ms 左右的延迟但系统稳定性直线上升。选 LangGraph 做编排意味着你的流程天然是“状态机”形态。LangGraph 的节点函数需要接收一个 State 对象并在执行完毕后返回 State 的新片段。这个模式在你后面做持久化、断点续跑时会省很多事因为 LangGraph 的检查点机制已经把状态快照做好了。选 FastAPI 做接口层意味着你的 Agent 天然是异步模型。这里的坑是LangGraph 的一些节点函数如果用了同步阻塞操作会把事件循环卡住。我后面的并发章节会展开讲这里先记住一个原则Async 边界要划在编排层工具层不要让 Python 碰阻塞 IO。3.2 决策点二流程控制是 ReAct 还是图编排流程控制是 Agent 项目里最容易“先简单后炸锅”的地方。一开始你只需要让模型调一两个工具用 ReAct 循环就绰绰有余但等业务复杂到需要多轮人工审批、条件分支、异常回退时ReAct 的 while 循环会变得根本无法维护。我的建议是无论项目初期多简单都用图编排的方式建模。哪怕是只有“意图识别 → 工具调用 → 生成回复”三个节点的图也比硬编码的循环更容易扩展。LangGraph 里定义节点核心就是写一个函数from langgraph.graph import StateGraph, START, END from typing import TypedDict, Annotated class AgentState(TypedDict): messages: list current_tool: str tool_result: str retry_count: int def call_model(state: AgentState) - AgentState: # 调用大模型决定下一步动作 response llm_with_tools.invoke(state[messages]) return {messages: state[messages] [response]} def execute_tool(state: AgentState) - AgentState: # 执行工具调用把结果写回状态 result call_tool(state[current_tool], state[messages][-1]) return {tool_result: result, retry_count: state[retry_count] 1} def should_continue(state: AgentState) - str: # 根据状态决定走哪条边 if tool_calls in state[messages][-1].additional_kwargs: return execute_tool return end graph StateGraph(AgentState) graph.add_node(call_model, call_model) graph.add_node(execute_tool, execute_tool) graph.add_edge(START, call_model) graph.add_conditional_edges(call_model, should_continue, { execute_tool: execute_tool, end: END, }) graph.add_edge(execute_tool, call_model)这段代码的核心逻辑是模型节点决定要不要调工具要调就走进工具节点工具执行完回到模型节点由模型判断是否继续。should_continue这个函数就是图里的“决策边”比起在 ReAct 循环里硬编码条件这种写法在增加分支时只需要加一条边不需要动循环主体。流量控制层面图编排还有个 ReAct 做不好的事情人工审批节点。很多动作比如发送邮件、执行删除操作不能由模型直接决定需要在图中设置一个human_review节点将流程挂起等用户确认后再继续。LangGraph 的interrupt_before参数正是为这个场景设计的graph.compile(checkpointercheckpointer, interrupt_before[human_review])这个能力在财务审批、内容发布等强管控场景里是刚需。3.3 决策点三记忆实现的长短期之分记忆是我认为 Agent 项目里“看起来简单、做起来最坑”的模块。很多初学者把记忆等同于“把历史消息塞进 Prompt 里”这在对话轮次少的时候确实够用但一旦用户和 Agent 的交互超过 20 轮或涉及多个会话之间的关联这种粗暴做法立刻崩溃。先捋清楚长期记忆和短期记忆在工程上的天然差异短期记忆指当前会话内的上下文。实现上一般用消息列表随着对话推进不断追加到接近模型上下文窗口限制时需要做截断或压缩。这里的关键决策点是“截哪些”最早的消息最长的工具返回还是用摘要替代长期记忆指跨会话的用户偏好、业务事实、历史结论。实现上通常需要外部存储常见方案是向量数据库存语义向量、按相似度召回或 KV 存储存结构化偏好精确读写。我踩过的坑是在长期记忆上直接用了向量检索结果用户偏好这种带“明确键值”的信息被向量化后召回率很不稳定。后来我把长期记忆拆成两层结构化事实层用 Redis Hash 按用户维度存储比如“用户的收货地址”“用户所在城市”这种信息不需要语义检索按 key 读就能拿到非结构化经验层用向量库比如“用户上周问过一个关于退货规则的问题”这类信息需要语义匹配。这样做的准确率比单用向量库高了不止一个档次。短期记忆的截断策略也需要提前想清楚。我在生产环境用的策略是“摘要 窗口”混合每一轮对话结束后用一个小模型比如 gpt-4o-mini 级别把前面所有内容压缩成一段摘要存在状态里真正发给主模型的上下文只保留最后 6 轮完整消息加上这一段摘要。这个方案在成本和效果之间平衡得很好。3.4 决策点四状态管理的持久化与恢复状态管理是 Agent 工程跟普通 Web 项目最大的不同点。普通接口是无状态的或者状态只在 HTTP 会话层面但 Agent 的执行过程是长链路、多步骤的中间任何一个环节出错都可能导致整个任务“白干”。所以把 Agent 的执行状态落盘是实现可靠恢复的前提。LangGraph 的检查点机制帮我省了很多事。它的原理是每个节点执行完后把当前状态快照存到持久化存储里。如果流程中断可以从最近的检查点恢复而不是从头开始。from langgraph.checkpoint.sqlite import SqliteSaver memory SqliteSaver.from_conn_string(checks.sqlite) graph graph.compile(checkpointermemory) # 执行时传入 thread_idLangGraph 会把同一条执行链路的状态绑定到这个 ID 上 config {configurable: {thread_id: user-123-task-456}} result graph.invoke({messages: [user_input]}, configconfig)这里有一个容易被忽略的细节检查点存的不只是消息内容还包括每个节点已执行的状态、当前待执行的节点、以及用户自定义状态里存的临时数据。这意味着你不但能做崩溃恢复还能实现“中断-恢复-继续”的交互模式比如用户在 Agent 执行到中途时要求“暂停一下我改个参数”这在电商订单流程、多步审批场景中非常实用。状态管理上另一个值得说的是不要把大对象塞进状态里。我见过有人把一整个 PDF 解析结果塞进 State 里结果检查点文件瞬间膨胀到几百 MB光序列化就卡好久。正确做法是状态里只存文件路径或数据库 ID需要内容时再去读。3.5 决策点五并发策略与性能优化实战Agent 系统跟普通接口的并发模型有一个本质差异单个 Agent 任务的耗时可能是分钟级甚至小时级的而普通接口通常在毫秒级返回。这意味着你的并发控制不能简单的“加个线程池”而是要深入理解任务的性质。会话内并发和会话间并发是完全不同的两个维度会话内并发同一个 Agent 任务里多个工具调用能不能并行执行比如你要同时查天气、查航班、查酒店这三个调用互相独立串行执行会白白浪费几秒甚至几十秒。LangGraph 里可以设计成“扇出-扇入”结构一个节点同时发起多个工具调用等待全部返回后聚合。会话间并发系统同时有多少个 Agent 任务在跑这里要做的是“异步化 排队 限流”。FastAPI 层面把任务直接交给后台 task 去跑用 Redis 做消息队列用独立的 worker 进程消费任务避免请求线程被长时间占用。我在生产环境里做压测时发现一个性能杀手Python 的 GIL 在大模型调用返回后、解析 token 流时会导致事件循环卡顿。这个问题在同步调用大模型 SDK 时尤其明显。后来我把大模型 SDK 的调用全部换成了异步客户端并且规定 FastAPI 层和 LangGraph 层都禁止使用同步阻塞的第三方库。扛并发的另一个关键点是把“计算密集”和“IO 密集”分离。大模型调用天然是 IO 密集等网络返回计算密集的部分主要在工具层的数据处理后处理。让 Rust 工具网关承担计算密集的部分Python 编排层专注异步调度整个系统跑下来 CPU 占用率和 P95 延迟都有明显改善。注意给并发策略做压测时不要只关注 QPS更要关注 E2E 延迟分布从请求进入到任务完成的整体耗时。Agent 任务的 E2E 延迟会因工具调用次数、模型响应长度波动巨大压测报告里至少要区分 P50、P95、P99 三个档位否则很容易误判系统容量。3.6 决策点六可观测性与故障排查体系Agent 系统的可观测性比普通 Web 系统要难做得多。原因在于一个 Agent 任务的执行轨迹是由模型动态决定的不是代码里写死的链路。你没法提前知道模型会调用哪些工具、走哪条分支。所以日志和追踪必须能完整还原“模型每一步在想什么、做了什么、拿到了什么结果”。我从头到尾坚持三个层面的观测覆盖Trace 层面每次完整的 Agent 任务生成一个 trace ID贯穿 FastAPI 入口、LangGraph 节点、工具调用、外部 API 请求全链路。同时在 LangGraph 的每个节点函数里注入 trace ID 到日志上下文确保所有日志可以按任务维度聚合。状态快照层面利用 LangGraph 的检查点机制在关键节点执行前后把状态打印出来或存起来。调试时可以一帧一帧回放 Agent 的执行过程定位“模型在哪一步产生了错误决策”。Token 与成本层面大模型调用是花钱的一个 Agent 任务可能内部调了十几次模型成本必须可观测。我在每次模型调用后统计输入 token、输出 token、缓存命中情况写入监控看板。上线初期我在这个环节踩过一个大坑Agent 在一个循环里反复重试同一个工具单次任务消费了几十万 token账单直接飙到预期之外。故障排查上最重要的一个心法是“不要让模型真的一点一点试错”。线上环境出现过 Agent 连续调用同一个工具三次、拿到同样的错误结果、仍然不肯停止的现象。后来我在每个工具节点的入参里加了retry_count字段把这个数字注入到模型的上下文里让模型“知道”自己已经失败多少次了并要求它失败两次后必须换一种方案。这种“外部状态注入”的方法比单纯在 Prompt 里写“请谨慎调用工具”有用得多。3.7 决策点七安全审计与权限隔离的底线思维安全审计是很多 Agent 项目的“事后补丁”但我强烈建议把它作为架构层面的内建能力。Agent 跟普通 API 的最大区别在于模型的决策是概率性的它可能某一次调用出现“越权冲动”而且概率不可归零。所以安全边界不能依赖模型自律必须在代码层面做强制约束。我在实际项目中执行的安全策略分成四层工具权限白名单Agent 能调用的工具集合是配置化的不是从模型输出里直接反序列化出函数名就调用。解析模型返回的tool_calls后先校验工具名是否在白名单里不在就直接拒绝。这个校验甚至可以基于一个合法函数的注册表避免模型“凭空捏造”工具名。参数强制校验工具入参要用严格的 JSON Schema 校验尤其是数字范围、枚举值、路径穿越这类问题。我曾经遇到模型生成{delete_file: ../../etc/passwd}这种参数虽然工具内部有路径校验没造成实际问题但想想都后怕。参数校验层务必由代码完成不依赖模型。敏感操作二次授权发送真实短信、扣款、删除数据、公开内容等操作在设计工具 schema 时就要标注为“高风险操作”。流程图上高风险操作必须经过human_review节点由用户确认后才能执行。这个约束要写死在编译后的图结构里运行时不可绕过。审计日志留痕模型的每一次决策、每一次工具调用、每一次参数修正都要记录。不只是为了排查问题更是为了答案“可追溯”。用户问“为什么扣款成功但系统没提示”你要能翻出当时的全链路 trace精准回答是哪一步出了错。另外还要考虑多租户场景下的数据隔离。如果你的 Agent 服务同时给多个租户使用工具调用涉及的数据读取必须强制注入租户 ID并在数据库查询层做强制过滤防止一个租户的 Agent 工具被模型错误地传参后读到另一个租户的数据。这种隔离要在 DB 层和 ORM 层做死不能只靠业务代码自觉。4. 完整实操过程从零搭建一个带并发能力和观测体系的生产级 Agent4.1 项目初始化与目录结构设计这部分我以一个真实的简化项目为例手把手带你跑通整个流程。项目需求是做一个“客服工单自动分类 处理建议生成”的 Agent输入是用户工单文本输出是分类标签、处理建议、以及是否升级到人工的判断。第一步建目录结构。我习惯按“接口层 / 编排层 / 工具层 / 存储层”四个维度拆分agent_project/ ├── api/ # FastAPI 接口层 │ ├── main.py # 入口注册路由 │ └── routes/ │ └── ticket_agent.py # 工单处理路由 ├── orchestrator/ # LangGraph 编排层 │ ├── graph.py # 构建图 │ ├── nodes.py # 节点函数 │ └── state.py # 状态定义 ├── tools/ # 工具层 │ ├── registry.py # 工具注册与白名单 │ ├── ticket_tools.py # 工单相关工具分类、查历史、升级 │ └── rust_gateway/ # Rust 实现的工具网关 ├── storage/ # 存储层 │ ├── memory_store.py # 长期记忆存取 │ └── checkpoint_db.py # 检查点数据库 ├── config/ │ ├── settings.py # 全局配置 │ └── prompts.yaml # 各类 Prompt 模板 └── tests/ └── test_agent_flow.py这个目录的关键是把“工具层”单独拆出来。工具是 Agent 与外部世界的唯一触点工具层混乱会导致安全审计和权限控制都很难下手。4.2 用 LangGraph 定义带分支与人工审批的 Agent 流程我设计这个工单 Agent 为五节点图意图识别 → 信息补充 → 工具执行 → 决策输出 → 人工审批。其中“人工审批”节点只在系统判断“工单需要升级处理”时才进入。节点定义的核心代码如下from langgraph.graph import StateGraph, START, END from typing import Literal class TicketState(TypedDict): ticket_text: str category: str suggestion: str need_human: bool human_confirmed: bool messages: Annotated[list, operator.add] trace_id: str def classify_node(state: TicketState) - TicketState: # 用模型判断工单分类 prompt load_prompt(classify_prompt).format(ticketstate[ticket_text]) resp llm.invoke(prompt) return {category: resp[category], need_human: resp[need_human]} def tool_node(state: TicketState) - TicketState: # 根据分类调用对应的处理工具 if state[category] refund: result call_refund_tool(state[ticket_text]) elif state[category] technical: result call_tech_help_tool(state[ticket_text]) else: result {suggestion: 无法自动处理转人工} return {suggestion: result[suggestion], need_human: True} def route_after_classify(state: TicketState) - Literal[tool_node, decision_node]: # 某些分类不调用工具直接生成建议 return tool_node if state[category] in (refund, technical) else decision_node def decision_node(state: TicketState) - TicketState: # 根据工具结果决定是否要人工审批 return {need_human: True} if state.get(need_human) else {need_human: False} def human_review_node(state: TicketState) - TicketState: # 挂起等待用户确认确认后继续 return {human_confirmed: True} builder StateGraph(TicketState) builder.add_node(classify_node, classify_node) builder.add_node(tool_node, tool_node) builder.add_node(decision_node, decision_node) builder.add_node(human_review_node, human_review_node) builder.add_edge(START, classify_node) builder.add_conditional_edges(classify_node, route_after_classify) builder.add_edge(tool_node, decision_node) builder.add_conditional_edges(decision_node, lambda s: human_review_node if s[need_human] else END, {human_review_node: human_review_node, END: END}) builder.add_edge(human_review_node, END) graph builder.compile(checkpointerSqliteSaver.from_conn_string(state.db), interrupt_before[human_review_node])这段代码里的几个关键细节值得展开interrupt_before[human_review_node]是让图在进入人工审批节点前自动挂起此时graph.invoke返回值中会带上__interrupt__字段。前端拿这个字段提示用户“需要确认”用户点击确认后用新的输入再次invoke图会自动从挂起点继续执行而不是从头跑。route_after_classify返回的字面量类型和条件边的映射字典保证了路由的正确性。LangGraph 会根据返回值去找对应的目标节点映射字典里没写的目标会直接报错这比裸写 if-else 安全得多。State 里的trace_id在节点间流转这样所有日志可以通过它关联到同一次任务。4.3 工具层的封装与 Rust 网关对接工具层是 Agent 最容易“脏乱差”的地方。我的做法是所有工具都用统一的装饰器注册并在注册时声明工具名、参数 schema、风险等级、是否需要人工授权。from tools.registry import register_tool from pydantic import BaseModel class RefundCheckParams(BaseModel): ticket_id: str user_id: str amount: float Field(gt0, lt100000, description退款金额必须大于0且小于10万) register_tool( namerefund_check, risk_levelhigh, # 高风险操作需人工授权 requires_humanauto, # 自动进入 human_review 流程 ) def refund_check(params: RefundCheckParams) - dict: # 这里对接 Rust 网关通过 HTTP 调用实际业务系统 gateway_url settings.RUST_GATEWAY_URL resp httpx.post(f{gateway_url}/refund/validate, jsonparams.model_dump()) resp.raise_for_status() return resp.json()Rust 网关的职责是接管所有外部系统交互这样业务系统不需要暴露大量 API 给 Python 层。我用 Rust 的axum框架写了几个核心接口每个接口都做了严格的参数校验use axum::{extract::State, Json}; async fn refund_validate( State(pool): StateDbPool, Json(params): JsonRefundValidateReq, ) - ResultJsonRefundValidateResp, AppError { let exists check_ticket_exists(pool, params.ticket_id).await?; if !exists { return Err(AppError::NotFound(ticket not found.into())); } let risk compute_refund_risk(params.user_id, params.amount)?; Ok(Json(RefundValidateResp { risk_score: risk })) }对比 Python 实现同样的接口逻辑在 Rust 里能做到更高的单位吞吐而且 panic 会被 axum 捕获转成 500 响应不会拖垮整个进程。如果你的团队 Rust 经验不足用 Go 做这个网关层也是个不错的选择核心思想是一样的把有状态、耗资源的外部交互全部下沉到独立进程中尽量少让 Python 编排层直接面对网络风暴。4.4 并发控制从同步阻塞到异步编排的改造实录我在早期版本里遇到过一个经典事故FastAPI 接口收到 20 个并发请求每个请求都在 LangGraph 里调用同一个大模型 SDK结果 Python 的事件循环被同步阻塞卡住后续所有请求排队等待接口 P95 延迟从 800ms 直接飙升到 6 秒。排查后发现根源在于 LangGraph 的节点函数里用了同步版的大模型客户端。改造方案分两步第一步把节点函数全部改成async大模型调用改用异步客户端async def classify_node(state: TicketState) - TicketState: async with async_llm_client.acquire() as client: resp await client.chat.completions.create( modeldeepseek-chat, messages[{role: user, content: prompt}], ) return {category: resp.choices[0].message.content}第二步LangGraph 的ainvoke本身就是异步入口FastAPI 路由直接 await 它app.post(/ticket-agent) async def handle_ticket(request: Request): body await request.json() config {configurable: {thread_id: body.get(trace_id)}} result await graph.ainvoke({ticket_text: body[text]}, configconfig) return {result: result}改造后压测效果明显在 50 并发下P95 延迟稳定在 1.2 秒左右CPU 占用也降了下来。并发控制里还有一个容易忽略的点避免多个线程或任务共享一个可变对象。比如把大模型客户端实例化在模块顶层那它就是被所有请求共享的。如果 SDK 内部有线程不安全的状态在高并发下会出现莫名的数据错误。解决方法是把客户端做成“每请求一个实例”或者使用连接池并确保 SDK 的线程安全声明。4.5 可观测性落地日志、追踪、成本监控一次讲透我在可观测性上的落地经验是不要等技术债积累到没法排查时才补一开始就要把三大件装上。日志方面我统一使用structlog结构化日志把 trace ID、节点名、工具名、耗时作为固定字段输出。举一个生产环境里的真实日志片段{ trace_id: conv-8f3a-20240512-001, node: tool_node, tool: refund_check, status: ok, duration_ms: 312, risk_score: 0.42, timestamp: 2024-05-12T10:23:45.123Z }这样的日志按 trace_id 聚合后可以复盘整个任务的完整时间线从入口到每一个工具的耗时分布、模型决策点、最终结果一目了然。追踪方面我在 FastAPI 中间件里注入 OpenTelemetry 的 span把 LangGraph 的执行轨迹透传到 Jaeger。LangGraph 支持自定义回调我写了一个简单的回调类在每个节点执行前后记录 spanclass TracingCallbackHandler: def on_node_start(self, node_name, state, *args, **kwargs): with tracer.start_as_current_span(fnode:{node_name}) as span: span.set_attribute(state.keys, list(state.keys())) span.set_attribute(trace_id, state.get(trace_id, )) def on_node_end(self, node_name, state, *args, **kwargs): with tracer.start_as_current_span(fnode_end:{node_name}) as span: span.set_attribute(state.output, json.dumps(state, ensure_asciiFalse, defaultstr))成本监控方面我在模型调用层包了一层统计装饰器每次调用后记录模型名、输入输出 token 数、价格、总成本定时汇总到 Prometheus。这里要特别关注 Agent 任务的“单任务成本”因为它跟普通 API 的单次调用成本完全不同。我在生产环境里给每个 Agent 任务设了成本阈值比如单任务模型费用上限 0.1 元超过阈值就强制中断避免模型在异常循环里产生额外费用。4.6 安全审计的代码实现权限隔离和二次授权怎么落地安全审计在前面的决策讲解里提到了设计层面这里给一套可运行的代码实现。我用一个中间件来控制所有进入 LangGraph 的工具调用确保任何工具在执行前都过了三道关卡from tools.registry import TOOL_REGISTRY from fastapi import HTTPException def safe_execute_tool(state: dict, tool_name: str, params: dict) - dict: # 关卡一白名单校验 registered TOOL_REGISTRY.get(tool_name) if not registered or not registered.enabled: raise HTTPException(status_code403, detailftool {tool_name} not allowed) # 关卡二Schema 参数校验 validated registered.schema_model.model_validate(params) # 关卡三风险等级处理 if registered.risk_level high: # 必须经过人工审批节点否则拒绝执行 if not state.get(human_confirmed, False): raise HTTPException(status_code400, detailhigh-risk operation needs human confirmation) # 记录审计日志 audit_logger.info(tool_executing, tool_nametool_name, paramsvalidated.model_dump(), trace_idstate.get(trace_id)) return registered.func(validated)这个safe_execute_tool是所有工具节点的统一入口。你在 LangGraph 的节点函数里调用工具时不要直接调业务函数而是走这个统一入口。这样安全策略的修改只需要改一处所有工具统一生效。关于人工审批我强烈建议把它做成“异步双向确认”Agent 挂起后系统通过消息队列或 WebSocket 通知前端前端展示工具名称和参数用户确认或否决后再通过另一个接口继续执行图。而不是简单地“等待同一个 HTTP 请求一直挂着”。原因很现实真实业务里人工确认可能就是几分钟HTTP 连接早超时了。用异步确认用户体验和系统稳定性都更好。5. 常见问题与排查技巧实录5.1 模型在工具调用时反复出错怎么缓解这是所有 Agent 项目最先遇到的坎。模型要么不按 schema 输出工具调用要么输出了但参数格式错误要么拿了工具结果却不继续往下走。排查思路按顺序来先确认 Prompt 里的工具描述是否足够具体。工具描述里要写清楚“这个工具是做什么的、输入什么、输出什么、在什么情况下调用”。我见过模型反复调用错误工具的原因居然是工具描述里的示例太模糊模型根本没理解两个相似工具的区别。再确认工具入参的 schema 设计是否简单直接。字段尽量用基本类型不要用深层嵌套的 JSON 对象。模型的 JSON 生成能力对嵌套结构很不友好。如果模型总是忘记带上某些必填参数可以把这些参数从“需要模型生成”改为“从外部状态注入”。比如用户 ID、trace ID 这类字段不要出现在模型的工具入参里而是节点执行时从 State 里取。容错层面我给工具节点加了一个“解析失败则重新生成”的循环机制最多重试三次把每次失败的原因拼接回 Prompt 让模型修正。这个策略在实际项目中能把工具调用的解析成功率从 80% 提升到 97% 左右。5.2 并发一高就报错或超时问题多半出在这三个地方第一看事件循环是否被阻塞。我之前已经讲过Python 异步环境里只要有一个同步阻塞调用整个进程都会卡住。检查方法很简单在 FastAPI 接口里加一个/health探针路由专门记录事件循环的响应延迟如果探针延迟异常起伏说明有阻塞。第二看外部依赖的客户端是否有连接池上限。大模型 SDK 默认的连接数可能只有 10并发一高就排队超时。解决方法是调大连接池、启用 HTTP/2 多路复用、或者加一层 Redis 缓存减少重复请求。第三看数据库连接池是否被打满。LangGraph 的检查点机制会在每个节点执行后写一次数据库如果你的任务有很多节点高频写入会很快打满连接池。解决方法是检查点库单独用一个连接池并设置合理的连接上限。另外Agent 任务本身是长耗时任务不建议直接在 HTTP 请求里同步等待完成。更好的做法是接口只返回task_id后台异步跑 Agent 任务执行完成后通过 WebSocket 或轮询结果接口告知前端。这样并发压力从“同时处理 N 个同步长连接”变成“同时处理 N 个异步任务”系统容量完全不同。5.3 长期记忆不准确优先检查这三点长期记忆召回不准很多时候不是向量库的问题而是数据组织方式的问题。第一检查存储的粒度。我上面说过用户偏好这类确定事实应该用结构化存储不要塞进向量库。向量召回适用于“模糊语义匹配”的场景如果你用向量库查精确事实召回不准是必然的。第二检查索引构建时有没有做切块。一篇很长的历史对话日志如果不切块直接向量化模型检索时只能命中整篇的近似向量精度会很差。正确做法是按语义段落或固定长度切块为每个块独立建索引。第三检查召回后的上下文拼接顺序。长期记忆召回的内容和当次会话内容的拼接顺序会影响模型的理解。我习惯把“用户基本信息”“历史偏好”“相关历史片段”按优先级放到系统提示词的固定位置而不是混在对话消息里。5.4 线上事故排查清单我每次发版前后必看一遍这份清单是我在踩了多次坑之后整理出来的每次发版前和出事故后都会逐条过一遍模型调用的超时时间是否设置合理。我见过大模型响应慢导致 Agent 节点一个接一个超时最后整条任务失败的场景。超时设置要比模型 P99 延迟高 20% 左右。工具调用的失败重试是否有限制。无限重试会让 Agent 陷入死循环增加成本也拖垮性能。所有重试必须设置最大次数。检查点存储是否有过期清理机制。如果不清理检查点数据会积压数据库越来越大恢复速度越来越慢。我设置了 7 天 TTL超过就自动删除。高风险操作的二次授权是否真的生效。测试时容易图省事直接绕过但线上必须确保。成本阈值是否配置到了每次模型调用。新模型上线时单次任务成本可能会因为模型更贵而翻倍成本阈值要跟着重新评估。6. 从“能做出来”到“能扛住”我的经验与建议6.1 先跑通最小闭环再追求架构演进我见过太多团队在第一个 Agent 项目上就追求“完美架构”结果三个月过去连一个能跑的闭环都没有。我的建议是第一版哪怕用最土的 ReAct 循环只要能跑通“用户提问 → 模型决策 → 工具调用 → 返回结果”的最小闭环就值得庆祝。最小闭环跑通之后再按七个决策点的优先级逐步演进。我自己的优先级排序大致是安全审计最先做这是底线其次是可观测性没有观测就没法优化再然后是并发策略业务量上来之前不用做最后是长期记忆和流程控制的高级特性。6.2 永远假设模型会犯低级错误这是我在 Agent 项目里学到的最重要的一课。模型不是确定性的程序它在 99% 的情况下表现正常但那 1% 的低级错误多了一个空格、把数字写反、凭空捏造工具名一旦发生在生产环境可能就是事故。所以所有关键环节都要用代码兜底不要相信模型的“自觉”。工具调用白名单、参数 schema 校验、结果二次校验、人工审批、成本阈值——这些机制不是为了限制模型的能力而是为了保护系统的确定性边界。在这个边界之内让模型自由发挥效果会好得多。6.3 给新手的三条实操建议第一多用 LangGraph 而不是自己写状态机这能让你更快理解 Agent 的节点和边模型。第二日志要结构化从一开始就养成带 trace ID 记录的习惯后面排查问题的收益是百倍级别的。第三先手工测试单条 Agent 任务的完整链路再上压测不要在没验证功能正确性时就开始调性能。我最后想分享一个项目里的体会Agent 工程跟传统后端开发最大的不同是你要跟一个“不可控的执行者”协作。传统的代码是你写什么就执行什么但 Agent 是模型在决定执行路径。所以Agent 工程的核心能力不是写代码本身而是设计边界和兜底机制的能力——让模型在边界内自由发挥在边界外永远别让它越线。这个能力只能靠实际做项目慢慢积累七要素帮你建立结构认知七个决策点帮你找到发力方向剩下的就是在真实项目里一遍遍调试、踩坑、优化的过程了。