☰
Agent框架化实战:ReAct与Plan-and-Solve统一抽象与实现
2026/10/7 6:30:56 网站建设 项目流程

1. Agent范式框架化的核心思路拆解

1.1 为什么要把Agent范式“框”起来

很多人第一次接触Agent开发,脑子里浮现的画面是:写一个循环,让大模型不断调用工具、观察结果、再决定下一步。这个循环写起来可能就几十行代码,跑个Demo完全没问题。但一旦要接入真实业务、要处理多轮对话、要支持多种工具、要记录中间状态、要控制成本,那几十行代码就会迅速膨胀成几百行甚至上千行的“面条式”逻辑。Agent范式的框架化,本质上就是把这团面条梳理成可复用、可测试、可扩展的结构。

我自己的体会是,Agent开发最怕的不是模型不够聪明,而是流程不可控。你永远不知道模型下一步会调用哪个工具,也不知道它会不会陷入死循环,更不知道它什么时候会把上下文撑爆。框架化的第一个价值,就是把这些不确定性收拢到几个明确的抽象里:思考节点、行动节点、观察节点、终止条件。一旦这些抽象固定下来,后续换模型、加工具、改提示词,都只是替换某个零件,而不是重写整个系统。

从热词里也能看出大家的关注点:Agent、框架化、SimpleAgent、ReAct、Plan-and-Solve。这几个词其实指向同一个问题——如何用一套通用的骨架,承载不同的推理范式。ReAct强调“边想边做”,Plan-and-Solve强调“先规划再执行”,它们看起来是两种策略,但在框架层面,它们共享的组件远多于差异。框架化的目标,就是让这些范式成为可插拔的策略,而不是两套独立的代码。

1.2 框架化要解决的三个核心痛点

第一个痛点是状态管理。Agent在执行过程中会产生大量中间状态:当前的任务目标、已经执行过的步骤、每个步骤的输入输出、工具调用的结果、模型的原始回复。如果没有统一的状态容器,这些信息就会散落在各个函数里,调试时根本追不到问题。框架化要求把这些状态集中管理,并且提供清晰的读写接口。

第二个痛点是工具调用的标准化。不同工具的参数格式、返回结构、错误处理方式都不一样。如果每次加工具都要改Agent主循环,那这个系统就没法规模化。框架化需要定义统一的工具接口:输入是什么、输出是什么、异常怎么抛、超时怎么处理。这样新增工具就只是注册一个函数,而不是修改核心逻辑。

第三个痛点是推理策略的可替换性。今天用ReAct,明天想试试Plan-and-Solve,后天可能想混合使用。如果推理逻辑和主循环耦合在一起,切换策略就意味着重写。框架化要把“怎么想”和“怎么做”分开,让推理策略成为一个独立的模块,主循环只负责调度。

1.3 从SimpleAgent到通用框架的演进路径

SimpleAgent通常是最小可运行版本:一个while循环,加上模型调用和工具执行。它的优点是直观,缺点是脆弱。任何一点需求变化,比如要支持多轮记忆、要限制最大步数、要记录执行轨迹,都会让代码变得臃肿。

框架化的演进路径,我建议分三步走。第一步,把SimpleAgent里的各个职责拆成独立函数:构建提示词、调用模型、解析动作、执行工具、更新状态。第二步,定义数据结构,把状态、动作、观察结果都变成明确的对象,而不是散落的字典。第三步,引入策略接口,让ReAct和Plan-and-Solve都实现同一套接口,主循环只依赖接口,不依赖具体实现。

这个演进过程不需要一次性完成。你可以先跑通SimpleAgent,然后在遇到具体问题时逐步重构。关键是心里要有那张框架图,知道每个模块的边界在哪里。

2. ReAct与Plan-and-Solve的框架化实现细节

2.1 ReAct范式的核心循环与实现要点

ReAct的核心是“Thought-Action-Observation”循环。模型先输出一段思考,然后决定调用哪个工具,工具返回结果后,模型再基于新信息继续思考。这个循环看起来简单,但实现时有几个细节非常关键。

