:Agent 与工具系统完整集成——从 ReAct 到自主决策)
前言:零件齐了,该组装了第一篇我们实现了 ReAct 核心循环,第二篇构建了工具系统(注册、校验、安全、追踪)。两个模块各自能跑,但接在一起才是真正的 Agent。这篇文章做一件事:把 ReAct 循环和工具系统完整集成,让 Agent 从"能想能做"变成"能在思考、调用工具、观察结果之间自主循环,直到完成任务"。一、集成的核心挑战把循环和工具接在一起,看起来简单——循环里调用工具,工具结果喂回循环。但细节上会出四个问题:问题表现解决思路停止条件不明确Agent 无限循环,Token 烧光四层停止条件(见第三节)工具结果过大搜索结果 5000 字全塞进对话历史,上下文爆炸结果截断 + 关键信息提取工具选择错误Agent 调了不存在的工具,反复重试动态工具列表注入 + 错误反馈任务漂移执行到第 5 步,Agent 忘了原始问题每步注入原始任务 + 进度追踪二、集成架构┌─────────────────────────────────────────────────┐ │ Agent.run(query) │ │ │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Thought │───▶│ Action │───▶│Observation│ │ │ │ LLM 推理 │ │ 执行工具 │ │ 结果注入 │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ ▲ │ │ │ └──── 检查停止条件 ←──────────────┘ │ │ └──── 未满足 → 继续循环 │ │ └──── 满足 → 返回最终答案 │ │ │ │ 上下文管理(每步动态组装): │ │ SystemPrompt + 工具列表 + 原始任务 + 历史记录 │ └─────────────────────────────────────────────────┘三、停止条件:四个维度保底Agent 最怕无限循环。我们用四层停止条件:fromenumimportEnumclassStopReason(Enum):FINAL_ANSWER="final_answer"# LLM 主动输出 Final AnswerMAX_STEPS="max_steps"# 达到最大步数MAX_TOKENS="max_tokens"# Token 超限MAX_COST="max_cost"# 费用超限NO_PROGRESS="no_progress"# 连续 3 步无进展DANGEROUS_LOOP="dangerous_loop"# 检测到死循环(重复同一工具+参数)classAgentStopper:"""四层停止条件检查器"""def__init__(self,max_steps:int=15,max_tokens:int=50000,max_cost:float=0.50,no_progress_threshold:int=3):self.max_steps=max_steps self.max_tokens=max_tokens self.max_cost=max_cost self.no_progress_threshold=no_progress_threshold self.total_tokens=0self.total_cost=0.0self.steps=0self.last_actions=[]# 最近 N 个 action 记录self.last_useful_result=0# 最后拿到有用结果的步数defcheck(self,step:int,tokens_this_step:int,cost_this_step:float,action:str=None,params:dict=None,result:str=None,has_final:bool=False)-tuple[bool,StopReason]:self.steps=step self.total_tokens+=tokens_this_step self.total_cost+=cost_this_step# Layer 1: LLM 主动结束ifhas_final:returnTrue,StopReason.FINAL_ANSWER# Layer 2: 硬上限ifstep=self.max_steps:returnTrue,StopReason.MAX_STEPSifself.total_tokens=self.max_tokens:returnTrue,StopReason.MAX_TOKENSifself.total_cost=self.max_cost:returnTrue,StopReason.MAX_COST# Layer 3: 连续无效步数检测ifactionandresult:action_key=f"{action}:{json.dumps(params,sort_keys=True)}"self.last_actions.append(action_key)# 检查最近 no_progress_threshold 个 action 是否都相同iflen(self.last_actions)self.no_progress_threshold:self.last_actions.pop(0)if(len(self.last_actions)=self.no_progress_thresholdandlen(set(self.last_actions[-self.no_progress_threshold:]))==1):returnTrue,StopReason.DANGEROUS_LOOP# 检查是否有实质性进展ifresultand"错误"notinresultand"查不到"notinresult:self.last_useful_result=step# Layer 4: 长期无进展ifstep-self.last_useful_resultse