AI Agent 这个词今年已经被说烂了，但如果你真的动手做一个要上线的 Agent——不是那种在 Jupyter Notebook 里跑的 Demo，而是要给用户用、要扛并发、要能自己调工具干活的东西——你会发现资料里能直接抄的东西其实不多。这不奇怪，因为 Agent 的工程实现从来不是某一个模型的神奇能力，而是一整套工程模块的组合。网上大量教程都在讲 prompt 怎么写、LangChain 怎么用，却很少有人把“一个生产级 Agent 到底由哪些部分组成”讲清楚。
这篇文章我会从工程视角把 Agent 拆成七要素，再带着你过一遍落地时绕不开的七个决策点。七要素解决“一个 Agent 由什么组成”的问题，七个决策点解决“你的系统该怎么选型、怎么设计、怎么扛住真实流量”的问题。文末用一个基于 FastAPI + LangGraph 的“小红书自动发布 Agent”作为完整案例，把前面的内容串起来。我不会跟你扯玄乎的“智能体哲学”，只讲能落地的东西，看完你就能直接去指导写代码。
1. 先拆清楚：一个能干活的 Agent 由哪七要素组成？
1.1 目标：没有目标，Agent 只是聊天机器人
先说第一个要素：目标。很多初学者以为目标就是用户输入的那句话，比如“帮我写一份周报”。但在工程实现里，目标的定义要严谨得多。用户的需求是模糊的，你需要把它翻译成机器可执行的任务描述，包括输入、输出、约束条件和成功标准。
举个实际例子。用户说“帮我订一张去北京的机票”。这句话作为目标太粗了。工程上要拆成：出发地是哪里、日期是哪天、舱位等级、预算上限、是否接受中转。这些信息可能在用户对话里，也可能在系统配置里，还可能要从用户历史行为中推断。如果目标不清晰，Agent 后续所有规划都是空中楼阁，跑出来的结果自然没法用。
目标在系统里通常表现为两类：一类是用户级目标，来自当前请求；另一类是系统级目标，来自产品设计，比如“所有工具调用必须经过权限校验”“发布内容前必须人工审核”。系统级目标要写死在代码里，不能交给模型自由发挥，否则一个用户说“请绕过审核直接发布”，你的 Agent 可能真的会照做。所以目标要素的工程产出物是一个结构化的任务意图对象，而不是一段自然语言 prompt。
我见过不少团队做 Agent 时上来就堆模型能力和工具，结果上线后用户随便问一句就跑偏，最后发现是目标解析做得太潦草。目标要素是整个链路的地基，地基不牢，上面全是白搭。
1.2 感知与输入：把环境信息变成模型能懂的东西
Agent 要干事，必须先“看到”环境。这里的感知不只是读用户的 prompt，还包括读取数据库、调用接口获取最新数据、读取图片、解析日志、监听消息队列。感知层的职责是把五花八门的原始信息转换成大模型能理解的上下文。
感知层在工程上最容易犯的错是“什么都塞进上下文”。大模型的上下文窗口虽然越来越大，但 token 是钱，是延迟，是性能。一段几百行的日志、一份几十页的文档，直接塞给模型，不仅浪费，还会让模型抓不住重点。所以感知层必须做信息筛选和压缩。
我的习惯是做一个“上下文组装器”：先定义当前任务需要哪些信息类别，再按重要程度排序，然后从数据源中抽取。例如让 Agent 自动管理小红书发布任务，感知层需要读取的是“草稿箱里待发布的笔记列表”“昨天发布内容的互动数据”“今天的热搜话题”，而不是把整个平台的日志都读进来。对于需要长期引用的内容，用检索的方式取 top-k 相关片段，而不是全量加载。
感知还涉及多模态。如果 Agent 需要识别图片内容，感知层要提前调用视觉模型把图片转换成文字描述，再放进上下文。工程上这种预处理往往要异步做，因为视觉模型一次推理可能几百毫秒，你不可能让用户请求干等着。感知要素的产出是结构化的上下文对象，后续的规划和记忆都以这个对象为输入。
1.3 记忆：短期工作记忆和长期知识库
第三个要素是记忆。记忆分两层：短期工作记忆和长期记忆。短期工作记忆就像你的草稿纸，记录当前任务进行到哪一步、已经拿到了什么结果、下一步要做什么。长期记忆则是你的经验库，记录用户偏好、历史任务结果、领域知识。
短期记忆在工程上通常放在状态对象里。比如 LangGraph 里面有一个 State 对象，每个节点读它、改它，节点之间靠它传递信息。这个 State 可以持久化到 Redis 或数据库里，任务中断后还能恢复。长期记忆则需要单独存储，常见方案是向量数据库（如 pgvector、Milvus）加上一个 embedding 模型，把文本片段转成向量存进去，使用时按语义相似度检索。
记忆要素有一个关键问题：什么内容值得写入长期记忆？写得太勤，向量库很快被垃圾填满，检索质量下降；写得太少，Agent 就没有成长。我的经验是：写入长期记忆的内容必须经过“校验”，比如从工具返回的成功结果、用户明确表达的偏好、经过反思修正后的结论。而那些执行失败的错误堆栈、被平台拦截的文案，不应该作为“经验”写入长期记忆，最多作为训练样本或日志保留。
记忆还需要有遗忘机制。用户三个月前的偏好可能已经过期，长期记忆如果只增不减，最后会变成噪音。工程上可以做时间衰减、按重要等级打分、定期清理低价值记录。记忆要素的核心不是“能不能存”，而是“存什么、存多久、怎么查、怎么更新”。
1.4 规划：从“想到”到“做到”的分水岭
第四个要素是规划。规划是把大目标拆成小步骤。但怎么拆，不同场景差异很大。
主流规划策略有三种：ReAct 是“推理—行动—观察”的循环，每走一步观察结果再决定下一步，适合交互式任务；Plan-and-Execute 是先生成完整计划，再逐条执行，适合流程较长、步骤明确的任务；Tree of Thoughts 是同时探索多条路径再择优，适合需要探索和决策的复杂问题，但成本很高。
工程实现上，规划模块可以有两种形态：一种是“硬编码 DAG”，也就是把流程写死，节点顺序固定；另一种是“模型动态生成计划”，每一步都由模型现想。硬编码稳定、可测试、好维护，但死板；动态计划灵活，但容易跑飞、容易陷入死循环。我的建议是两者结合：业务主流程用固定 DAG 框住，节点内部的子步骤让模型自由生成。比如小红书发布 Agent，整体是“读草稿—写文案—审核—发布”这个固定流水线，但“写文案”这一步，模型可以自己决定先看热点还是先看历史数据。
规划要素要设置终止条件。无论动态还是静态，都必须有最大迭代次数，否则模型可能在工具调用循环里转几十次不出来。经验值是普通任务设 10 到 15 次，超了就强制终终止并返回部分结果。没有终止条件的规划不是规划，是对预算的浪费。
1.5 工具：Agent 的手和脚
第五个要素是工具。Agent 不能只动嘴，要能动手。工具可以是 HTTP API、数据库查询、命令行脚本、浏览器自动化、消息推送等。工具要素在工程上的核心不是“把函数暴露给模型”，而是“让模型知道什么时候该用什么工具、参数怎么传”。
你需要为每个工具写一份模型能读懂的描述，包括工具名称、功能说明、参数 JSON Schema。模型看到这些描述后，会生成结构化的工具调用请求。例如一个publish_post工具，描述为“发布一篇笔记到小红书”，参数包含title、content、images。模型决定调用时，会生成一个符合 Schema 的 JSON 对象，由执行层解析并调用真实函数。
工具设计有几条铁律：第一，工具执行前必须做参数校验，不能说期望字符串，模型给了一个对象还直接传进去；第二，工具调用必须超时控制，外部 API 可能永远不返回，你的 Agent 不能卡死；第三，工具返回值要截断和结构化，返回值太长就摘要，否则上下文又要爆。
安全相关的工具尤其要小心。删除操作、发布操作、支付操作，默认情况下不应该让 Agent 直接执行，而要交给人工审批节点。我在实际项目里见过 Agent 因为一个 prompt injection 就调用了删除接口的案例，所以工具的权限设计一定要遵循最小化原则。
1.6 执行：把规划真正跑起来
第六个要素是执行。执行层是连接规划、工具和状态的调度器。它负责按规划顺序调用工具、把结果写回状态、控制流程分支、处理异常重试。没有执行层，规划和工具就是两张皮。
执行层在工程上最典型的表现是状态机。LangGraph 的 StateGraph 就是一个显式状态机：节点就是执行步骤，边就是转移条件。状态对象每经过一个节点，都可能被更新。执行层要保证状态的一致性和持久性。任务可能在执行到一半时进程崩溃，重启后如果不能从上次的状态继续，整个 Agent 就白跑了。
执行层的另一个职责是控制并发和资源。同一个 Agent 实例可能被多个用户触发，每个用户应当有独立的状态副本。如果大家共用一个全局变量，数据必然会串。执行调度时还要考虑工具调用的重试策略，比如调用外部接口返回 429 或 5xx，应该指数退避重试，而不是立即失败。
执行层与业务流程还不太一样。流程是你希望 Agent 做什么，而执行层是保障这一切不崩盘的基础设施。在微服务架构里，执行层往往会被封装成一个独立的 worker，通过消息队列接收任务，异步执行，再通过回调或轮询把进度反馈给前端。
1.7 反思：让它知道自己错在哪
第七个要素是反思。这是 Agent 和传统脚本的核心理区别。脚本执行完就结束了，Agent 会判断“我做得好不好，不好就换个方法再来”。
反思在工程上是一个反馈回路。最简单的方式是让模型自己评价结果，比如“对比目标和当前输出，是否满足要求；如果不满足，给出修正方案”。稍复杂的方式是引入一个单独的评判模型（Critic），专门负责挑刺。再复杂一点，还可以用执行结果量化指标来触发反思，比如发布失败率升高、用户点击率下降，就让 Agent 调整策略。
反思不仅仅是模型层面的“自省”，它还包括代码层面的反馈处理。工具调用失败时返回的错误信息、输入数据不合法、流程走到死胡同，这些都需要执行层捕获，并触发反思节点进入“纠错模式”。反思节点可以决定是重试当前步骤、调整计划，还是终止任务交给人工。
反思要素的一个常见误区是让模型无脑反思，每次失败都重新规划，结果不但没改善还浪费大量 token。工程上要限制反思的次数，比如最多重新规划两次，再失败就降级。反思的价值在于“知道什么时候该放弃”，而不是永不放弃。
2. 七个决策点：从 Demo 到生产级 Agent 的分叉路
2.1 决策点一：单 Agent 还是多 Agent？
七要素清楚了，接下来是工程实现里真正让人头疼的七个决策点。第一个决策点是架构层面：你的系统用单 Agent 还是多 Agent？
单 Agent 就是用一个模型实例走完感知、规划、执行、反思整个循环。多 Agent 是拆成多个角色，比如规划者、执行者、批评者，每个角色有自己的模型和上下文。选择关键看任务复杂度、成本预算和运维难度。
单 Agent 的优势是简单、便宜、状态可控。一个流程里所有信息都在这一个 Agent 的上下文里，不会出现信息不对称。缺陷是上下文容易长，单点容易“能力不足”。多 Agent 的优势是角色隔离，每个 Agent 的上下文聚焦，可以并行处理不同模块，短板是协作成本高：Agent 之间怎么通信、怎么避免重复劳动、怎么保持最终目标一致，这些都是额外要解决的问题。
我的选择原则很简单：能用单 Agent 解决的绝不引入多 Agent。只有当任务明显可以拆成多个领域，比如“先做市场分析，再写代码，最后写测试报告”，并且这些领域之间的上下文重叠很小，才值得拆多 Agent。像小红书自动发布这种线性流程，单 Agent 足够了，如果硬拆成“规划 Agent”和“执行 Agent”，反而是给自己找麻烦。
2.2 决策点二：框架选型——LangGraph、AutoGPT、自研还是 Rust/Spring AI？
第二个决策点是技术选型。现在“AI Agent 主流架构”相关的框架五花八门，但真正值得放进生产环境的并不多。
LangGraph 是目前我用的最多的，它把 Agent 建模成显式状态图，节点和边可编程、可持久化，适合复杂流程控制。AutoGPT 那种自由规划的模式看起来很酷，但生产环境里容易失控，而且 token 消耗极大，我一般只拿它做原型验证。在 Java 技术栈里，Spring AI 是个不错的选择，它把模型调用、prompt 模板、结构化输出封装得很好，适合已有 Spring Boot 的团队。在要求极致并发和低延迟的场景，有人用基于 Rust 语言写 AI Agent 的核心执行引擎，Rust 的内存安全和异步性能确实能扛高并发，但开发门槛高，生态相对没那么现成。
你需要先搞清楚自己项目的核心诉求。如果是快速做 MVP、内部工具，用 LangGraph 或类似框架；如果是高并发低延迟的服务，考虑 Rust 重写核心节点；如果是 Java 后端团队想平滑接入，Spring AI 更顺滑；如果连代码都不想写，用扣子这类低代码平台搭一个智能体应用也完全可以——但低代码平台的扩展性和私有化部署能力通常有限，真到上生产的阶段，你还是得回到手写工程。
框架不是越火越好，而是越匹配越好。团队的技术栈熟悉度往往比框架本身的能力更重要。框架的坑只有用过才知道，选一个团队能 hold 住的，比选一个看起来牛但没人会的要强得多。
2.3 决策点三：任务规划策略怎么选？
第三个决策点是规划策略。前面讲过 ReAct、Plan-and-Execute、ToT，这里说说怎么选。
先看你任务的确定性。如果任务步骤是固定的，比如“读取数据—清洗—生成报告—发送”，那就用 Plan-and-Execute，先让模型写一份计划，再按计划执行，这样每一步你都能预测，方便做中间状态持久化。如果任务需要根据中间结果不断调整，比如“帮用户答疑，需要查资料，然后根据查到的内容决定是否要追问”，那就用 ReAct，边做边看。
另外一个考虑是 token 成本。ReAct 每走一步都要把当前历史传给模型，长任务跑到后面，上下文越滚越大。Plan-and-Execute 的好处是先产生计划，计划本身不大，执行时可以只关注当前步骤，成本显著降低。我做一个长流程任务时，通常先生成计划，然后每一步单独执行并记录结果，全部执行完再做汇总。
无论选哪种，都必须设置终止条件。我给每个 Agent 任务设一个最大步骤数，比如 15 步，模型每一步都消耗时间和 token，不能让它无限循环。还要设置超时时间，一个任务最多跑 3 分钟，超时就返回超时错误。规划策略不是越复杂越好，能稳定完成任务的策略才是好策略。
2.4 决策点四：记忆怎么管，才不会越跑越蠢？
第四个决策点是记忆管理。很多 Agent 刚开始好用，越用越蠢，多半是记忆没管好。
先拆记忆类型。短期记忆对应一个任务会话内的状态，我一般放到 Redis，用 task_id 作 key，搞成一个可序列化的结构体。任务结束后设置过期时间，比如 24 小时，避免 Redis 内存被撑爆。长期记忆则放到向量库。这里有一个关键参数：embedding 模型的选择。中文场景我用text2vec-large-chinese或者 OpenAI 的text-embedding-3-small，效果和成本跨度很大，建议先在自己的数据集上跑几个候选模型看召回效果。
检索设置上，chunk 大小和 top-k 会对效果产生直接影响。我的经验是每段文本切 300 到 500 字，检索取 top 5 到 10 条，召回内容再按时间或相关性重新排序。chunk 太小容易截断语义，太大又容易混入无关内容，300-500 是一个平衡点。
记忆还要考虑一致性。写入长期记忆的内容必须是与事实一致的成功经验。例如用户明确说“我不喜欢太长的文案”，这条要写入；而模型某个时刻输出质量很差，被用户纠正了，那这个错误输出不能写进去。我遇到过 Agent 因为错误输出被写进长期记忆，导致后面每次生成都带同样错误的问题。所以写入长期记忆前，最好经过程序校验或人工确认。
2.5 决策点五：工具协议怎么设计，才能灵活扩展？
第五个决策点是工具协议。Agent 能调多少工具，不取决于你写了多少函数，而取决于你给模型提供了多少清晰、可理解、可校验的接口描述。
协议的核心是 JSON Schema。每个工具要有稳定的名称、一个人类可读的描述、一个严格的参数 Schema。举个例子，发布工具的 Schema 可以写成：
{ "name": "publish_post", "description": "发布一条内容到小红书，页面包含标题、正文和可选图片列表", "parameters": { "type": "object", "properties": { "title": { "type": "string", "minLength": 1, "maxLength": 20 }, "content": { "type": "string", "minLength": 1, "maxLength": 1000 }, "images": { "type": "array", "items": { "type": "string" }, "maxItems": 9 } }, "required": ["title", "content"] } }模型看到这个 Schema 后，会返回一个结构化调用请求，但你不能直接信它。执行层需要用 Pydantic 或其他校验库对参数做二次校验。模型经常会产生不符合要求的参数，比如 content 是数字、images 格式不对。校验失败后，把校验错误信息返回给模型，让它修正后再调用一次。这个过程我最多允许两次，两次都失败，就放弃这个工具，换个方案。
工具扩展还涉及注册机制。我习惯把工具定义集中在一个 registry 里，启动时扫描所有@tool装饰器修饰的函数，自动生成模型需要的工具描述列表。这样每新增一个工具，只要写一个普通函数并加上装饰器，模型立刻就能感知到。工具返回值也要规范化：要么是纯文本，要么是结构化的 JSON，并带上状态码和错误码。模型只有在明确知道工具调用失败原因时，才能做出正确的下一步决策。
另外，现在 MCP 协议也值得关注。它相当于工具调用的标准协议，可以跨框架共享工具。如果团队工具很多，或者要提供开放平台，建议按 MCP 的思路设计。否则，自研一个简单的 registry 也够用。
2.6 决策点六：Agent 怎么扛并发、不掉链子？
第六个决策点是最多人问的：AI Agent 怎么扛并发。这是一个综合性问题，不是加两台服务器就能解决。
先明确一个事实：Agent 的一次会话不是一次普通的 HTTP 请求。一个 Agent 任务里可能要调好几次 LLM，每次 2 到 5 秒，中间还有工具调用、状态更新。如果用户请求进来后同步执行，整个线程或协程要被占住十几秒甚至几十秒，这在 Web 服务里是不能接受的。所以正确的架构是：HTTP 请求进来，立刻创建一个任务并返回 task_id，真正耗时的工作放到后台 worker 或消息队列里异步执行。
后台执行也要控制并发度。LLM API 有速率限制，假设你的 LLM 供应商允许每分钟 10 次请求，而你有 100 个并发任务，每个任务平均调用 4 次模型，那么一分钟内就需要 400 次调用，显然会超限。我当时用 Semaphore 控制同时运行的 Agent 数量，比如同时最多 5 个任务，每个任务内部的模型调用串行，这样每分钟调用量基本稳定。
任务状态隔离是另一个并发重点。千万不要把状态放在全局变量或类变量里。每个 task 要有独立的状态对象，存到 Redis 或数据库里。多实例部署时，如果两个 worker 同时处理同一个 task_id，要先拿分布式锁，没有锁就跳过，保证幂等。
还有任务调度。一批定时任务、一批用户手动触发，如果都立刻执行可能瞬间打爆下游接口。我用 Celery 的队列和预取限制（worker_prefetch_multiplier=1）来控制。每个任务执行前还要做好请求去重：同一篇笔记不能重复发布，所以发布工具要带一个业务幂等键，比如 drafts_id，下游 API 如果发现已经处理过这个键就直接返回成功，避免重复发送。
最后要注意可伸缩性。Agent 任务的瓶颈通常在 LLM API 和外部工具，不一定是 CPU 或内存。所以扩容不一定是加机器，也可以做请求排队、调整并发度、对不同优先级任务设置不同队列。上线前要做压测，重点看 LLM API 的调用频次和延迟分布。
2.7 决策点七：可观测性和安全护栏怎么做？
第七个决策点是可观测性和安全。Agent 是一个不可完全确定性的系统，如果看不到每一步的输入输出，出问题根本没法排查。
可观测性要从三个层面做。第一层是日志：每个节点执行前和执行后都记录一条日志，包含 task_id、节点名、模型调用时间、token 数、工具调用参数和返回值摘要。我用 structlog 做结构化日志，方便后续检索。第二层是链路追踪：一次 Agent 任务就是一个 trace，里面串起多次模型调用和工具调用，用 Langfuse 或自建一张 trace 表都能实现。第三层是指标监控：任务成功率、平均耗时、工具调用失败率、token 消耗量，这些指标要上监控大盘，设置告警，比如成功率低于 80% 就报警。
安全护栏是生产级 Agent 的底线。输入侧要防 prompt injection，尤其是工具返回的内容可能携带着来自第三方数据的恶意指令。输出侧要加内容审核，比如小红书发布前必须过一遍敏感词和图片合规检查。权限控制要做到最小粒度，Agent 只能调用当前任务需要的工具，而不是所有工具都能调。
特别提醒一下，任何涉及真实资金、法律效力、公开传播的高风险操作，都不要让 Agent 全自动执行。比如个人使用 AI Agent 做期货交易，技术上确实可以对接行情 API 和交易接口，让 Agent 根据策略自动下单，但这种全自动无人值守的模式风险极高，我强烈不建议。金融交易真金白银，波动和不确定性太大，Agent 的一次误判可能造成无法挽回的损失。如果一定要尝试，也只能设计成“Agent 负责分析和给出建议，人工审批后才执行交易”的人机协同模式，同时设置仓位上限和强制熔断机制。这不是技术问题，这是责任和安全问题。
3. 一个完整案例：用 FastAPI + LangGraph 实现“自动发小红书”的 Agent
3.1 需求拆解和架构
说了这么多理论，我用一个最近做的“让 AI Agent 自动发小红书”的例子把上面的内容串一遍。这个需求很简单：每天早上读取草稿箱里待发布的笔记，让 Agent 根据当前热点生成标题和正文，配上图片，经过人工审核后发布。
整个系统我用 FastAPI 做 HTTP 接口，LangGraph 做 Agent 状态机，Celery 做异步任务队列，Redis 存状态，PostgreSQL 存结果数据。FastAPI 的异步性能和生态都很好，是我做这类服务的主力框架。有人问为什么不用 Django，Django 也可以，但如果你让 AI Agent 开发 Django 项目，会遇到很多同步阻塞的问题，FastAPI 和 Agent 这种异步 I/O 密集型场景更搭配。
架构流程是这样的：外部触发器（可以是定时任务，也可以是一个 HTTP 请求）调用 FastAPI 接口，接口把任务参数（比如草稿 ID 列表）写入 Celery 队列并立即返回 task_id。Celery worker 拿到任务后，把 task_id 作为输入创建一个 LangGraph 状态图实例，然后执行各个节点。节点包括：读取草稿、生成文案、生成配图、合规审核、请求人工确认、正式发布。每个节点的输入输出都写入状态对象，并同步持久化到 Redis。
3.2 核心代码：状态定义、节点和工具
先定义 Agent 的状态结构。LangGraph 用 TypedDict 定义状态，每个节点都可以读写这些字段：
from typing import TypedDict, Optional class PublishState(TypedDict): task_id: str draft_id: str draft_content: str generated_title: Optional[str] generated_content: Optional[str] image_urls: list[str] review_status: str # pending / approved / rejected publish_result: Optional[str] retry_count: int然后定义两个核心工具：一个是读取草稿，另一个是生成发布内容。草稿工具从数据库里按状态取数据：
@tool def fetch_draft(draft_id: str) -> dict: """从草稿表读取一条待发布笔记内容""" row = db.query(Draft).filter(Draft.id == draft_id).first() if not row: return {"ok": False, "error": "draft not found"} return {"ok": True, "content": row.content, "topic": row.topic}生成工具不直接调发布接口，而是把生成结果保存到审核表：
@tool def save_review_record(draft_id: str, title: str, content: str, images: list[str]) -> str: """保存待审核发布内容，返回审核记录ID""" record = ReviewRecord(draft_id=draft_id, title=title, content=content, images=images) db.add(record) db.commit() return record.idLangGraph 的状态图这样构建：
from langgraph.graph import StateGraph graph = StateGraph(PublishState) graph.add_node("fetch_draft", fetch_draft_node) graph.add_node("generate_copy", generate_copy_node) graph.add_node("build_images", build_images_node) graph.add_node("review_approval", review_approval_node) graph.add_node("publish_release", publish_release_node) graph.add_edge("fetch_draft", "generate_copy") graph.add_edge("generate_copy", "build_images") graph.add_edge("build_images", "review_approval") graph.add_edge("review_approval", "publish_release") graph.add_conditional_edges( "review_approval", should_retry, {"retry": "generate_copy", "approve": "publish_release", "reject": "__end__"} ) app = graph.compile()这里的关键是should_retry函数，如果审核不通过且重试次数没超，就回到生成节点重新生成，最多重试 3 次。审核节点实际上做的是“人工确认”，我在这里直接阻塞等待人工接口回调，而不是让模型自己判断能不能发。发布操作也只是一个“调用平台发布 API”的普通工具，但它只在review_status为 approved 时才会被调用，这样从流程上杜绝了未经审核直接发布的可能。
3.3 部署与调度：让它真正跑起来
部署方面，我用 Docker Compose 编排了四个服务：fastapi 应用、celery worker、redis、postgres。FastAPI 提供两个接口：一个提交任务，一个查询任务进度。提交任务时生成 UUID 作为 task_id，放入 Celery queue；查询任务时读 Redis 里的状态对象，返回给前端。前端可以轮询这个接口，也可以用 WebSocket 推送。我用的是轮询，简单可靠。
定时调度用 Celery Beat，每天早上 9 点触发一次批量任务。因为小红书平台对发布频率有控制，同一账号一天不要发太多，所以我限制每次任务最多取 5 条草稿，并且每条草稿发布间隔不小于 30 分钟。这个限制写在调度配置里，Agent 自己是不能绕过的。上线后我会观察每天的发布成功率，如果连续两天低于 90%，就检查是文案质量问题还是接口限流问题。
在并发控制上，Celery worker 加--concurrency=4参数，同时只跑 4 个 Agent 任务。另外我对每个任务内部加了信号量，限制同时调用生成模型不超过 2 个并发，避免瞬间打爆模型 API。每个任务的状态都独立存在 Redis，互不干扰。
4. 常见问题与避坑实录：我在工程实现中踩过的坑
4.1 上下文被撑爆，Agent 变“失忆”
我最早做 Agent 时，为了给模型更多信息，把用户对话历史、工具返回的完整 JSON 全塞进上下文。结果跑了几轮之后，模型开始忽略早期的指令，回答质量直线下降。这就是上下文窗口被无关信息占满导致的“中间遗忘”。
解决方法是给上下文做分层：核心系统指令永远在最前面，当前任务状态其次，历史对话经过摘要压缩后放在最后。工具返回的内容不能全文保留，只把关键字段抽取出来存入状态。比如一个查询接口返回了几百条记录，我会让模型先做一次“信息抽取”，把需要的数据整理成表格或摘要，然后丢弃原始 JSON。我用一个固定的 Prompts 模板控制不同层级的占比，默认系统指令约占 10%，状态信息占 30%，历史摘要占 20%，剩余留给新内容。这样整个任务跑下来，上下文长度基本可控。
4.2 工具参数幻觉，模型编造调用
大模型在生成工具调用参数时要么多传、要么少传、要么传错类型。我遇到过一个案例，模型调用一个send_message工具，把user_id参数填成了None，然后系统误发给默认用户，事后排查特别麻烦。
从那以后，所有工具入口都加了严格的参数校验。我用 Pydantic 定义参数模型，校验不通过就返回一个标准化错误消息给模型，让模型重新生成。同时限制工具重试次数为 2 次，如果模型连续两次生成非法参数，就放弃这个工具调用，改为走“降级路径”。降级路径可以是问用户、跳过该步骤，或者用另一组更简单的工具完成目标。
工具返回值的格式同样要规范。我要求所有工具返回一个统一的字典：{"ok": bool, "data": ..., "error": ...}。这样模型能快速判断是否成功。如果返回的是纯文本，模型也能用，但解析效果会差很多。工具设计的规范性直接影响 Agent 的稳定性，这比模型能力还重要。
4.3 并发状态下数据串线，状态污染
当时我为了提高并发，把 Agent 的执行状态放在一个全局字典里，键是 task_id。本地测试没问题，一上测试环境，多个任务同时跑，偶尔出现 A 任务的数据串到 B 任务结果里。查了半天，发现是全局字典被多个协程同时读写导致的数据竞争。
解决方法是把状态外置到 Redis，每个 task_id 独立 key，读写都用 Redis 的原子操作。每次节点执行前从 Redis 拉取状态，执行后再写回。后来协程之间不再共享内存，串线问题彻底消失。这里有个细节：如果你用 LangGraph，State 对象本身必须是可以序列化的，不能在里面放模型对象、数据库连接等不可序列化的东西，否则存不进 Redis。
还有一个易踩的坑是回调。异步任务里如果调用了外部 API，而外部 API 的回调地址配置错误，可能导致另一个任务被唤醒。我会在回调事件里带上完整的 task_id 和签名校验，处理时才确认这个 task_id 是当前 worker 正在处理的，否则直接忽略。
4.4 Agent 评测怎么做？不能靠“感觉”
Agent 的随机性导致你很难判断“改了一版 prompt 到底有没有变好”。最开始我靠肉眼测几条样本，感觉效果不错就上线，结果线上反馈完全不一样。后来我建了一个回归评测集，里面固定了大约 20 个典型任务，包括正常任务、边界任务、恶意输入。
每个评测任务我会检查两个层面的东西。第一是“过程正确性”：Agent 是否按预期调用了工具、工具参数是否合法、是否在重试次数内完成；第二是“结果可用性”：生成的内容是否满足要求、有没有明显错误。跑完评测集，计算成功率，只有成功率不低于历史基线才允许上线。这个流程虽然简单，但确实帮我挡掉了很多“感觉不错但实际有问题”的改动。
长期来看，我还在收集线上失败的日志，定期补充到评测集里。比如我遇到过平台接口改版导致某类草稿发布失败，后来把类似案例加入评测集，每次调工具参数都会跑一遍，防止 Regression。
说实话，Agent 工程的复杂度比传统后端高一个量级。传统接口只要保证输入输出稳定，Agent 还要面对模型的不确定性、工具的不可靠性、状态的持续演化。但正因如此，当你把七要素理清、七个决策点走完，你的 Agent 就不再是一个所有脚本的缝合怪，而是一个真正有边界的、可观测的、能稳定干活的系统。我自己最开始也是从“写个 prompt 套个 LangChain”开始，踩了无数并发和状态坑，才一步步总结出这套清单。
最后再分享一个小技巧：新项目从零开始搭 Agent 时，不要一上来就铺大架构。先按七要素列一个最简清单：目标写清楚、感知只接一条数据、记忆先用 Redis 字符串、规划写死直线流程、工具只做一个、执行用同步函数、反思只做一次重试。跑通这样一条最简链路后，再迭代着补上并发、持久化和安全。每次加一个决策点，都要先明确“为什么加”，而不是“别人都有所以我也加”。这比什么都重要。