首先是提示词的结构。ReAct的提示词需要明确告诉模型:你可以使用哪些工具、每个工具的参数格式是什么、你需要按照“Thought: ... Action: ... Action Input: ...”的格式输出。很多人在这一步偷懒,结果模型输出的格式五花八门,解析器直接崩溃。我的经验是,提示词里一定要给一个完整的示例,并且用明确的标记把示例和实际任务分开。

其次是动作解析的容错。模型不总是乖乖按格式输出,有时候会多写一段解释,有时候会把Action写成小写,有时候会忘记Action Input。解析器不能假设模型永远正确,必须能处理这些异常。常见的做法是用正则表达式提取关键字段,如果提取失败,就把模型的原始输出作为观察结果返回,让它自己纠正。

第三是循环终止条件。ReAct需要一个明确的结束信号,通常是模型输出“Final Answer: ...”或者达到最大步数。如果没有终止条件,模型可能会无限循环下去。我一般会设置两个阈值:最大步数和最大Token消耗,任何一个超标就强制终止,并返回当前最好的结果。

# ReAct循环的简化框架 def react_loop(task, tools, max_steps=10): state = {"task": task, "history": [], "step": 0} while state["step"] < max_steps: prompt = build_react_prompt(state, tools) response = call_llm(prompt) thought, action, action_input = parse_response(response) if action == "Final Answer": return action_input observation = execute_tool(action, action_input, tools) state["history"].append({ "thought": thought, "action": action, "input": action_input, "observation": observation }) state["step"] += 1 return "达到最大步数,任务未完成"

这段代码看起来简单,但每个函数里面都有大量细节。比如build_react_prompt需要把历史记录格式化成模型能理解的形式,parse_response需要处理各种格式异常,execute_tool需要处理工具不存在、参数错误、执行超时等情况。

2.2 Plan-and-Solve的规划与执行分离

Plan-and-Solve的思路和ReAct不同。它先让模型生成一个完整的计划,把任务拆成若干步骤,然后逐步执行每个步骤。执行过程中,模型可以根据实际情况调整后续计划,但整体上有一个明确的路线图。

这种范式的优势在于全局视野。ReAct是走一步看一步,容易陷入局部最优;Plan-and-Solve先看全局,再动手,适合步骤较多、依赖关系复杂的任务。但它的缺点也很明显:如果初始计划有问题,后续执行就会一路错下去。所以框架化实现时,需要加入计划修正机制。

我的做法是在每个步骤执行后,让模型判断当前计划是否仍然有效。如果发现偏差,就重新生成剩余步骤的计划。这样既保留了全局规划的优势,又增加了灵活性。

# Plan-and-Solve的框架化实现 def plan_and_solve(task, tools, max_replans=3): plan = generate_plan(task, tools) results = [] replan_count = 0 for step in plan: result = execute_step(step, tools) results.append(result) if not verify_step(step, result): if replan_count < max_replans: plan = replan(task, results, tools) replan_count += 1 else: break return synthesize_results(results)

这里的关键是generate_plan和replan的质量。计划不能太粗,否则执行时无从下手;也不能太细,否则模型容易在细节上出错。我一般会要求计划包含3到7个步骤,每个步骤有明确的输入和预期输出。

2.3 两种范式的统一抽象与切换策略

ReAct和Plan-and-Solve虽然思路不同,但在框架层面可以统一。它们都需要:状态容器、工具注册表、模型调用接口、结果解析器、终止条件判断。差异只在于决策时机:ReAct每一步都重新决策,Plan-and-Solve在开始时决策一次,执行中偶尔修正。

框架化实现时,可以定义一个ReasoningStrategy接口,包含next_action(state)方法。ReAct的实现是每次调用都返回下一步动作,Plan-and-Solve的实现是返回当前计划的下一步。主循环只调用next_action,不关心具体策略。

class ReasoningStrategy: def next_action(self, state): raise NotImplementedError class ReActStrategy(ReasoningStrategy): def next_action(self, state): # 基于当前历史,让模型决定下一步 pass class PlanAndSolveStrategy(ReasoningStrategy): def __init__(self, task, tools): self.plan = generate_plan(task, tools) self.index = 0 def next_action(self, state): if self.index < len(self.plan): action = self.plan[self.index] self.index += 1 return action return "Final Answer"

