AI Agent实战：从ReAct循环到工具调用的工程实现

最近总有人问我：AI Agent到底和普通的"套壳聊天"有什么区别？我通常不直接回答，而是反问一句——你做的那个东西，模型能自己去调用天气接口、查数据库、把算好的结果再回给你吗？如果答案是不能，那不管宣传上怎么写，它都还算不上Agent，顶多是个带对话界面的查询工具。真正意义上的AI Agent，是一个把大模型当大脑、把外部API当手脚的闭环决策系统。

这几个月我陆续做了几个不同用途的Agent项目，从模拟项目X到某个跨平台系统，从最开始只会调模型接口，到后来能跑通"规划—调用—反馈—纠错"的完整循环，踩了不少坑，也沉淀出一套比较务实的架构和算法设计。这篇文章把核心的东西一次讲透：架构怎么搭、算法怎么选、源代码怎么组织。标题里的"In Action"我理解得很实在——不是概念科普，是能跑、能部署、能在真实任务里出结果的工程方案。

1. Agent与普通LLM应用的分水岭：从"回答问题"到"完成任务"

1.1 套壳对话和真正的Agent差距在哪里

普通的大模型应用，本质是一个"文本进、文本出"的映射。你问它"明天某城市最高温度多少"，它能凭训练数据里的知识给你一个模糊回答，但它不会真的去查实时数据。即便你把实时数据塞进Prompt里，它也只是在"阅读理解"——读一段文字，然后总结。

Agent不一样。Agent是目标导向的：你给它一个任务，它自己决定需要调用什么工具、按什么顺序调用、如何解读工具返回的结果，直到任务完成或者确认失败。同样是"明天某城市温度"这个需求，Agent会自己去调用天气查询工具，拿到JSON数据后判断"最高温度低于20度，需要提醒带外套"，然后继续执行下一步——比如把结论写入笔记。整个过程里的每一步，模型都在参与决策，但每一步的执行动作都落在真实的工具调用上。

我在项目里最直观的感受是：普通LLM应用是"问答式"的，Agent是"行动式"的。问答式关心的是生成质量，行动式关心的是任务完成率。评价指标不一样，技术栈自然也不一样。

1.2 一个可工作的Agent必须过三道关卡

第一道关卡是意图与目标拆解。用户说"帮我看看明天的天气，合适的话提醒我带外套，顺便记到笔记里"，这句模糊的需求要能被拆成"查询天气—判断温度—决定是否提醒—写入笔记"这样一串可执行的动作。很多初版Agent就死在这一步：模型拿到一句话，直接凭感觉回答，没有生成行动计划。

第二道关卡是工具选择与参数调用。模型要从注册表里挑出"天气查询"这个工具，并且把参数填对——城市名是哪个、日期是明天还是后天、温度单位是什么。这一步看着简单，实际是幻觉重灾区：模型经常编出一个不存在的城市名，或者把日期格式写成"明天"而不是"2025-01-15"。要让模型稳定地填对参数，工具定义和参数Schema必须写得极其明确。

第三道关卡是状态维持与自我纠错。工具返回的结果可能和预期不符：查不到数据、权限失败、返回格式变了。一个合格的Agent要在这一步把错误信息当作新的Observation喂回给模型，让它自己换一种方式重试，而不是直接崩溃或者硬着头皮往下走。

这三道关卡全部打通，才算是一个最小可用Agent。下面我按架构、算法、代码三个层面依次展开。

2. 架构设计：大脑、手脚、记忆三层模型的取舍

2.1 为什么把Agent拆成三个独立模块

我给Agent做架构时，参考了人脑的分工方式：大脑负责思考，手脚负责执行，记忆负责存储和调取信息。对应到代码里，就是三块东西：

• 大脑（LLM调度器）：负责读任务、生成行动计划、分析工具返回结果、决定下一步动作。
• 手脚（工具执行器）：负责真正的外部调用——HTTP请求、数据库查询、文件读写、发消息等等。
• 记忆（上下文管理器）：负责管理对话历史、任务中间状态、关键信息摘要。

