做 AI Agent 有段时间了经常在群里看到两类提问一类是“Agent 到底是什么”一类是“Agent 怎么扛并发”。前者在讲概念后者已经是工程问题了。我自己经历过从 Demo 到上线的过程最大的感触是Agent 本身跑通很容易难的是把它当成一个真实服务来设计。所以我不打算再重复那些“大模型工具Agent”的空话而是用“七要素”和“七个决策点”这两条主线把 Agent 的工程实现完整拆一遍。七要素是零件回答“一个 Agent 跑起来需要哪些东西”七个决策点是岔路口回答“从原型到生产环境你在哪些地方必须做选择”。不管你是用 LangChain、LangGraph、CrewAI、Spring AI、扣子这类低代码平台还是干脆用 Rust 自研底层这套架构逻辑都是通用的。这篇文章适合有一定后端基础、想把手里的 Agent Demo 做成真实服务的开发者也适合刚入门的同学把这个框架当成学习地图。1. 先把底牌摊开Agent 的七要素1.1 目标与感知Agent 的任务书和传感器目标定义是 Agent 的第一个要素也是最容易被敷衍的环节。很多人写 System Prompt 就一句话“你是一个智能助手”这在 Demo 里没问题但放到真实业务里模型很快就会跑偏。我见过一个做内容审核的 Agent因为目标定义里只说“审核文本”没有写“审核标准是什么、判断不了怎么办、最终输出格式是什么”结果模型反反复复给中间推理过程而不是最终结论。目标定义在工程上要做成可验证的东西任务成功长什么样、失败长什么样、允许模型在什么范围内做决定。这句话听起来简单但实际落地时你会发现自己得把业务规则一条条翻译成模型能理解的约束条件。感知输入指的是 Agent 能拿到哪些外部信息。模型本身没有耳朵和眼睛用户消息、工具返回值、数据库查询结果、定时触发的任务上下文都属于感知层。工程上常犯的错是只把“最新一条用户消息”喂给模型却忽略了任务初始参数、历史上下文、甚至当前时间。比如一个做订单催付的 Agent如果感知层没有订单状态字段它就会靠猜来回答。感知层设计得好不好直接决定模型输出是“有据可依”还是“一本正经地胡说”。1.2 上下文、规划与工具Agent 的记忆、大脑和双手上下文是 Agent 的临时记忆也是工程实现里最烧钱的部分。每个模型的上下文窗口都是有限的一套对话跑二三十轮之后光历史消息就可能占掉大半窗口。常见的做法有三种直接截断、做摘要、接向量检索。截断最简单但会丢关键信息摘要保留主线但细节会失真向量检索适合知识库场景但召回不准时会引入噪声。实际项目里通常是组合拳——短期会话窗口存最近几轮中间层做滚动摘要跨会话知识走向量库。这块没有银弹但有一个原则上下文里的每一段内容都要想清楚“模型真的需要它来完成当前任务吗”不需要的就别塞进去省 Token 也是在降本。规划是 Agent 区别于普通 API 调用的核心要素。ReAct 模式是边想边做每一步根据当前观察决定下一步动作Plan-and-Execute 是先列计划再逐项执行执行过程中可以调整计划。规划模式的选型直接影响成本和延迟ReAct 灵活但模型要反复被调用Token 消耗高Plan-and-Execute 第一轮就要输出完整计划适合步骤相对固定的任务。工具层则是 Agent 能力的边界。不要以为多接几个工具就好工具描述本身就是一种提示词写得不清楚模型就不知道该什么时候用、什么时候不用。我见过一个 Agent 死活不肯调用查询接口后来发现是工具描述里没写清楚“这个接口用于什么场景、返回什么字段、失败会怎样”模型根本判断不了该不该用它。1.3 执行与反馈Agent 闭环里最容易被低估的两个环节执行就是把规划转化为真实动作。这里有个工程细节很容易被忽略工具调用要有事务感和幂等设计。比如 Agent 调了一个“创建订单”的接口第一次超时了它重试了一次结果下了两单。模型层的事务很难保证原子性但执行层可以加一层防护——所有写操作之前先检查幂等键超时后重试带上同一个幂等 ID。这听起来更像后端工程而不是 AI 工程但恰恰是这种细节决定了 Agent 能不能在真实生产环境里待得住。执行完成后还要校验结果不能默认工具调用成功就等于任务完成。工具返回里的业务错误码、模型输出里不合法的 JSON、部分成功部分失败的结果都要有对应处理。反馈是闭环的最后一环也是最容易被砍掉的一环。很多 Agent 跑完一个流程就直接返回结果不做质量评估。但生产环境里模型输出不稳定的问题一定会出现而反馈节点就是兜底的那道防线。可以手动做一个校验节点比较输出格式、检查关键字段是否缺失、模拟测试用例跑一遍。更轻量的方式是让模型自己反思一遍输出虽然会多花一次模型调用但能明显提升成功率。反馈结果同时要回流到系统里失败的原因记录下来后续要么滑到人工处理要么做重试。没有反馈闭环的 Agent 本质上只是一个带工具的对话接口出问题只能靠人工盯日志那就失去 Agent 的意义了。2. 七个决策点每个按钮按下去之前想清楚2.1 决策点一框架选型——自研、LangGraph、CrewAI还是低代码框架选型在初期看起来是个技术偏好问题实际是掌控权问题。用 LangGraph 这类图形化状态机框架好处是你对 Agent 的控制流有完整掌控每个节点的输入输出落库、回环条件、中断恢复都明确可查代价是要写更多样板代码。CrewAI 偏高层抽象适合快速搭建多智能体协作场景但内部状态流转被封装了一层排查问题要翻框架源码。低代码平台比如扣子这类可视化编排平台上手最快适合快速验证想法但到后期定制化需求一多平台能力会变成天花板。Spring AI 适合 Java 生态Rust 自研则是极致性能和可控性的选择但这两条路的工程成本都不低。我个人的建议是先问自己“这个 Agent 的控制流会不会复杂到需要显式画出来”。如果只是单轮工具调用用最轻的框架就好如果涉及多轮循环、条件分支、人工审批、状态持久化直接上 LangGraph 这类状态机别竞品用个简单封装就跟着选。框架选型的本质是选“状态管理的复杂度”放在哪一层——放在框架里你就要吃透框架放在自己代码里你就要能兜住所有边角情况。2.2 决策点二任务编排——工作流为主还是自主循环为主任务编排的决策点是“控制权和灵活性的取舍”。工作流模式是提前画好流程图节点顺序固定模型只是在节点内部求值自主循环模式是给模型一个目标和一组工具让它自己决定调用顺序直到认为任务完成。前者像给员工发一张 SOP 表格后者像让员工自由发挥。真实业务里完全自主的循环反而不靠谱模型很容易在中间偏航而且 Token 消耗不可控。我现在的做法是“工作流为骨架模型在骨架内做局部决策”任务必经的阶段显式写出来比如“输入校验→计划生成→执行工具→结果校验→回复生成”每个阶段内部的细节才让模型自主选择工具和参数。这里有个实用的判断标准如果这个任务的关键步骤顺序是固定的那就写成工作流别让模型去“规划”一个固定流程如果任务本身是开放式的比如“研究这个问题并输出报告”才用自主循环。LangGraph 用条件边就能灵活实现两种模式混搭——执行完强制回到校验节点校验通过才放行进入下一步。自主循环不是不能用而是要把循环的退出条件写死而不是等模型自己“觉得做完了”。2.3 决策点三记忆与状态——该记什么、存哪里、多久清一次状态管理是 Agent 工程实现里最容易被新手跳过、又最影响体验的一环。很多团队把 Agent 状态直接放在内存里服务一重启全部丢失用户追问一句“上次你说帮我查的那个事怎么样了”Agent 直接失忆。生产环境里状态要持久化最常见的方案是给每条任务一个 session_id状态数据存在 Redis 或数据库里不仅记录对话上下文还要记录“任务当前执行到哪一步、已经调用了哪些工具、结果是什么、哪些步骤已完成”。这就像工作流引擎里的流程存根宕机恢复就靠它。记忆策略上要给不同类型的信息区分生命周期会话内的短期记忆放 Redis过期时间设短跨会话的用户偏好和长期事实放向量库或 KV 存储任务执行的中间产物单独落库不跟对话上下文混在一起。还有一个经验Agent 内部思考的中间状态比如计划草稿、工具调用返回值不要一股脑塞给模型当对话历史否则上下文膨胀得飞快。先想清楚“哪些状态是模型推理必需的”其余全部放外部存储需要时再检索。2.4 决策点四并发模型——先算 QPS再谈架构“AI Agent 怎么扛并发”是热词但很多人的出发点错了一上来就问用 Kafka 还是 Redis Stream其实第一步是算清楚你到底需要扛多少并发。我给个可复用的估算方法先统计单任务平均耗时假设一个 Agent 任务要调 6 次模型每次模型响应平均 8 秒外加工具调用 2 秒单任务总耗时大约 50 秒。如果你的业务目标是每分钟处理 30 个任务0.5 QPS那么同时需要的 worker 数大约是 30 × 50 / 60算出约 25 个并发执行槽。这还只是理论值模型 API 还有 RPM每分钟请求数和 TPM每分钟 Token 数限制要把重试导致的耗时膨胀和 30% 的峰值冗余算进去。工程实现上我不建议为了“并发”一上来就上复杂的消息队列。先用异步 IO 信号量限流 进程内任务队列起步扛不住再引入外部队列。这里有一个我自己踩过坑的细节LangGraph 实例本身不是线程安全的每个请求必须 new 一个 graph 实例或者用带 session_id 的 checkpoint 隔离状态千万不能用一个全局 graph 对象接所有请求。FastAPI 的异步接口里用 asyncio.Semaphore 做并发控制既能限制同时运行的 Agent 任务数又能避免流量突然打满模型 API。等单机撑不住的时候再考虑多实例部署把 agent 执行做成独立 worker 服务。2.5 决策点五可观测性——Agent 最缺的不是能力是“回放”Agent 系统比普通 API 服务难排查的地方在于同样的输入两次输出可能完全不同工具的调用链也可能不一样。没有 trace出问题只能到处打日志然后靠猜。我的建议是给每个 Agent 任务生成一个 trace_id把每次模型调用、工具调用、节点流转、Token 消耗全部记录到一个可检索的存储里。开源工具里 Langfuse 和 LangSmith 都值得试前者支持自托管后者和 LangChain 生态集成好。如果不方便接第三方退一步也要自己建一张任务审计表至少记录任务输入、计划结果、每步工具名和参数、工具返回摘要、模型输出、最终结果、耗时和 Token 数。成本核算也归在这一决策点。模型输出的 Token 消耗不是小数目一个跑 6 轮调用的任务光输出 Token 可能就上万。我把每次调用的输入/输出 Token 数累计到任务级别再按模型单价换算成金额每天跑一个成本报表。没有这层观测你根本不知道哪些任务类型在偷偷烧钱等问题积累到账单爆掉才发现就晚了。2.6 决策点六安全与治理——工具权限、注入防护、人在环路Agent 最大的安全隐患是工具权限过大。模型不是你它分不清哪些操作后果严重。核心原则是最小权限给 Agent 的工具尽量只读写操作的接口必须标清楚幂等键涉及支付、下单、发送消息等敏感操作必须在流程里插入人工确认节点模型只能生成“待审批”请求由人来执行最终动作。我之前做用户运营 Agent 时所有触达类操作都走“Agent 生成内容→人工一键确认→系统发送”的链路虽然牺牲了一部分自动化但换来的是敢下放权限的底气。提示注入要单独提一下。工具返回的内容如果是外部输入比如网页内容、接口返回的文本里面可能藏有恶意指令这时候直接把内容拼进 LLM 上下文等于让外部内容控制你的 Agent。处理方式是隔离外部内容只作为“数据”传给模型不与 system prompt 混在一起同时校验工具返回内容里是否包含可疑指令模式。还有一类风险容易被忽视——个人把 Agent 用在金融交易这种真实资金场景里比如做期货交易哪怕只是辅助决策也必须看清楚策略逻辑是否经过回测验证涉及真金白银的操作建议先做人工审批不要指望一个用大模型包装的工具能替你扛市场风险。同样的接入第三方平台做自动化运营前先确认平台规则允许的范围不要拿账号安全去赌。2.7 决策点七部署形态与模型路由——单模型还是多模型分工最后一个决策点是部署形态。模型路由用得好成本和效果能同时优化。大模型虽然聪明但贵且慢小模型便宜但能力有限。实际项目中我习惯做分层规划类步骤用能力强的大模型因为逻辑推理质量决定整个任务上限摘要、改写、格式整理这类机械步骤用便宜的小模型工具调用结果校验这种确定性任务甚至可以不经过模型直接用代码判断。做路由时给不同步骤配置不同的 model_nameLangChain 里一个 model 参数就能切换成本立刻降下来。部署形态上如果 Agent 有大量耗时操作不太适合塞在 API 网关后面做同步请求响应。我会把 Agent 执行拆成独立的 worker 服务API 层只负责接收任务、写入队列、返回“受理成功”的状态Agent 服务从队列拉任务执行结果写回到任务表。前端轮询或 Webhook 拿结果。这样用户不用傻等几十秒后端也不容易被慢请求拖垮。Serverless 不是不行但要注意冷启动加上模型调用延迟对交互式 Agent 来说体验可能不太理想。3. 实操环节用 FastAPI LangGraph 搭一个能接真实请求的 Agent3.1 准备工作与最小结构状态定义、节点与工具注册这一节我以自托管部署最常用的组合为例FastAPI 做服务入口LangGraph 做状态图编排。这也是目前社区里最常见的路径如果你用的是 Spring AI 或 Rust 自研架构思路完全一致只是落地的库不同。第一步是定义 AgentState它承载整个任务的生命周期数据。不要把所有内容都堆在 messages 里把计划、工具结果、重试次数、最终答复都拆出来这样既方便在节点间流转也方便之后做持久化。from typing import TypedDict, Annotated import operator class AgentState(TypedDict): messages: Annotated[list, operator.add] # 对话历史 task_input: str # 任务初始输入 plan: str # 当前计划 tool_name: str # 当前要调用的工具名 tool_args: dict # 工具参数 tool_result: str # 工具返回结果 retry_count: int # 反馈节点重试次数 final_answer: str # 最终回复工具注册这一步LangChain 里用 tool 装饰器但我更建议手动定义一个列表统一管理每个工具的“名字、描述、参数结构、实际执行函数”。这样做的原因是工具描述本身是给模型看的提示词集中放一个文件里方便调优。tools [ { name: query_order, description: 根据订单ID查询订单状态适用于用户询问订单进度时调用。输入为订单号字符串。, parameters: {order_id: string}, function: query_order_impl, } ]3.2 核心节点与循环规划、执行、校验三节点用 LangGraph 实现一个“规划→执行→校验”循环是覆盖大部分业务场景的最小闭环。plan 节点负责拆解任务并输出当前步骤这里我用一个结构化提示词让模型输出 JSON 格式的“下一步动作”。executor 节点读取 plan 中的工具名和参数执行工具并把结果写回 state。validator 节点判断任务完成还是继续循环。from langgraph.graph import StateGraph, END def plan_node(state: AgentState): # 调用 LLM 输出计划这里简化成从 llm 拿到 action JSON action llm.invoke( f根据任务{state[task_input]}和历史消息输出下一个动作的JSON ) return {plan: action, messages: [(assistant, action)]} def execute_node(state: AgentState): action state[plan] tool find_tool(action[tool_name]) result tool[function](**action[args]) return {tool_name: action[tool_name], tool_result: result} def validate_node(state: AgentState): # 判断是否完成或是否需要回退到 plan_node if state[tool_result]: # 这里替换成真实校验逻辑 return {final_answer: state[tool_result]} elif state[retry_count] 2: return {retry_count: state[retry_count] 1} else: return {final_answer: 执行失败已超过最大重试次数} graph StateGraph(AgentState) graph.add_node(plan, plan_node) graph.add_node(execute, execute_node) graph.add_node(validate, validate_node) graph.add_edge(plan, execute) graph.add_edge(execute, validate) graph.add_conditional_edges(validate, lambda s: END if s[final_answer] else plan) app graph.compile()我把“校验节点”特意放在执行之后而不是让模型自己判断“做完没”。这是从无数失败案例里总结出来的模型对“是否完成”的判断远没有代码校验可靠。如果你要处理更复杂的场景把校验逻辑写成纯函数用工具结果里的业务字段做断言只有断言通过了才允许进入下一步。3.3 用 FastAPI 暴露接口并接入并发控制服务层我用 FastAPI 的异步接口暴露 Agent 能力。这里有一个关键点千万不要把 graph 对象作为全局单例直接供所有请求调用并发场景下状态会串。正确做法是给每个请求生成一个独立的 graph 实例或者用带 thread_id 的 checkpointer 隔离状态。我习惯在请求上下文里按 session_id 编译一个实例并用 asyncio.Semaphore 做信号量限流。from fastapi import FastAPI import asyncio app FastAPI() semaphore asyncio.Semaphore(20) # 同时最多跑 20 个 Agent 任务 async def run_agent_task(task_input: str, session_id: str): async with semaphore: # 用 ainvoke 异步调用编译好的 graph config {configurable: {thread_id: session_id}} result await graph.ainvoke( {messages: [], task_input: task_input, tool_name: , tool_args: {}, retry_count: 0}, configconfig ) return result[final_answer] app.post(/agent/run) async def run_agent(req: RunRequest): # 这里应该用后台任务或队列避免同步阻塞简化起见直接 await result await run_agent_task(req.task_input, req.session_id) return {answer: result}信号量的数量不是拍脑袋定的。结合前面的 QPS 估算公式假设单任务 50 秒目标 0.5 QPS需要的并发槽是 25。信号量设得太小会排队设得太大又容易打爆模型 API。我建议压测跑一轮观察模型 API 的延迟和限流情况再调。接口层如果任务耗时太长前端等不起就需要改成“提交任务→返回 task_id→后台 worker 执行→前端轮询结果”的模式这一步我留到下一节说。3.4 从同步接口到异步任务把 Agent 放到后台 worker当任务耗时超过十几秒同步接口的用户体验就崩了。我的处理方式是把 Agent 任务丢到消息队列或任务表主服务立即返回 task_id后台 worker 慢慢跑结果写回任务表前端轮询或通过 Webhook 获取结果。这个改造点正好把“Agent 逻辑”和“任务调度逻辑”解耦以后扩容也只扩 worker 服务就行。# 任务表结构简化 task_id uuid session_id varchar status varchar -- pending/running/success/failed input_data jsonb output_data jsonb created_at timestamp updated_at timestamp队列长度要盯着这是“扛并发”的关键指标。如果 pending 状态的任务积压超过预设阈值那就是 worker 跟不上请求速度该扩容了。后台 worker 的并发数同样用信号量控制同时它本身就是个独立的常驻进程崩溃重启后可以根据任务表里的 running 状态做恢复——结合 3.2 里讲的 checkpointer把半途任务重新拉进 graph 继续跑而不是从头开始。4. 常见问题与排查技巧实录4.1 模型输出不稳定JSON 解析失败、字段缺失这是 Agent 上线后遇到最多的坑。模型告诉你“我会输出 JSON”结果输出里带了 Markdown 代码块或者字段名变了或者直接给出自然语言回答。我的建议是不要依赖“模型保证格式”而是在解析层做防御先用正则提取 JSON 片段解析失败就触发一次修复节点把“格式错误原因原始输出正确示例”一起塞给模型让它重新输出。同时提示词里给出一个 few-shot 示例比单纯写“请输出 JSON”有效得多。4.2 工具调用循环Agent 卡死或重复调用同一个工具Agent 有时候会陷入同一个工具调用循环比如模型反复查询同一个订单号就是不推进下一步。排查时先看 trace 里每步的工具调用参数和模型输出如果模型的下一步动作确实还是一样说明它自己绕不出来。这种场景靠“更多提示”是没用的要靠硬限制LangGraph 里设置 recursion_limit控制整个图最多执行多少步每个节点设置超时时间工具调用结果里加入“此信息已查询勿重复调用”的标记。防线越多越好真实环境里绝对不能只指望模型自觉收敛。4.3 并发下的状态污染A 用户用到 B 用户的数据这个问题的根源基本都是全局变量或共享 graph 实例。LangGraph 的 graph 对象如果被多个任务复用作图运行时state 会被后一个任务覆盖。排查方式是检查部署日志里两个 session_id 的 trace_id 是否交错到同一个实例上。修复不难要么每个请求创建新实例要么确认 checkpointer 配了 thread_id 做隔离。我自己的习惯是服务启动时只建一次图定义graph 是可复用的元结构每次请求用 compile 结果搭配不同 thread_id这样性能和隔离都能兼顾。FastAPI 接口层尤其要小心全局缓存所有任务级数据一律跟着 session 走。4.4 上下文爆掉Token 费用飙升、模型开始遗忘前文多轮 Agent 跑久了上下文不可能一直塞。我遇到过一个运营场景任务执行到第 8 轮时模型已经把最初的需求忘干净了因为中间工具返回的日志塞满了窗口。解决分两层每轮工具调用返回结果先截断再进上下文超长内容哈希后存外部存储需要时再查任务整体加一个“上下文预算”超过阈值触发摘要节点把前面内容压缩成一段摘要替代原始历史。这个摘要节点可以用便宜的小模型来做成本低效果好。另外现在主流模型支持 system prompt 和对话历史分离工具输出尽量放在 tool 角色消息里不要往 system 里塞。4.5 问题速查为方便后续排查这里做一个速查表对应上面的问题以及对应的处理动作。现象可能原因排查思路解决办法JSON 解析失败模型输出被 Markdown 包裹/字段缺失查看原始输出日志正则提取 修复节点 few-shot同一个工具反复调用模型未理解“已执行”步骤看工具调用链参数设置 recursion_limit 结果标记用户请求串数据全局 graph 实例被并发复用对比 trace 的 thread_id按 session_id 隔离状态实例Token 费用飙升工具返回全文进上下文统计每轮 token 消耗截断工具返回 滚动摘要 上下文预算模拟后结果与预期不符反馈节点只做了“模型自我判断”用代码校验工具返回字段换成确定性校验逻辑请求超时模型调用延迟与工具延迟叠加看单步耗时统计接口异步化 队列 后台 Worker排查问题的时候第一件事永远是看 trace。没有 trace 的 Agent 系统所有排查都是猜。这也是我强烈建议大家在项目第一天就接可观测性工具的原因等出了事故再补你会发现自己连“事故时的模型输出长什么样”都不知道。再补一个真实经验Agent 的并发问题通常不是框架问题是资源上限问题。你以为要上 K8s其实只是模型 API 的 TPM 被打满。挂在队列外面做限流远不如控制住每个任务内部的模型调用次数来得实在。把规划、执行、校验的流程设计得越紧凑并发能力天然越强这个收益比任何缓存优化都明显。七要素解决的是“Agent 能不能跑通”七个决策点解决的是“Agent 能不能跑久”。我在不同项目里反复迭代这套框架之后最大的感受是技术选型重要但没有重要到决定生死真正拉开差距的是你对状态、并发、可观测和安全这些工程细节的把控。如果你现在手里有个 Agent Demo建议按七要素先检查一遍闭环是否完整再拿着七个决策点逐条过一遍——特别是记忆持久化和工具权限这两条90% 的上线事故都出在它们身上。