这种抽象的好处是,切换策略只需要换一个实现类,主循环代码完全不用动。你甚至可以在运行时动态切换:简单任务用ReAct,复杂任务用Plan-and-Solve。

2.4 工具注册与调用的标准化设计

工具是Agent的手脚,工具调用的标准化程度直接决定了框架的扩展性。我见过很多项目,每加一个工具就要改一次主循环,最后主循环里堆满了if-else。正确的做法是定义一个工具注册表,每个工具包含:名称、描述、参数schema、执行函数。

class Tool: def __init__(self, name, description, parameters, func): self.name = name self.description = description self.parameters = parameters self.func = func def execute(self, **kwargs): # 参数校验 validate_params(kwargs, self.parameters) try: return self.func(**kwargs) except Exception as e: return f"工具执行失败: {str(e)}" class ToolRegistry: def __init__(self): self.tools = {} def register(self, tool): self.tools[tool.name] = tool def get(self, name): return self.tools.get(name) def list_descriptions(self): return "\n".join([ f"{t.name}: {t.description}" for t in self.tools.values() ])

参数schema我建议用JSON Schema格式,这样既能用于校验,也能直接嵌入提示词让模型理解参数要求。工具执行函数要捕获所有异常,把错误信息作为观察结果返回给模型,而不是让整个流程崩溃。

3. 完整实操:从零搭建一个框架化Agent

3.1 环境准备与依赖选型

搭建框架化Agent,语言选择上Python是最顺手的,生态最全,调试也方便。如果你追求性能,可以考虑用Rust或Go重写核心循环,但开发效率会下降不少。我的建议是先用Python跑通逻辑,等性能成为瓶颈再考虑迁移。

依赖方面,核心只需要一个HTTP客户端来调用模型API。如果你用OpenAI兼容的接口,httpx或requests就够了。不需要一上来就引入LangChain这类重型框架,它们抽象层太多,反而会掩盖核心逻辑。自己从零写一遍,对理解Agent运行机制帮助极大。

pip install httpx pydantic

pydantic用来定义数据模型,让状态和动作的结构更清晰。如果你不想引入额外依赖,用dataclass也可以。

3.2 核心数据结构定义

框架化的第一步是把数据结构定清楚。我一般定义三个核心类:AgentState、Action、Observation。

from pydantic import BaseModel from typing import List, Optional, Dict, Any class Action(BaseModel): name: str input: Dict[str, Any] class Observation(BaseModel): action_name: str result: str success: bool class AgentState(BaseModel): task: str history: List[Dict[str, Any]] = [] current_step: int = 0 max_steps: int = 15 final_answer: Optional[str] = None def add_round(self, thought: str, action: Action, observation: Observation): self.history.append({ "step": self.current_step, "thought": thought, "action": action.dict(), "observation": observation.dict() }) self.current_step += 1 def is_finished(self): return self.final_answer is not None or self.current_step >= self.max_steps

AgentState是整个框架的核心,所有中间信息都集中在这里。这样做的好处是,调试时只需要打印一个对象就能看到完整执行轨迹,序列化保存也很方便。

3.3 提示词模板与解析器实现

提示词模板决定了模型输出的质量,解析器决定了框架的健壮性。这两者需要配合设计:模板里要求的格式,解析器必须能处理;解析器能处理的格式,模板里要明确说明。

REACT_PROMPT_TEMPLATE = """你是一个智能助手,可以使用以下工具: {tool_descriptions} 请按照以下格式回答: Thought: 你的思考过程 Action: 工具名称,或者 "Final Answer" Action Input: 工具的输入参数,JSON格式 如果已经得到最终答案,请输出: Thought: 我已经知道答案了 Action: Final Answer Action Input: {{"answer": "你的最终答案"}} 历史记录: {history} 当前任务:{task} 请开始:""" def build_prompt(state, tool_registry): history_text = format_history(state.history) return REACT_PROMPT_TEMPLATE.format( tool_descriptions=tool_registry.list_descriptions(), history=history_text, task=state.task )