这三个部分必须拆开，因为它们的失败模式和迭代节奏完全不同。换一个更强的大模型，只需要改大脑这一层；增加一个新的外部系统对接，只需要改工具层；发现上下文太长导致效果变差，只需要调记忆层的策略。如果把它们耦合在一个大函数里，每次改动都会牵一发动全身。

另一个原因是观测的粒度。拆开之后，每一步都可以单独打日志：大脑输出了什么决策、工具执行耗时多久、记忆层压缩掉了哪些内容。出问题的时候，能直接定位到是哪一层出了问题，而不是在一大坨逻辑里猜。

2.2 工具注册表：给Agent一份"能力清单"

工具注册表是整个Agent架构的心脏。它的作用是把Agent能做的所有外部操作，抽象成一份结构化的能力清单，让模型"知道"自己有什么工具可用，以及每个工具怎么用。

我用的核心数据结构长这样：

@dataclass class Tool: name: str # 工具名，Agent根据这个名字调用 description: str # 一句话说明，告诉Agent这个工具能做什么 parameters_schema: dict # 参数JSON Schema，描述每个参数的类型和约束 function: callable # 实际执行的函数

这里最关键的是parameters_schema。不用schema，模型就不知道参数该填什么、填什么类型、哪些是必填。用了schema之后，模型在生成调用时就有了明确的格式约束。

举个例子，一个天气查询工具的注册信息是这样的：

