在真正选 AI Agent 架构之前我先问你一个特别基础的问题你现在的“Agent”到底是封装了一次大模型 API 调用的脚本还是一个能承载完整决策循环的执行运行时我见过太多团队把这两件事混为一谈。产品经理说“我们要做个 Agent”后端同学就开始对比各种框架折腾了一周发现代码还是停留在拼 prompt、调模型、解析输出、再拼 prompt。这就是典型的“先选架构、后抠本质”的反向操作。Agent 架构选型的真正起点是先弄清楚你的应用正处在那条演进路径的哪一级台阶上是单次 LLM 调用、可循环的 Agent 闭环还是需要并发、记忆、工具沙箱、可观测性共同作用的 Harness Runtime。这篇文章我不想给你一张框架对比表就完事。我会从一次最普通的 LLM 调用开始逐步拆到完整 Harness Runtime 的各个组件然后给出我在实际项目中反复用的一套选型判断框架以及一个可以直接拿去改的最小运行骨架。1. 从一次 LLM 调用说起Agent 的最小原子单元1.1 单次调用到底包含什么一次 LLM 调用并没有多神秘它本质上就是“输入一段文本输出一段文本”。放在工程视角下你只需要关心四个参数模型标识、system prompt、user message、以及采样参数temperature 之类。返回值通常也就三样生成的文本内容、token 消耗统计、以及结束原因正常结束、长度截断、内容过滤触发等。但如果你只是把这个过程封装成一个chat()函数那它只是一个“远程函数”还不是 Agent。我见过很多项目所谓的“Agent”本质上就是这样一个远程函数外面套了一层业务逻辑问天气就调天气 API问数据库就调查询接口。这种实现没有状态、没有工具决策、没有循环模型一旦答非所问整个流程就断了。这里我想强调一个容易被忽略的点单次调用的核心价值不在输出文本而在可控性。你需要管住输出格式、超时时间、重试策略、token 上限。很多团队在生产环境里踩的第一个坑就是模型偶尔返回了一段超长文本直接把下游解析系统打崩或者某个请求卡住HTTP 连接被占满后续调用全部排队。所以从第一次调模型开始就要把“带超时、带重试、带结构化输出约束”的封装当成默认选项。1.2 从原子调用到函数调用Function/Tool Calling真正让 Agent 和普通聊天机器人分道扬镳的是函数调用能力。模型不再只是“输出文本给你看”而是可以输出一个结构化的动作指令比如{ function: search_order, arguments: { order_id: A12345 } }这一步的意义很关键模型从“内容生成器”变成了“决策器”。你的程序负责执行动作模型负责根据当前情况决定动作。执行完之后把调用结果作为新消息再次发送给模型模型再决定下一步动作。这个循环一旦跑起来你才拥有了一个真正意义上的 Agent 雏形。但也是从这里开始架构问题变得复杂。函数定义写在哪里工具结果回传时体积多大要不要截断模型连续调用失败怎么办循环次数上限是多少这些问题一个接一个冒出来单靠“封装一次调用”是解决不了的。1.3 你真正需要的是“调用抽象层”面对多家模型供应商、多种模型版本你还需要一个调用抽象层。我一般这样做定义一个统一的ChatModel接口包含chat(),chat_with_tools(),stream_chat()三个方法然后把不同厂商的 SDK 适配进去。class ChatModel(ABC): abstractmethod def chat(self, messages: list[dict], tools: list[dict] | None None, temperature: float 0.2, max_tokens: int 2048) - ChatResult: ...这个抽象层不是为了炫技而是为了在选型时保留退路。你可能会因为价格、延迟、效果切换模型也可能会因为业务需要同时跑多个模型做评测。没有这层抽象到时候你会被供应商 SDK 死死绑住。不过我也要泼一盆冷水抽象层只解决“模型调用”这一个环节它解决不了 Agent 循环、状态和工具管理问题。换句话说它只是完整 Harness Runtime 的第一块砖。2. 从单次调用到 Agent 闭环循环是核心2.1 ReAct 模式的本质Agent 闭环最经典的实现是 ReAct推理Reason然后行动Act。模型先表达自己的思考可以不放给用户看然后选择一个工具调用系统执行工具把结果回传模型根据新情况继续推理。这个循环持续到模型认为任务完成。我第一次上手实现这个循环时以为难点在“让模型思考”后来发现难点在“循环的工程纪律”。你需要给循环设置硬边界最大迭代次数我通常设置 5~10 次防止模型陷入无限循环。单次工具调用的超时时间比如 15 秒。累积 token 预算防止对话上下文无限制膨胀。当模型输出非法 JSON 时的重试策略。没有这些边界再聪明的模型也会在复杂的上下文里钻牛角尖。你肯定不想看到一个查询任务因为模型反复调同一个工具而跑了 20 分钟。2.2 工具的定义与容错工具是 Agent 和真实世界交互的桥梁。在设计工具列表时我建议每个工具都提供三段信息工具描述、输入参数 JSON Schema、执行代码。描述尤其重要因为模型依靠描述决定何时调用它。描述写得含糊模型就会乱调参数定义不严格模型就会传错类型。我的习惯是给工具描述加上“什么时候用”“什么时候不要用”两个字段。例如搜索订单信息当用户提供订单号时使用。不要用于搜索商品列表商品列表请调用 search_products。工具执行出错是常态。不要一错就让整个 Agent 失败而是把错误信息变成一条“观察结果”回传给模型给它自我纠正的机会。比如result run_tool(tool_name, arguments) if result.is_error: observation f工具执行失败{result.error_message} else: observation f工具执行结果{result.data}让模型看到错误、自己决定下一步是提升 Agent 鲁棒性成本最低的手段。2.3 记忆不只是把聊天记录塞进去很多教程喜欢把记忆简单等同于“把之前的对话消息拼起来再发给模型”。放在 demo 里没问题放在生产里会出事。记忆要考虑三类短期记忆当前任务窗口内的对话与工具结果受上下文长度限制。长期记忆跨会话保存用户偏好、历史事实需要向量检索或结构化存储。工作记忆正在执行的计划、已经完成的步骤、尚未完成的子任务。我的建议是先把短期记忆做扎实再上长期记忆。短期记忆的常见问题是“上下文爆炸”因为每次工具调用的输出都会追加到 messages 里。解决办法有两个一是截断工具结果只保留摘要二是累积 token 数超限后由模型对历史做一次总结压缩。3. Agent 架构的三层演进脚本、循环、运行时3.1 L0脚本编排不是贬义词这里我想先说一个反直觉的观点并不是所有场景都需要 Agent 循环。如果你的业务流程是固定的“查订单 - 计算退款 - 通知用户”那用代码顺序调用比让模型自由决策更稳定、更便宜、延迟更低。L0 是脚本编排业务步骤固定模型只在关键节点被调用一次或者几次。这种架构没有决策循环但它是很多稳定业务的最优解。我做过一个内部工单分类系统就是典型的 L0用户提交工单模型做一次分类和优先级判断其余流程靠代码串联。上线后效果很稳成本可控问题定位也快。3.2 L1单 Agent 决策循环当任务步骤不能被预先枚举时你就得升级到 L1一个模型循环配合动态工具选择。比如“根据用户描述生成一份竞品分析报告”你需要搜索、浏览、总结但具体步骤取决于用户到底关注什么无法提前写死。L1 就是前一章讲的 ReAct 循环。从 L0 到 L1 的升级最容易犯的错误是“路径失控”。模型每一步都是对的但整体策略可能跑偏。我的解决办法是引入“任务计划”阶段进入循环前先让模型输出一份简短的执行计划然后每一步都对照计划执行。这样即使局部跑偏也能通过人工检查计划快速发现。3.3 L2多 Agent 与 Harness RuntimeL2 阶段你开始面对多个 Agent、多个模型、多个工具的协同。这时候你会遇到一批全新问题A Agent 的输出怎么给 B Agent多个 Agent 是否共享状态工具执行在哪里跑失败重试是在 Agent 内部还是 Runtime 层面安全策略怎么统一管控这些问题已经超出了“循环”的范畴它们属于运行时Runtime的职责。我习惯把这一层称为 Harness Runtime——专门承载 Agent 执行的支撑系统。它包含调度、状态、记忆、工具沙箱、可观测性、安全策略等模块。选型到这一层你才真正需要认真思考“架构怎么选”。这里我特别强调一句不是每个团队都应该直接干到 L2。如果你的用户量不大、任务类型单一L1 就是成本和效果的最佳平衡点。很多团队在 L1 没跑稳的情况下强行上 L2最后被分布式状态同步问题拖垮。4. 完整 Harness Runtime 的组件拆解4.1 调度器谁来决定下一个动作调度器是 Runtime 的核心入口负责决定一个任务接下来该执行什么。单 Agent 场景下调度器就是 Loop多 Agent 场景下调度器还需要决定把任务交给哪个 Agent回收结果以及处理并行和依赖关系。我在实际项目里很少一开始就上复杂的调度框架。通常用一个任务队列 状态机的组合就够了每个 Agent 执行单元持有“任务状态”和“上下文引用”调度器按状态推进。只有当你需要动态编排比如根据用户输入临时创建新的子任务时才需要考虑引入更复杂的图执行引擎。4.2 状态存储Agent 的工作台Agent 执行过程中必须有人记着“当前进行到哪一步了”这就是状态存储。状态包括任务元信息、执行计划、历史消息索引、工具执行记录等。最简单的方式是纯内存 Redis 持久化进程内跑任务时用内存字典关键节点异步写入 Redis。这里我不建议把所有消息和历史记录都塞进 Redis只存关键状态完整日志交给可观测性系统。否则数据量和读写放大问题很快会让你头疼。状态管理的经典错误是“多副本状态不一致”。如果你部署了多个 Worker 进程一定要把“状态更新”设计成幂等操作。我碰到过一个真实事故两个 Worker 同时消费了同一队列里的任务各自执行了一遍工具调用最后订单被重复修改。根因就是任务消费时缺少分布式锁或幂等键。4.3 工具沙箱不要让 Agent 裸奔当 Agent 拥有了调用外部系统的能力安全问题就浮出水面。我坚持一个原则工具的权限必须小于等于用户的权限且所有工具调用都要有审计记录。沙箱分几个等级完全开放仅用于本地开发不用于生产。网络受限只能访问白名单域名。只读优先涉及写操作的工具一律二次确认。参数校验所有工具入参必须经过 JSON Schema 校验过滤危险字段。比如一个“发送邮件”工具绝不能允许 Agent 直接传任意收件人和内容必须在工具内部强制校验收件人域名、发送频率、内容长度。这不是防模型是防意外。模型本身没有恶意但它可能被构造出来的 prompt 诱导去做危险操作。4.4 可观测性没有日志就别谈调优Agent 应用的调试难度比传统后端高一个量级因为每次执行链路都不同。我强烈建议从第一天就记录三类日志决策轨迹模型每一步的输入输出、工具轨迹调用参数和返回结果、成本轨迹每一步的 token 消耗和延迟。决策轨迹尤其重要。我通常把模型的“思考过程”“工具选择依据”和“最终结论”分开记录方便事后复盘。线上排查问题时我几乎都是先拉起最近一次任务的完整轨迹看模型在哪一步做出了多余调用或者哪一步工具结果异常然后再决定是调 prompt 还是调工具逻辑。{ trace_id: task_001, steps: [ {action: thought, content: 用户需要查询订单先调用搜索接口}, {action: tool_call, tool: search_order, arguments: {order_id: A12345}}, {action: tool_result, success: true, summary: 找到1条订单金额99.0元}, {action: final_answer, content: 你的订单金额是99元。} ], total_tokens: 1823 }4.5 安全策略与重试机制Harness Runtime 还应该内置统一的重试、限流、降级策略。模型供应商接口偶尔会返回 5xx、超时、限流你需要的是在 Runtime 层统一处理而不是在每个 Agent 里重复写。我的策略是普通调用重试 2 次指数退避工具调用如果是幂等的可以重试非幂等的不重试限流时优先丢弃低优先级任务。成本控制方面要给每个任务设置预算上限。一旦任务累计 token 超过阈值Runtime 应强制让模型收敛或者降级到更小的模型。5. 架构选型的判断标准别被框架带着走5.1 六个选型维度我平时判断一个项目该用哪种 Agent 架构基本只看六个维度。把它们做成一个表会直观很多维度判断内容偏 L0/Script偏 L1/Loop偏 L2/Runtime任务确定性步骤是否可提前枚举高低不可枚举且跨域延迟要求用户能等待多久短5s可接受 5~30s分钟级以上成本敏感度对单次任务 token 费用很敏感一般可接受较高预算可解释性是否需要复盘每一步强较强必须全链路追踪并发规模同时执行任务数高中高且需要调度团队维护力是否有专职支撑人员小中必须专职每个项目都拿这张表过一遍答案基本就出来了。注意这几个维度之间不是等权重的团队维护力经常是一票否决项——没有专人维护就不要选最重的架构。5.2 三类典型场景的选型示例场景一客服会话摘要。步骤固定延迟要求中等token 成本敏感。选 L0用一次模型调用加几个后处理规则性价比最高。场景二企业内部智能运维助手。需要查监控、查日志、执行变更多步骤且依赖上一步结果。选 L1 单 Agent 循环配上严格工具白名单和人工审批节点。场景三复杂业务数据处理流水线比如从非结构化文档中抽取数据并入库。需要多个专用模型协作且涉及长文档、多轮校验。这类才值得上 L2 Harness Runtime。5.3 自研还是用现成框架几乎每个团队都会问我这个问题。我的态度很明确Demo 或 MVP 阶段直接用现成框架快速验证业务逻辑进入生产阶段大概率要自研关键路径上的组件。原因是现成框架帮你解决了通用问题但你真正的业务价值往往在特定工具、特定状态机、特定安全策略上。框架抽象越厚重你把业务逻辑嵌进去就越难。我曾把一个早期跑在通用框架上的 Agent 重构到自研 Runtime耗时三周但重构后调试效率提升了至少一倍因为每一步决策日志都是我能直接理解的业务语言而不是框架内部的抽象对象。6. 实操落地一个最小 Harness Runtime 骨架6.1 核心数据结构和 Agent 循环纸上谈兵聊到这里直接上一个可运行的最小骨架。我要实现的是一个 L1 级别的单 Agent 循环但结构上预留了升级 L2 的接口。核心只有四个类AgentExecutor调度循环。ToolRegistry工具注册与执行。Context状态容器。TraceLogger决策轨迹记录。先看数据结构dataclass class Step: step_id: int action: str # thought / tool_call / tool_result / final content: str meta: dict field(default_factorydict) dataclass class AgentContext: task_id: str user_id: str messages: list[dict] steps: list[Step] memory: dict # 简化版工作记忆 token_used: int 0 max_tokens: int 10000然后是主循环class AgentExecutor: def __init__(self, model: ChatModel, registry: ToolRegistry, max_steps: int 8): self.model model self.registry registry self.max_steps max_steps def run(self, task: str, context: AgentContext) - str: context.messages.append({role: user, content: task}) for step_id in range(self.max_steps): response self.model.chat( messagescontext.messages, toolsself.registry.schemas() ) context.token_used response.usage.total_tokens if self._exceeded_budget(context): return self._force_finish(context, token budget exceeded) if response.tool_calls: for call in response.tool_calls: context.steps.append(Step(step_id, tool_call, f{call.name} {call.arguments})) obs self.registry.execute(call) context.messages.append({ role: tool, tool_call_id: call.id, content: obs.summary }) context.steps.append(Step(step_id, tool_result, obs.summary)) else: answer response.content context.steps.append(Step(step_id, final, answer)) return answer return self._force_finish(context, max steps reached)这个循环虽然简单但已经具备 Agent 的核心特征工具动态选择、结果回传、步骤上限保护、token 预算保护。6.2 工具注册与参数校验工具注册器的关键是对外暴露给模型的 Schema 和执行逻辑分开。Schema 描述模型怎么调用执行逻辑决定系统怎么执行class ToolRegistry: def __init__(self): self._tools: dict[str, tuple[dict, Callable, JSONSchema]] {} def register(self, name: str, description: str, schema: dict, fn: Callable): self._tools[name] ({type: function, function: { name: name, description: description, parameters: schema, }}, fn, schema) def schemas(self) - list[dict]: return [meta[0] for meta in self._tools.values()] def execute(self, call) - ToolObservation: meta, fn, schema self._tools.get(call.name) if not meta: return ToolObservation(successFalse, summaryf未知工具: {call.name}) try: validated_args validate(call.arguments, schema) result fn(**validated_args) return ToolObservation(successTrue, summarytruncate(str(result), 500)) except Exception as e: return ToolObservation(successFalse, summaryf工具异常: {type(e).__name__}: {e})两个细节值得说明。第一工具结果一定要截断我从 2000 字符压到 500 字符模型照样能理解token 省一半以上。第二参数校验必须发生在执行之前否则模型传错一个字段工具可能产生脏数据。6.3 记忆压缩的简单实现当消息列表太长时可以用一次“总结式调用”压缩旧消息。我的做法是设定一个阈值比如上下文超过 6000 token 时把最旧的一部分消息交给模型压缩成一段摘要def compress_messages(model: ChatModel, messages: list[dict], threshold: int) - list[dict]: total estimate_tokens(messages) if total threshold: return messages head messages[:15] rest messages[15:] summary model.chat(messages[ {role: system, content: 压缩以下对话为摘要保留关键事实与未完成任务。}, {role: user, content: json.dumps(head, ensure_asciiFalse)} ]) return [{role: system, content: f历史摘要: {summary.content}}] rest这个方案有代价压缩本身也要消耗 token而且摘要可能丢失细节。实用做法是把摘要任务放在后台异步执行并在摘要中显式标记“未完成事项”避免 Agent 断点续做时忘记关键步骤。6.4 并发控制与部署注意L1 骨架在生产跑起来后第一个敌人就是并发。模型 API 有 QPS/TPM 限制你的工具也可能有。所以我建议在 Runtime 外部加一层“并发闸门”信号量控制同时执行的任务数token 桶控制 API 请求速率。from asyncio import Semaphore class ExecutionGate: def __init__(self, max_parallel: int, rate_limit: float): self.sem Semaphore(max_parallel) ...更关键的是任务隔离每个 Agent 任务必须有独立 Context不能在多个任务间共享同一个消息数组。新手最容易犯的错就是为了省事把 Context 设计成了模块级全局变量。并发一上来两个任务互相覆盖消息排错能排到崩溃。7. 常见问题与排查技巧实录7.1 模型陷入工具死循环现象Agent 反复调用同一个工具比如一直查询订单列表就是不进入下一步。排查方法先看决策轨迹确认模型每次 query 的条件有没有变化。如果工具返回结果一直相同说明模型缺少“去重”的依据。解决办法在消息里增加“该结果已存在请勿重复查询”的提示或者设置工具调用去重哈希相同参数不重复执行直接返回上次结果。7.2 工具结果太长导致上下文爆炸现象一次检索返回上万字符Agent 瞬间超出 token 预算。排查方法看 TraceLogger 中每条 tool_result 的 summary 长度。治理优先级先截断、再摘要、最后向量检索只拿 TopK。我现在的经验法则是能提字段名就别提全表能拿 10 条结果就别拿 100 条。7.3 并发下状态串号现象任务 A 的步骤日志出现在任务 B 里或者两个任务的最终答案对调。排查方法检查 Context 是否有全局复用。检查异步任务是否存在共享内存的字典。根治办法是每个任务显式传递 task_id所有持久化 key 都必须带 task_id 前缀。这类问题排查最花时间所以我强烈建议在框架层面禁止全局状态强制每个模块从 Context 取数据。7.4 token 成本失控现象一个简单任务消耗了远超预期的 token账单飘红。排查方法先看 token 消耗轨迹找出大头在哪一步。常见原因工具结果未截断、历史消息未压缩、模型输出过长。另一个被忽略的点工具 Schema 本身也会消耗 token。工具定义太长模型每次调用都要重复输入。把不太常用的工具从主 Schema 移到“二阶段调用”可以显著省 token。7.5 一个快速自查表症状最可能原因优先检查项Agent 答非所问system prompt 与工具定义冲突决策轨迹、一轮的 thought循环不终止缺少 max_steps 或终止条件模糊循环逻辑、模型提示词工具调用频繁失败参数 Schema 与真实函数不匹配校验返回的 error message并发任务错乱共享 Context状态存储键是否带 task_id延迟高工具调用串行等待工具结果是否要同步返回7.6 调试 Agent 的独家技巧最后分享几个小技巧。第一给模型加一个“隐身调试开关”在 system prompt 里附加一段“请在每个 thought 前输出 标签上线时再关闭”。这样可以低成本拿到模型内部推理过程而不需要额外日志系统。第二做“最小复现用例”特别高效把线上失败任务的 messages 原样保存离线重放不断改 prompt 和代码验证。能做到离线重放的 Agent 系统半年后你会感谢当初的自己。第三对工具调用加一条“请求ID”贯穿前后端你在日志平台搜请求ID就能串起模型决策和真实业务操作排查效率翻倍。我在实际项目中体会最深的是Agent 架构选型没有银弹只有取舍。唯一能保证的规律是越复杂的架构越需要配套的工程治理能力。如果你现在的团队连基础的可观测性日志都还没有建立先把 L0/L1 做好等到真的需要调度器、多 Agent 协同和动态编排的时候再升级到完整 Harness Runtime 也不迟。这个演进路径比我见过的任何一次性大改造都更稳。