解析器要处理各种异常情况。我的经验是,模型输出格式错误的比例大概在5%到10%之间,所以解析器必须足够宽容。

import re import json def parse_response(response: str): # 提取Thought thought_match = re.search(r"Thought:\s*(.+?)(?=Action:|$)", response, re.DOTALL) thought = thought_match.group(1).strip() if thought_match else "" # 提取Action action_match = re.search(r"Action:\s*(.+?)(?=Action Input:|$)", response, re.DOTALL) action_name = action_match.group(1).strip() if action_match else "" # 提取Action Input input_match = re.search(r"Action Input:\s*(.+)", response, re.DOTALL) action_input = {} if input_match: raw_input = input_match.group(1).strip() try: action_input = json.loads(raw_input) except json.JSONDecodeError: # 尝试提取JSON片段 json_match = re.search(r"\{.*\}", raw_input, re.DOTALL) if json_match: try: action_input = json.loads(json_match.group(0)) except: action_input = {"raw": raw_input} else: action_input = {"raw": raw_input} return thought, action_name, action_input

注意:解析器不要假设模型一定按格式输出。我遇到过模型把Action写成“行动”、把Action Input写成“参数”的情况,所以正则表达式要尽量宽松,并且在提示词里反复强调格式要求。

3.4 主循环与策略调度实现

主循环是整个框架的调度中心,它不包含具体的推理逻辑,只负责协调各个模块。

class Agent: def __init__(self, llm_client, tool_registry, strategy="react"): self.llm = llm_client self.tools = tool_registry self.strategy = strategy def run(self, task: str, max_steps: int = 15) -> str: state = AgentState(task=task, max_steps=max_steps) while not state.is_finished(): # 1. 构建提示词 prompt = build_prompt(state, self.tools) # 2. 调用模型 response = self.llm.call(prompt) # 3. 解析响应 thought, action_name, action_input = parse_response(response) # 4. 判断是否结束 if action_name == "Final Answer": state.final_answer = action_input.get("answer", "") break # 5. 执行工具 tool = self.tools.get(action_name) if tool: result = tool.execute(**action_input) observation = Observation( action_name=action_name, result=str(result), success=True ) else: observation = Observation( action_name=action_name, result=f"工具 {action_name} 不存在,可用工具:{list(self.tools.tools.keys())}", success=False ) # 6. 更新状态 state.add_round(thought, Action(name=action_name, input=action_input), observation) return state.final_answer or "任务未完成"

这个主循环只有几十行,但已经包含了Agent运行的所有核心环节。新增工具只需要注册到tool_registry,切换策略只需要传入不同的strategy参数。

3.5 执行轨迹记录与调试技巧

调试Agent最有效的方法是把每一步的完整信息都记录下来。我一般会在AgentState里加一个trace字段,记录每个步骤的耗时、Token消耗、模型原始输出。

import time def run_with_trace(self, task): state = AgentState(task=task) trace = [] while not state.is_finished(): start = time.time() prompt = build_prompt(state, self.tools) response = self.llm.call(prompt) elapsed = time.time() - start trace.append({ "step": state.current_step, "prompt_length": len(prompt), "response": response, "elapsed": elapsed, "timestamp": time.time() }) # ... 后续处理 return state.final_answer, trace

有了这份trace,排查问题时就能清楚看到:是提示词太长导致模型注意力分散,还是某个工具返回结果格式不对导致模型理解错误,还是模型本身在某个步骤上反复纠结。

4. 常见问题与排查技巧实录

4.1 模型不按格式输出怎么办

这是最常见的问题,没有之一。模型可能会用中文写“行动”而不是“Action”,可能会把JSON写成Python字典格式,可能会在Action Input里塞一段自然语言。我的处理策略分三层。

第一层是提示词加固。在提示词里用加粗、分隔线、示例等方式反复强调格式。我通常会在提示词开头、工具列表后面、任务描述前面各放一次格式说明,重要的事情说三遍。