weather_tool = Tool( name="query_weather", description="查询某个城市未来N天的天气，返回最高温度、最低温度、天气状况", parameters_schema={ "type": "object", "properties": { "city": {"type": "string", "enum": ["北京", "上海", "广州", "深圳", "杭州"]}, "date": {"type": "string", "description": "日期，格式YYYY-MM-DD"}, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"], "default": "celsius"} }, "required": ["city", "date"] }, function=query_weather_impl )

注意那个enum字段，它把城市名约束在了一个白名单里。实测下来，加了枚举约束之后，模型编造城市名的概率从大概三成降到了接近零。后面讲参数校验时我还会再说这个问题。

2.3 规划器要不要单独存在

这是一个很容易被问的问题：ReAct框架本身已经能在循环里逐步决策了，为什么还要单独搞一个规划器？我的答案是：看任务复杂度。

如果任务是"查询天气—做判断—写记录"这种不超过五步的短链条，ReAct够用，不需要单独规划器。模型在每一轮直接输出"我下一步要调用什么工具"，循环几轮就结束了。

但如果任务是"调研十个数据源—清洗数据—生成报表—发邮件—归档"，几十步的链条，每一步都要模型现场决定，很容易在中途迷失方向，做着做着忘了最初要干什么。这时候就需要一个前置的规划器：先让模型一次性生成完整的执行计划，把每一步拆解清楚，然后执行器按计划逐步执行。

我的实践原则很简单：先做ReAct，跑通了再加规划器。第一版Agent用ReAct跑完一个三到五步的闭环，不要一上来就上复杂的规划器架构。等遇到"跑着跑着忘了任务目标"的问题，再引入规划器也不迟。

3. 核心算法：ReAct循环的输入输出协议与防呆设计

3.1 ReAct的输入输出协议

ReAct（Reasoning + Acting）是Agent最常用的控制算法，核心思想是让模型在"推理"和"行动"之间交替：推理出下一步该做什么，做一个动作，观察结果，再推理，再行动……直到得出最终答案。

工程实现上，每一轮循环要做三件事：拼Prompt、让模型输出结构化决策、执行决策并收集反馈。

Prompt的组织方式非常关键。我用的模板大致是这样：

你是一个能调用外部工具的智能助手。你可以使用以下工具： {tools_description} 你的输出必须是严格的JSON格式，选择以下两种结构之一： 1. {"thought": "你对当前状态的思考", "action": "工具名", "action_input": {...}} 2. {"thought": "你已经得到最终答案", "final_answer": "返回给用户的结论"} 当前任务：{task} 历史轨迹： {trajectory} 请继续。

这里有一个特别重要的设计：强制JSON输出，而不是让模型说人话。在工程里，解析自然语言输出非常痛苦——同一个意思模型能换十种说法，正则根本写不完。直接限制输出格式，解析就变成了一个json.loads的事情。模型的JSON生成能力在目前的模型里已经很稳了。

trajectory是历史轨迹，格式是完整的"Thought/Action/Observation"流水：

Thought 1: 用户想查天气并判断是否带外套，我需要先查询明天的天气 Action 1: {"tool": "query_weather", "args": {"city": "上海", "date": "2025-01-15"}} Observation 1: {"success": true, "data": {"city": "上海", "date": "2025-01-15", "max_temp": 18, "min_temp": 10, "condition": "多云"}}

每一轮循环，把新的Observation追加到trajectory里，然后重新发给模型。这就是ReAct的完整闭环。

3.2 工具调用的参数清洗与校验

模型生成的action_input不能直接传给工具执行。在真实运行中，模型经常给出不完全合法的参数：日期格式错了、枚举值根本不存在、必要字段漏掉了。所以必须在执行前加一道参数清洗层。

我写的校验函数有两层逻辑。第一层用JSON Schema做基础校验：

def validate_arguments(schema: dict, args: dict) -> tuple[bool, str]: try: jsonschema.validate(instance=args, schema=schema) return True, "" except jsonschema.ValidationError as e: return False, f"参数校验失败: {e.message}"

第二层是执行业务规则校验。比如日期不能是过去日期、城市必须在可用列表里、查询范围不能超过7天。这些规则JSON Schema表达不了或者表达起来很别扭，就单独写一个校验函数挂在工具注册表上：

def validate_weather_args(args: dict) -> str: city = args.get("city") date = args.get("date") if city not in SUPPORTED_CITIES: return f"暂不支持城市 '{city}'，请从 {SUPPORTED_CITIES} 中选择" if date < datetime.now().strftime("%Y-%m-%d"): return "日期不能早于今天" return ""

校验失败之后怎么办？不是报错给用户，而是把这个错误信息回填到Observation里，让模型自己想办法修正。这个设计后面会详细展开，它是Agent自我纠错的基石。

3.3 循环终止条件与防呆设计

没有终止条件的Agent是危险的。我在代码里强制设置了三个退出条件，任何一个满足就停止：

• 正常结束：模型输出了final_answer字段，循环结束，把这个答案返回给用户。
• 主动认输：模型输出了give_up字段并附带原因，表示它经过多次尝试认为任务无法完成。这比硬着头皮瞎编一个答案强得多。
• 达到最大步数：超过max_steps（我一般设5到10），强制终止，返回"任务未在允许步数内完成"。

max_steps这个东西是保命用的。没有它，一个陷入循环的Agent会无限调用工具，烧光你的预算。我见过最离谱的一次是测试阶段忘了设最大步数，一个Agent在一个查询失败后反复重试了40多轮，等发现时费用已经很高了。

另一个防呆设计是Observation超时截断。工具返回的数据可能很大——比如数据库查询返回了几百行记录。如果原封不动塞回Prompt，几轮下来上下文就爆了。我的做法是：每个Observation最多保留2000个字符，超出部分截断并提示"结果过长，已截断"。3.1节里轨迹中的Observation都保留精炼信息，不是全量数据。

4. 从零实现一个可运行的Agent：注册表、主循环与轨迹解读

4.1 公共接口定义

代码部分，我们直接实现一个最小可运行版本。为了通用性，我抽出一个LLMClient接口，不绑定任何具体模型厂商，你在自己的项目里替换成实际的SDK即可。

class LLMClient: """大模型调用客户端封装，实际项目中替换为具体实现""" def chat(self, system: str, messages: list[dict]) -> str: """输入系统提示词和对话消息，返回模型生成的文本""" raise NotImplementedError

然后是工具注册表：

from dataclasses import dataclass, field @dataclass class Tool: name: str description: str parameters_schema: dict function: callable validate: callable = lambda args: "" TOOL_REGISTRY: dict[str, Tool] = {}

4.2 实现两个具体工具

第一个是天气查询工具。真实项目中你会对接某个天气服务API，这里为了演示，直接用静态数据模拟：

def query_weather_impl(city: str, date: str, unit: str = "celsius") -> dict: # 实际项目改为调用天气服务API mock_db = { ("上海", "2025-01-15"): {"max_temp": 18, "min_temp": 10, "condition": "多云"}, } data = mock_db.get((city, date)) if data is None: return {"success": False, "error": "该日期暂无天气数据"} return {"success": True, "data": {"city": city, "date": date, **data}} weather_tool = Tool( name="query_weather", description="查询指定城市未来某天的天气，返回最高温度、最低温度、天气状况。", parameters_schema={ "type": "object", "properties": { "city": {"type": "string", "description": "城市名，如上海"}, "date": {"type": "string", "description": "日期，格式YYYY-MM-DD"}, "unit": {"type": "string", "enum": ["celsius", "fahrenheit"]} }, "required": ["city", "date"] }, function=query_weather_impl ) TOOL_REGISTRY[weather_tool.name] = weather_tool

第二个是记事本工具，负责把Agent的判断结果追加到当天笔记里。这个工具有一个重要的幂等设计：调用时可以传入一个task_id，同一任务重复写入不会产生重复记录。

note_store: list[dict] = [] def append_note_impl(content: str, task_id: str = "") -> dict: # 幂等键：如果同一task_id已经写过了，直接返回已存在。 if task_id and any(n["task_id"] == task_id for n in note_store): return {"success": True, "data": {"message": "笔记已存在，未重复写入"}} note = {"task_id": task_id, "content": content, "created_at": time.strftime("%Y-%m-%d %H:%M:%S")} note_store.append(note) return {"success": True, "data": {"note_id": len(note_store)}} note_tool = Tool( name="append_note", description="将一段文字追加写入当天的笔记中，返回笔记ID。", parameters_schema={ "type": "object", "properties": { "content": {"type": "string", "description": "要写入笔记的内容"}, "task_id": {"type": "string", "description": "幂等键，用于避免重复写入"} }, "required": ["content"] }, function=append_note_impl ) TOOL_REGISTRY[note_tool.name] = note_tool

4.3 主循环实现

主循环是整个Agent的核心，逻辑不复杂，但每一步都要有日志。我用一个Agent类来组织：

class Agent: def __init__(self, llm_client: LLMClient, max_steps: int = 6): self.llm = llm_client self.max_steps = max_steps def build_system_prompt(self) -> str: tools_desc = "\n".join( f"- {t.name}: {t.description}\n 参数Schema: {json.dumps(t.parameters_schema, ensure_ascii=False)}" for t in TOOL_REGISTRY.values() ) return f"""你是一个能调用外部工具的智能助手。你可以使用以下工具： {tools_desc} 你的输出必须是严格的JSON格式，只能选择以下三种结构之一： 1. {{"thought": "思考过程", "action": "工具名", "action_input": {{...}}}} 2. {{"thought": "思考过程", "final_answer": "返回给用户的结论"}} 3. {{"thought": "思考过程", "give_up": "放弃原因"}} 注意：action_input中所有参数必须严格符合工具的参数Schema；如果工具返回错误，基于Observation重新思考，不要重复同样的失败调用。""" def run(self, task: str) -> str: system = self.build_system_prompt() trajectory = [] messages = [] for step in range(1, self.max_steps + 1): user_content = f"当前任务：{task}\n\n历史轨迹：\n" + "\n".join(trajectory) messages.append({"role": "user", "content": user_content}) raw_output = self.llm.chat(system, messages) print(f"--- Step {step} 模型输出 ---\n{raw_output}") try: decision = json.loads(raw_output) except json.JSONDecodeError: # 模型输出非JSON时的降级处理 trajectory.append(f"Thought {step}: 模型输出无法解析，请重新输出合法JSON") continue thought = decision.get("thought", "") if "final_answer" in decision: return decision["final_answer"] if "give_up" in decision: return f"任务未完成，Agent主动放弃。原因：{decision['give_up']}" action_name = decision.get("action", "") action_input = decision.get("action_input", {}) if action_name not in TOOL_REGISTRY: trajectory.append(f"Thought {step}: {thought}") trajectory.append(f"Action {step}: 尝试调用不存在的工具 '{action_name}'") trajectory.append(f"Observation {step}: 错误：没有名为 '{action_name}' 的工具，请从注册表中选择") continue tool = TOOL_REGISTRY[action_name] # 第一层：JSON Schema参数校验 valid, msg = validate_arguments(tool.parameters_schema, action_input) if not valid: trajectory.append(f"Thought {step}: {thought}") trajectory.append(f"Action {step}: {action_name} {json.dumps(action_input, ensure_ascii=False)}") trajectory.append(f"Observation {step}: 参数校验失败：{msg}") continue # 第二层：业务规则校验 biz_msg = tool.validate(action_input) if biz_msg: trajectory.append(f"Thought {step}: {thought}") trajectory.append(f"Action {step}: {action_name} {json.dumps(action_input, ensure_ascii=False)}") trajectory.append(f"Observation {step}: 业务校验失败：{biz_msg}") continue # 执行工具调用 try: result = tool.function(**action_input) obs_text = json.dumps(result, ensure_ascii=False) except Exception as e: obs_text = f"执行异常：{e}" trajectory.append(f"Thought {step}: {thought}") trajectory.append(f"Action {step}: {action_name} {json.dumps(action_input, ensure_ascii=False)}") trajectory.append(f"Observation {step}: {obs_text[:2000]}") # 注意：下一轮要把新轨迹发给模型，这里不需要保留messages里的旧记录 messages = [] return "任务未在允许步数内完成。"

主循环的关键设计有几个。第一个是循环内每次都重新拼user_content，把完整轨迹发给模型，不使用多轮对话叠加——因为轨迹本身在持续增长，多轮对话容易让模型迷失在哪一条是最近的状态。第二个是校验失败不直接放弃，而是把失败原因作为Observation回填，让模型自我修正。第三个是每一步都打印输出，这是后续排查问题的基础。

4.4 实际跑一个任务并分析轨迹

假设我们跑的任务是："查询上海明天（2025-01-15）的天气，如果最高温度低于20度就提醒我带外套，并把结论写入当天笔记。"

第一步，模型输出通常是这样的：

{"thought": "需要先查询上海明天的天气情况", "action": "query_weather", "action_input": {"city": "上海", "date": "2025-01-15"}}

工具返回成功：

{"success": true, "data": {"city": "上海", "date": "2025-01-15", "max_temp": 18, "min_temp": 10, "condition": "多云"}}

第二步，模型拿到这份数据，发现最高温度18度，低于20度，于是判断需要提醒带外套，并且决定把结论写入笔记：

{"thought": "最高温度18度，低于20度，需要提醒用户带外套。现在把结论写入笔记。", "action": "append_note", "action_input": {"content": "2025-01-15上海最高温度18度，建议带外套", "task_id": "task-demo-20250115"}}

笔记写入成功：

{"success": true, "data": {"note_id": 1}}

第三步，模型输出最终答案：

{"thought": "天气查询和笔记记录都已完成", "final_answer": "上海明天（1月15日）最高温度18度，低于20度，记得带外套。已记录到笔记中。"}

整个任务3步完成。这个轨迹清楚地展示了Agent的行为闭环：查询→分析→决策→执行→汇总。每一个Observation都真实影响了模型的下一个动作，而不是模型凭感觉回答。

4.5 这段代码能work的关键点

总结一下，这套实现能稳定跑起来靠的是四条原则：

1. 结构化输出，不是自然语言输出。强制JSON格式，解析简单，错误率大幅降低。
2. 工具结果永远放进Observation，不让模型自己记住。模型不需要"记住"返回了什么，它只需要读最新轨迹里的Observation。这样即使模型上下文容量有限，也不会遗漏关键信息。
3. 失败反馈进轨迹，不进异常。校验失败和调用失败都变成Observation文本，模型看到后可以自我修正；直接抛异常会让Agent流程中断。
4. 轨迹即状态，上下文全部显式化。每一步的思考、动作、观察都追加到轨迹中，模型始终能看到最近发生了什么、已完成了什么、还差什么。

5. 实战排坑：上下文爆炸、幻觉调用与外部服务故障

5.1 上下文爆炸：历史轨迹越滚越大

这是Agent从demo走向实用遇到的第一座大山。第1步的任务轨迹只有几百字，第10步的时候可能已经上万字了。模型要处理越来越长的输入，延迟飙升，费用上涨，而且会"注意力稀释"——被早期轨迹干扰，忽略最近的关键信息。

我踩过一次很惨：一个多步骤的数据处理Agent，跑到第14步时开始重复调用同一个工具，给出的理由引用了第3步的Observation，完全忽略了第10步已经完成的事实。根源就是轨迹太长，模型抓不住重点。

解决方案是滑动窗口+定期摘要。我的做法是：

• 轨迹超过6轮后，把最早的3轮压缩成一段摘要。
• 摘要不是"大意概括"，而是提取关键状态：已完成哪些动作、得到过哪些关键数据、当前卡在什么问题上。
• 保留最近3轮完整轨迹，摘要+近3轮组合成新的上下文。

代码实现很朴素：

def summarize_early_trajectory(trajectory, summary_prompt) -> str: # 用模型对最早的N轮做摘要，保留关键状态 early = "\n".join(trajectory[:n]) summary = self.llm.chat( "你是任务状态记录员，将以下Agent执行轨迹压缩成关键状态摘要。", [{"role": "user", "content": early + summary_prompt}] ) return summary

摘要之后，上下文长度能控制在一个稳定范围内，Agent的执行稳定性显著提升。第二次出现重复调用时，我才注意到早期信息其实已经被正确保留了。

5.2 模型"幻觉调用"：工具名和参数都是编的

模型在生成JSON时，偶尔会输出一个不在注册表里的工具名，或者给一个枚举外的参数值。我在测试中遇到的典型情况：明明只有query_weather和append_note两个工具，模型输出了search_web；城市参数填了一个"杭州"但enum里根本没这个值。

不能指望模型永远不出错，要让系统对出错有承载力。我做了三层防护：

第一层是工具名白名单。解析出action字段后，先查TOOL_REGISTRY，不存在就把错误Observation喂回去："没有名为 X 的工具，可用的工具是 Y、Z"。

第二层是Schema自动校验。用jsonschema库校验类型、必填项、枚举值，校验失败同样走Observation回填。

第三层是业务规则校验。比如日期不能是过去、查询范围不能超过N天。这类规则模型理解得不好，但写在工具的业务校验函数里，稳定可靠。

经过这三层，模型即使一开始填错，也能在下一轮根据错误信息自我修正，任务依然能成功。这套机制比"换更聪明的模型"性价比高得多。

5.3 外部API超时、限流与幂等

Agent的工具执行层，本质是在编排一堆外部API调用。外部API最常见的故障是超时和限流。

超时必须有。我给每个工具的执行函数都加了显式超时控制。在工具层而不是Agent层做超时，逻辑更清晰——工具知道自己依赖的外部服务是什么量级的。

限流要考虑两个层面。Agent内部是串行调用，问题不大；但多个用户同时跑Agent，对同一个外部API的QPS就叠加了。我的方案是给每个工具配置一个信号量，限制同一工具的最大并发数。

幂等设计是最容易被忽略的一块。Agent在遇到超时会重试，但重试可能会导致副作用操作重复执行。"写入、发送、删除"这类操作必须做幂等。我的记事本工具的task_id就是为这个设计的：先查是否已存在，存在就直接返回"已执行过"。实际对接真实系统时，"查询-判断-写入"三步要做在一个事务或一个接口里，否则并发下仍可能重复。

表格总结一下我常用的工具层异常处理策略：

5.4 实测的费用与延迟数据

这里给一组我实测的大致数据，供大家做预算参考。模型本身会有波动，但数量级是可信的：

最直观的教训是：每多一步循环，成本近似线性增长，延迟也线性增长。所以控制步数是控成本的第一手段，不比模型选型次要。我后来在系统提示词里明确要求"能用一次工具调用完成的事不要分两次"，实测平均步数从5.2降到了3.7。

6. 从单Agent到多Agent：扩展的正确姿势与代价

6.1 主管-专员模式

单Agent处理的任务复杂度有上限。当工具数量超过15个时，把全部工具描述塞进Prompt会让模型选择困难症发作，经常挑错工具。这时候我开始考虑拆成"主管-专员"模式：

• 主管Agent（Supervisor）：负责接收任务、拆解子任务、分派给专员、汇总结果。
• 专员Agent（Worker）：每个专员只持有少量工具，负责一个领域的执行。

好处很明显：每个专员的工具注册表很小，Prompt里的工具描述更聚焦，工具选择的准确率大幅提升；主管的Prompt不需要包含具体工具细节，专注于任务拆解和调度。

代价同样明显：多一交互就多一轮LLM调用，延迟和成本直接翻倍。所以我的建议是：工具少于10个时，单Agent足够了；工具超15个、任务类型明显分域时，再考虑主管-专员。

6.2 Agent之间的消息传递用什么格式

多Agent之间传递消息，最容易犯的错误是把一长串自然语言对话直接丢给下一个Agent。这会让接收方迷失重点，也会放大token消耗。

我用的消息格式是结构化的：

{ "task_id": "t-20250115-001", "source": "supervisor", "target": "weather_worker", "status": "pending", "instruction": "查询上海明天最高温度", "callback": "report_weather_result", "context": {"city": "上海", "date": "2025-01-15"} }

每个Agent处理完，返回这样一份结果：

{ "status": "success", "summary": "上海明天最高温度18度", "data": {"max_temp": 18}, "cost": {"input_tokens": 1200, "output_tokens": 400} }

结构化消息有几个好处：接收方不需要从自然语言里解析关键信息；可以方便地做路由和状态机；每个Agent的执行成本清晰可查。消息里坚决不传长文本数据，比如查询结果全量JSON，"数据放库、消息只传摘要"是我在多Agent项目里守得很死的原则。

6.3 观测体系：出问题时怎么回放

Agent系统的bug和传统软件不一样。传统软件的bug基本可复现，Agent的bug是概率性的——可能这次成功，下次同样的输入就失败了。所以观测体系比代码正确性更重要。

我每一版Agent都会强制记录三类日志：

1. 完整轨迹日志：每轮的Thought、Action、Observation全量落地。
2. 费用日志：每次LLM调用的输入输出tokens数、工具调用次数、总成本。
3. 成功/失败标记：任务最终状态是成功、放弃、还是超步数，附失败原因。

排查问题的时候，我最常用的手段就是轨迹回放：把失败任务的完整轨迹打印出来，逐行看模型是在哪一步开始跑偏的。95%的失败任务，看轨迹就能定位原因——是工具描述有歧义、是参数Schema有漏洞、还是模型在第N步产生了错误推理。

后来我又给轨迹打印加了一个增强功能：在每一步旁边标注这个动作是哪一层（大脑决策、参数校验、工具执行、结果观测）发生的，排查时一眼就能锁定问题模块。

最后说一点个人体会吧。如果只能记住一件事，我的建议是：先跑通最小的闭环，再谈架构复杂度。我第一版Agent只有150行代码，没有规划器、没有记忆管理、没有多Agent协作，但它在真实任务上完成了从"调用工具"到"根据结果决定下一步"的完整循环。后来所有的复杂化——上下文摘要、参数校验、多Agent协作——都是在这个150行的骨架上一点一点长出来的。

如果一上来就照着大厂的技术博客设计一堆模块，大概率会在复杂架构里迷失，连最核心的闭环都跑不通。

再送一个自查小技巧：跑完一个Agent任务后，打开轨迹日志问自己两个问题。第一，你的Agent在工具返回空结果时，是主动说"我换一种方式试试"，还是直接崩溃或瞎编？第二，它在连续两次失败后，是明确告诉你"这个我做不了"，还是继续无效重试？如果两个都有答案，你的Agent已经超过大多数demo级项目了。如果你的轨迹日志根本还没打开过，放下代码，先去把日志补上。