第二层是解析器容错。正则表达式要覆盖常见变体,比如Action[::]同时匹配中英文冒号,Action\s*Input允许中间有空格。JSON解析失败时,尝试用ast.literal_eval解析Python字典格式。

第三层是错误反馈。如果解析完全失败,不要把空动作传给工具执行,而是构造一个观察结果:“你的输出格式不正确,请严格按照 Thought/Action/Action Input 格式重新输出。”把这个反馈加入历史,让模型自我纠正。

4.2 工具调用陷入死循环怎么破

死循环通常有两种表现:模型反复调用同一个工具,或者模型在两个工具之间来回切换。根本原因往往是工具返回的结果没有提供足够的信息让模型做出新决策。

我的解决方案是在观察结果里加入引导信息。比如搜索工具返回空结果时,不要只返回“无结果”,而是返回“未找到相关信息,建议尝试其他关键词或使用其他工具”。这样模型就知道当前路径走不通,需要换策略。

另外,我会在状态里记录每个工具被调用的次数。如果某个工具连续被调用超过3次,就在提示词里加入提醒:“你已经多次调用XX工具,请考虑是否有其他方式解决问题。”

def check_loop(state): recent_actions = [h["action"]["name"] for h in state.history[-3:]] if len(recent_actions) == 3 and len(set(recent_actions)) == 1: return f"注意:你已经连续3次调用 {recent_actions[0]} 工具,请尝试其他方法。" return ""

4.3 上下文过长导致模型“失忆”

Agent执行步数多了之后,历史记录会越来越长,最终超出模型的上下文窗口。这时候模型会开始“失忆”,忘记最初的任务目标,或者忽略早期的关键信息。

处理这个问题有几种策略。最简单的是滑动窗口:只保留最近N步的历史,更早的用摘要代替。摘要可以由模型生成,也可以简单地保留任务描述和最终目标。

另一种策略是关键信息提取:每执行几步,就让模型总结当前进展和剩余任务,用总结替换详细历史。这样既保留了关键信息,又大幅压缩了长度。

def compress_history(history, max_rounds=5): if len(history) <= max_rounds: return history old_rounds = history[:-max_rounds] recent_rounds = history[-max_rounds:] summary = summarize_rounds(old_rounds) return [{"summary": summary}] + recent_rounds

提示:压缩历史时一定要保留任务描述和最终目标,否则模型会彻底迷失方向。我一般会把任务描述单独放在提示词的开头,不参与压缩。

4.4 工具执行超时与异常处理

工具执行可能因为网络问题、外部服务故障、参数错误等原因失败。如果框架不处理这些异常,整个Agent就会崩溃。我的做法是在工具执行层加超时和重试机制。

import signal class TimeoutError(Exception): pass def timeout_handler(signum, frame): raise TimeoutError("工具执行超时") def execute_with_timeout(func, args, timeout_seconds=30): signal.signal(signal.SIGALRM, timeout_handler) signal.alarm(timeout_seconds) try: result = func(**args) signal.alarm(0) return result except TimeoutError: return "工具执行超时,请检查输入或稍后重试" except Exception as e: return f"工具执行异常: {str(e)}" finally: signal.alarm(0)

异常信息要作为观察结果返回给模型,而不是直接抛出。这样模型有机会根据错误信息调整策略,比如换一个工具、修改参数、或者放弃当前路径。

4.5 常见问题速查表

问题现象可能原因排查方法解决方案
模型输出格式混乱提示词不够明确检查提示词中的格式说明和示例加固提示词,增加格式示例
解析器报错模型输出变体打印原始输出,检查正则匹配放宽正则,增加容错分支
工具调用死循环观察结果信息不足查看历史记录中的工具调用序列在观察结果中加入引导信息
上下文超长历史记录累积统计每步的Token消耗滑动窗口或摘要压缩
工具执行失败参数错误或外部故障检查工具输入和异常信息超时重试,错误反馈给模型
任务未完成就终止最大步数设置过小查看执行轨迹,判断是否接近完成调整max_steps,或优化提示词
模型忽略早期信息上下文过长导致注意力分散检查提示词中任务描述的位置把任务描述放在提示词开头和结尾

4.6 几个踩过的坑和独家技巧

第一个坑是工具描述太简略。我一开始写工具描述就一句话,结果模型经常选错工具。后来我把描述改成“功能+适用场景+不适用场景”三段式,工具选择准确率明显提升。比如搜索工具的描述写成:“用于查找事实性信息。适用于需要最新数据或外部知识的场景。不适用于数学计算或逻辑推理。”

第二个坑是Action Input参数名不匹配。模型有时候会用query,有时候用keyword,有时候用search_term。后来我在工具定义里加了参数别名映射,把常见变体都映射到标准参数名上。

第三个技巧是给模型一个“思考模板”。在提示词里不仅要求输出Thought,还给出思考的维度:当前进展是什么、还缺什么信息、下一步应该做什么。这样模型的思考质量会高很多,不容易跑偏。

第四个技巧是用少量示例引导。在提示词里放一两个完整的ReAct循环示例,模型模仿起来会顺畅很多。示例要覆盖成功路径和失败恢复路径,让模型知道出错时该怎么处理。

5. 框架化Agent的扩展方向与性能优化

5.1 多Agent协作的框架扩展

单Agent能力有限,复杂任务往往需要多个Agent分工协作。框架化设计为多Agent扩展提供了天然的基础:每个Agent可以看作一个特殊的工具,接受任务描述,返回执行结果。

class AgentTool(Tool): def __init__(self, name, description, agent): super().__init__(name, description, {"task": "string"}, self.run_agent) self.agent = agent def run_agent(self, task): return self.agent.run(task)

这样主Agent可以把子任务委托给专门的子Agent,比如一个负责搜索、一个负责计算、一个负责写作。子Agent的执行结果作为观察结果返回给主Agent,主Agent再决定下一步。

多Agent协作的关键是任务分解的粒度。分得太粗,子Agent压力大;分得太细,通信开销高。我的经验是,每个子任务应该能在3到5步内完成,超过这个范围就继续拆分。

5.2 记忆机制的引入与持久化

Agent的记忆分为短期记忆和长期记忆。短期记忆就是当前任务的执行历史,长期记忆是跨任务的知识积累。框架化实现时,短期记忆已经由AgentState承载,长期记忆需要额外设计。

最简单的长期记忆是向量数据库:把历史任务的成功经验和失败教训存入向量库,新任务开始时检索相似案例作为参考。这样Agent就能“吃一堑长一智”,不会在同一个地方反复跌倒。

class Memory: def __init__(self, vector_store): self.store = vector_store def recall(self, task, top_k=3): results = self.store.search(task, top_k=top_k) return [r["content"] for r in results] def remember(self, task, result, success): self.store.add({ "task": task, "result": result, "success": success, "timestamp": time.time() })

记忆写入时要注意脱敏,不要把敏感信息存进去。记忆检索时要设置相似度阈值,太低的相关性反而会干扰模型判断。

5.3 并发场景下的Agent性能考量

Agent执行是IO密集型任务,大部分时间花在等待模型响应和工具执行上。如果要支持并发,框架需要做到无状态化:每个请求创建独立的AgentState,共享的工具注册表和模型客户端要线程安全。

from concurrent.futures import ThreadPoolExecutor class AgentPool: def __init__(self, llm_client, tool_registry, pool_size=10): self.executor = ThreadPoolExecutor(max_workers=pool_size) self.llm = llm_client self.tools = tool_registry def run_batch(self, tasks): futures = [] for task in tasks: agent = Agent(self.llm, self.tools) futures.append(self.executor.submit(agent.run, task)) return [f.result() for f in futures]

并发时要注意模型API的速率限制,最好加一个令牌桶或信号量来控制请求频率。工具执行如果有共享资源,也要加锁保护。

5.4 成本控制与Token优化策略

Agent的Token消耗主要来自三部分:提示词、模型输出、历史记录。提示词里的工具描述和格式说明是固定开销,历史记录是增长最快的部分。

优化策略有几个方向。第一,精简工具描述,只保留必要信息,去掉冗余解释。第二,压缩历史记录,用摘要代替详细步骤。第三,限制输出长度,在提示词里要求模型简洁回答。第四,缓存重复计算,比如相同的工具调用结果可以缓存复用。

class CachedTool(Tool): def __init__(self, tool, cache_ttl=300): super().__init__(tool.name, tool.description, tool.parameters, self.cached_execute) self.tool = tool self.cache = {} self.cache_ttl = cache_ttl def cached_execute(self, **kwargs): key = json.dumps(kwargs, sort_keys=True) if key in self.cache: result, timestamp = self.cache[key] if time.time() - timestamp < self.cache_ttl: return result result = self.tool.execute(**kwargs) self.cache[key] = (result, time.time()) return result

我实测下来,合理的缓存策略能减少20%到30%的工具调用次数,Token消耗也能降低15%左右。对于高频重复的任务,效果更明显。

5.5 安全边界与输出过滤

Agent调用工具时可能产生副作用,比如写入文件、发送请求、修改数据。框架需要设置安全边界,防止Agent执行危险操作。

我的做法是给工具加权限标签:只读工具、写入工具、危险工具。危险工具执行前需要人工确认,或者只在特定条件下允许执行。另外,工具输入要做校验,防止注入攻击。

class SafeTool(Tool): def __init__(self, name, description, parameters, func, risk_level="low"): super().__init__(name, description, parameters, func) self.risk_level = risk_level def execute(self, **kwargs): if self.risk_level == "high": # 记录日志,等待确认 log_high_risk_call(self.name, kwargs) if not confirm_execution(self.name, kwargs): return "操作被拒绝:高风险工具需要人工确认" return super().execute(**kwargs)

输出过滤同样重要。模型生成的内容可能包含不当信息,需要在返回给用户前做一次过滤。我一般会检查是否包含敏感词、是否包含可执行代码、是否包含外部链接,根据业务场景决定是拦截还是替换。

5.6 框架化Agent的测试与评估

框架化之后,测试变得容易很多。你可以单独测试解析器、单独测试工具执行、单独测试策略切换,而不需要每次都跑完整流程。

我一般会建一个测试集,包含各种典型任务和边界情况:简单问答、多步推理、工具调用失败、格式异常、超长上下文。每次修改框架后跑一遍测试集,确保没有回归。

评估指标方面,除了任务完成率,我还会关注:平均步数、平均Token消耗、工具调用准确率、格式解析成功率。这些指标能帮你定位瓶颈在哪里。

def evaluate_agent(agent, test_cases): results = [] for case in test_cases: start = time.time() answer = agent.run(case["task"]) elapsed = time.time() - start results.append({ "task": case["task"], "expected": case["expected"], "actual": answer, "correct": case["expected"] in answer, "elapsed": elapsed, "steps": agent.last_state.current_step, "tokens": agent.last_token_count }) return results

这套评估流程跑下来,你能清楚看到框架在哪些场景下表现好,哪些场景下需要优化。我自己的经验是,工具调用准确率往往是最先需要优化的环节,因为模型选错工具后面全错。

5.7 从框架到产品的最后一公里

框架跑通之后,距离产品化还有一段路。你需要考虑:用户输入怎么接入、执行过程怎么展示、中间结果怎么保存、错误怎么友好提示、多轮对话怎么管理。

我的建议是先把执行轨迹可视化。用户不需要看到每一步的原始输出,但需要知道Agent在做什么、进展如何。一个简单的进度条加上当前步骤描述,体验就会好很多。

另外,要给用户提供中断和纠正的能力。Agent跑偏了,用户应该能随时叫停,补充信息,然后继续。这要求框架支持状态保存和恢复,AgentState的序列化设计在这里就派上用场了。

最后再分享一个小技巧:在生产环境里给Agent加一个“影子模式”。新策略或新工具先跑影子模式,只记录不执行,对比效果后再正式启用。这样能避免上线后才发现问题,减少对用户的影响。

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询