从while循环到协议驱动:构建可控AI Agent的工程实践
1. 项目概述:从“死循环”到“可控系统”的范式转变
“可控 Agent 不是一个 while 循环”,这句话乍一听像是句技术圈的“黑话”,但它精准地戳中了当前 AI Agent 开发中的一个普遍痛点。很多开发者,包括我自己在早期探索时,都曾陷入一个思维定式:构建一个 Agent,不就是写一个while True循环,在里面不断调用大模型、解析返回、执行工具,直到任务完成为止吗?这种模式简单直接,上手快,但当你真正把它投入到稍微复杂一点的场景,比如需要联网搜索、调用多个API、处理长文本分析时,问题就接踵而至了。Agent 可能会陷入某个查询的无限循环,或者在某个工具调用失败后直接“卡死”,更别提精细地控制 API 调用成本(预算)了。
这个项目标题提出的核心思想,正是对这种原始模式的升级。它主张将 Agent 的运行逻辑,从一段脆弱的、内聚的循环代码,提升为一套由协议(Protocol)驱动的、显式声明式的系统。这里的“协议”,并非特指某种网络通信协议,而是一种规范或契约,它明确定义了 Agent 在运行过程中必须遵守的规则,特别是关于预算(Budget)、工具(Tools)和失败状态(Failure States)的管理。这就像是从“人治”走向“法治”,Agent 不再仅仅依赖代码逻辑的偶然正确性,而是在一个清晰规则的框架内自主、可靠地运行。
2. 核心需求解析:为什么简单的while循环不够用?
要理解为什么需要引入协议,我们必须先拆解一个典型while循环式 Agent 的局限性。假设我们要开发一个“智能研究助手”Agent,它的任务是:根据用户提出的一个开放性问题(例如“比较 TensorFlow 和 PyTorch 在分布式训练方面的最新进展”),自动进行网络搜索、阅读并总结相关文章,最终生成一份报告。
2.1while循环模式的典型缺陷
在一个朴素的实现中,代码骨架可能如下:
# 伪代码示例:一个脆弱的 while 循环 Agent def naive_research_agent(question): context = question max_steps = 20 # 一个硬编码的“保险丝” step = 0 while not task_is_done(context) and step < max_steps: step += 1 # 1. 调用大模型,决定下一步行动 llm_response = call_llm(prompt=context, tools=[search, read, summarize]) action = parse_llm_response(llm_response) # 2. 执行行动 if action.name == “search_web”: result = search_web(action.arguments[“query”]) context += result elif action.name == “read_url”: result = fetch_and_extract(action.arguments[“url”]) context += result elif action.name == “final_answer”: return action.arguments[“answer”] else: # 未知动作?可能陷入困惑循环 context += “I got confused.”这个模式至少存在以下几个致命问题:
- 预算失控:每一次
call_llm和search_web都可能产生费用或消耗资源。循环中没有成本追踪机制。Agent 可能会为了一个细节进行数十次昂贵的搜索和长上下文推理,导致账单爆炸。max_steps只是一个生硬的停止阀,无法区分“有效步骤”和“无效循环”。 - 工具调用无约束:所有工具对 Agent 都是平等的、无限可用的。一个“恶意”或出错的提示可能导致 Agent 疯狂调用某个高负载或高成本的工具(例如,反复调用一个生成高清图像的接口)。
- 失败处理缺失:如果
search_web因网络问题失败,或者read_url遇到一个无法解析的页面,代码通常只是简单地捕获异常,然后将错误信息塞回context。大模型下次看到错误信息,可能会做出更不可预测的决策,甚至开始尝试“修复”不存在的错误,导致状态混乱。 - 状态管理隐式:任务是否完成 (
task_is_done)、当前进展如何,这些状态都隐含在拼接的context字符串和循环条件中。当流程复杂后,这种隐式状态极难调试和监控。
2.2 协议化管理的核心需求
因此,我们需要一套显式的协议来管理这些方面:
- 预算协议:定义 Agent “钱包”的规则。例如,总预算 2 美元,其中 LLM 调用每次不超过 0.1 美元,搜索工具每次 0.01 美元。每执行一步,就从相应预算池扣除。当某个池子耗尽,协议应禁止 Agent 再发起该类操作,并触发降级策略(如使用缓存、返回预算不足信息)。
- 工具协议:定义每个工具的“使用说明书”和“安全守则”。不仅包括函数签名,还包括调用频率限制、前置条件、后置状态影响以及失败时的标准返回值格式。工具调用不再是简单的函数执行,而是一次受协议约束的交互。
- 失败状态协议:定义一套清晰的、机器可读的状态码和状态转移图。例如,
TOOL_CALL_NETWORK_ERROR,LLM_RESPONSE_MALFORMED,BUDGET_EXCEEDED。当失败发生时,Agent 不是将错误文本扔回上下文,而是根据协议更新自己的内部状态机,并依据协议决定下一步是重试、切换工具、请求人工干预,还是优雅失败。
3. 设计思路:构建一个基于协议的 Agent 内核
将预算、工具和失败状态写进协议,意味着我们要重新设计 Agent 的核心执行引擎。它不再是一个过程式的循环,而是一个事件驱动的状态机,由协议规则来裁决每一步的可行性。
3.1 系统架构设计
一个基于协议的可控 Agent 系统可以抽象为以下核心组件:
[用户请求] -> [协议解析器] -> [可控执行引擎] -> [最终结果] | | [预算管理器] [工具路由与执行器] [状态管理器] [失败处理中间件] | | [协议定义文件] [工具协议定义库]- 协议定义文件:通常是一个结构化的配置文件(如 YAML、JSON 或 Python Dataclass)。它声明了本次任务运行的“宪法”。
- 协议解析器:加载协议文件,初始化预算管理器、状态管理器,并向执行引擎注册所有受协议约束的工具。
- 可控执行引擎:这是替代
while循环的核心。它管理着一个主循环,但在每一步迭代中,它需要:- 咨询状态管理器:当前是否处于可执行状态?有无未处理的致命错误?
- 咨询预算管理器:请求的 LLM 调用和工具调用是否在预算内?
- 调用决策模块(通常是 LLM):根据当前状态和约束,生成下一个动作意图。
- 将动作意图交给工具路由与执行器:该执行器会检查此工具调用是否符合其协议(频率、参数),然后执行。执行结果(成功或失败)被标准化后,提交给失败处理中间件。
- 失败处理中间件根据失败类型和协议中定义的策略,决定是重试、更新状态为错误,还是触发降级流程。
- 更新状态管理器,并决定循环是否继续。
3.2 协议内容的具体定义
下面以一个简化的 YAML 协议示例,展示如何将关键约束写进协议:
# agent_protocol.yaml version: “1.0” task: “research_assistant” budget: total: 2.0 # USD allocations: llm: max_per_call: 0.05 provider: “openai” model: “gpt-4-turbo” tool: web_search: cost_per_call: 0.002 fetch_url: cost_per_call: 0.001 tools: web_search: description: “Search the web for current information.” spec: # 函数签名等 constraints: max_calls_per_task: 10 rate_limit: “1 call per 2 seconds” allowed_domains: [“*.arxiv.org”, “*.github.com”, “*.towardsdatascience.com”] # 限制搜索范围,控制质量与安全 on_failure: - code: “NETWORK_ERROR” retry_policy: {max_attempts: 2, backoff_factor: 1.5} fallback_action: “use_cached_if_available” - code: “RATE_LIMITED” retry_policy: {max_attempts: 1, backoff_seconds: 60} fallback_action: “pause_and_continue” fetch_url: description: “Fetch and extract main content from a URL.” spec: # ... constraints: timeout_seconds: 10 on_failure: - code: “TIMEOUT” retry_policy: {max_attempts: 1} fallback_action: “mark_unreachable_and_continue” failure_states: definitions: - code: “BUDGET_EXCEEDED” level: “FATAL” message: “The allocated budget for {resource} has been exhausted.” - code: “TOOL_CALL_INVALID” level: “ERROR” message: “Tool call did not meet protocol constraints.” - code: “MAX_ITERATIONS_REACHED” level: “WARNING” message: “Agent stopped after maximum allowed steps.” escalation_policy: on_fatal: “stop_and_report” # 停止并向上游报告 on_error: “retry_or_degrade” # 根据工具协议决定 on_warning: “continue_with_log” # 记录日志但继续执行这个协议文件清晰地定义了资源边界、行为规范和应急流程,使得 Agent 的行为变得可预测、可审计、可管理。
4. 核心实现:打造可控执行引擎
理解了设计思路后,我们进入实现环节。我们将用 Python 构建一个简化但功能完整的可控 Agent 执行引擎。
4.1 定义核心数据模型
首先,我们需要定义承载协议和状态的核心数据类。
from dataclasses import dataclass, field from enum import Enum from typing import Any, Dict, List, Optional, Callable import time class FailureLevel(Enum): WARNING = “warning” ERROR = “error” FATAL = “fatal” class FailureCode(Enum): # 预算相关 BUDGET_EXCEEDED_LLM = “budget_exceeded_llm” BUDGET_EXCEEDED_TOOL = “budget_exceeded_tool” # 工具相关 TOOL_CALL_NETWORK_ERROR = “tool_call_network_error” TOOL_CALL_TIMEOUT = “tool_call_timeout” TOOL_CALL_RATE_LIMITED = “tool_call_rate_limited” TOOL_CALL_INVALID = “tool_call_invalid” # 参数错误等 # 流程相关 LLM_RESPONSE_MALFORMED = “llm_response_malformed” MAX_ITERATIONS = “max_iterations” @dataclass class FailureState: code: FailureCode level: FailureLevel message: str tool_name: Optional[str] = None timestamp: float = field(default_factory=time.time) metadata: Dict[str, Any] = field(default_factory=dict) @dataclass class Budget: total: float allocated: Dict[str, float] # e.g., {“llm”: 1.0, “tool_web_search”: 0.5} spent: Dict[str, float] = field(default_factory=lambda: {“llm”: 0.0, “tool”: 0.0}) def can_spend(self, category: str, amount: float) -> bool: category_spent = self.spent.get(category, 0.0) category_allocated = self.allocated.get(category, 0.0) return category_spent + amount <= category_allocated def spend(self, category: str, amount: float) -> bool: if self.can_spend(category, amount): self.spent[category] = self.spent.get(category, 0.0) + amount return True return False @dataclass class ToolConstraint: max_calls: Optional[int] = None rate_limit: Optional[float] = None # seconds between calls last_called: Optional[float] = None def can_call(self) -> tuple[bool, Optional[str]]: “”“检查是否允许调用。返回 (是否允许, 拒绝原因)”“” if self.max_calls is not None and self.call_count >= self.max_calls: return False, f“Max calls ({self.max_calls}) exceeded” if self.rate_limit is not None and self.last_called is not None: if time.time() - self.last_called < self.rate_limit: return False, f“Rate limit ({self.rate_limit}s) not met” return True, None4.2 实现协议加载与状态管理
接下来,我们实现协议加载器和核心的状态管理器。
import yaml class ProtocolLoader: @staticmethod def load_from_yaml(path: str) -> Dict[str, Any]: with open(path, ‘r’) as f: return yaml.safe_load(f) class StateManager: def __init__(self, protocol: Dict[str, Any]): self.protocol = protocol self.failure_states: List[FailureState] = [] self.current_step = 0 self.max_steps = protocol.get(“execution”, {}).get(“max_steps”, 50) self._is_fatal = False def record_failure(self, failure: FailureState): self.failure_states.append(failure) if failure.level == FailureLevel.FATAL: self._is_fatal = True # 这里可以添加日志上报、监控指标上报等 def can_continue(self) -> bool: if self._is_fatal: return False if self.current_step >= self.max_steps: self.record_failure(FailureState( code=FailureCode.MAX_ITERATIONS, level=FailureLevel.WARNING, message=f“Reached max steps ({self.max_steps})” )) return False return True def increment_step(self): self.current_step += 14.3 实现受协议约束的工具执行器
这是将工具调用从自由函数升级为受控操作的关键。
class ToolExecutor: def __init__(self, protocol: Dict[str, Any], budget_manager: ‘BudgetManager’, state_manager: StateManager): self.protocol_tools = protocol.get(“tools”, {}) self.budget_manager = budget_manager self.state_manager = state_manager self.tool_registry: Dict[str, Callable] = {} self.tool_constraints: Dict[str, ToolConstraint] = {} self._init_tools() def _init_tools(self): # 这里应该从协议中加载工具的实际实现函数 # 为简化,我们假设工具函数已注册到 self.tool_registry for tool_name, tool_spec in self.protocol_tools.items(): constraint_spec = tool_spec.get(“constraints”, {}) self.tool_constraints[tool_name] = ToolConstraint( max_calls=constraint_spec.get(“max_calls_per_task”), rate_limit=constraint_spec.get(“rate_limit_seconds”), ) def execute(self, tool_name: str, arguments: Dict[str, Any]) -> Dict[str, Any]: # 1. 检查工具是否存在 if tool_name not in self.tool_registry or tool_name not in self.protocol_tools: failure = FailureState( code=FailureCode.TOOL_CALL_INVALID, level=FailureLevel.ERROR, message=f“Tool ‘{tool_name}’ is not registered or defined in protocol.”, tool_name=tool_name ) self.state_manager.record_failure(failure) return {“success”: False, “error”: failure.message} # 2. 检查协议约束 constraint = self.tool_constraints.get(tool_name) if constraint: can_call, reason = constraint.can_call() if not can_call: failure = FailureState( code=FailureCode.TOOL_CALL_INVALID, level=FailureLevel.ERROR, message=f“Tool call constraint violated: {reason}”, tool_name=tool_name ) self.state_manager.record_failure(failure) return {“success”: False, “error”: failure.message} # 3. 检查并扣除预算 tool_spec = self.protocol_tools[tool_name] cost = tool_spec.get(“cost_per_call”, 0.0) budget_category = f“tool_{tool_name}” if not self.budget_manager.spend(budget_category, cost): failure = FailureState( code=FailureCode.BUDGET_EXCEEDED_TOOL, level=FailureLevel.FATAL, # 工具预算耗尽设为FATAL,可依据协议调整 message=f“Budget exceeded for tool ‘{tool_name}’.”, tool_name=tool_name ) self.state_manager.record_failure(failure) return {“success”: False, “error”: failure.message} # 4. 执行工具 try: tool_func = self.tool_registry[tool_name] result = tool_func(**arguments) # 更新约束状态,如最后调用时间 if constraint: constraint.last_called = time.time() constraint.call_count = getattr(constraint, ‘call_count’, 0) + 1 return {“success”: True, “data”: result} except Exception as e: # 5. 处理执行失败 failure_code = FailureCode.TOOL_CALL_NETWORK_ERROR # 根据异常类型细化 failure = FailureState( code=failure_code, level=FailureLevel.ERROR, message=f“Tool ‘{tool_name}’ execution failed: {str(e)}”, tool_name=tool_name, metadata={“exception”: str(e)} ) self.state_manager.record_failure(failure) # 这里可以加入协议中定义的 on_failure 重试逻辑 return {“success”: False, “error”: failure.message}4.4 组装可控执行引擎
最后,我们将所有组件组装成完整的执行引擎。
class ControlledAgentEngine: def __init__(self, protocol_path: str, llm_client, registered_tools: Dict[str, Callable]): self.protocol = ProtocolLoader.load_from_yaml(protocol_path) self.budget_manager = BudgetManager(self.protocol.get(“budget”, {})) self.state_manager = StateManager(self.protocol) self.tool_executor = ToolExecutor(self.protocol, self.budget_manager, self.state_manager) self.llm_client = llm_client # 注册工具 self.tool_executor.tool_registry.update(registered_tools) def run(self, initial_query: str) -> Dict[str, Any]: context = {“query”: initial_query, “history”: []} result = None while self.state_manager.can_continue(): self.state_manager.increment_step() # 1. 检查LLM调用预算 llm_cost_estimate = 0.02 # 简化估算 if not self.budget_manager.spend(“llm”, llm_cost_estimate): self.state_manager.record_failure(FailureState( code=FailureCode.BUDGET_EXCEEDED_LLM, level=FailureLevel.FATAL, message=“LLM budget exhausted.” )) break # 2. 调用LLM进行决策 try: llm_response = self._call_llm_for_decision(context) action = self._parse_llm_response(llm_response) except Exception as e: self.state_manager.record_failure(FailureState( code=FailureCode.LLM_RESPONSE_MALFORMED, level=FailureLevel.ERROR, message=f“Failed to get or parse LLM response: {e}” )) # 可以尝试修复或重试,这里简单跳出 break # 3. 执行动作 if action[“type”] == “tool_call”: tool_result = self.tool_executor.execute(action[“tool”], action[“args”]) context[“history”].append({“action”: action, “result”: tool_result}) if tool_result[“success”]: # 将成功结果融入上下文,供下次LLM决策 context[“latest_data”] = tool_result[“data”] else: # 失败结果也记录,LLM可以学习处理失败 context[“latest_error”] = tool_result[“error”] # 根据失败严重程度,可能触发状态转移 # 这里可以根据 protocol[‘failure_states’][‘escalation_policy’] 实现复杂逻辑 elif action[“type”] == “final_answer”: result = action[“answer”] break # 任务成功完成 else: # 未知动作类型 self.state_manager.record_failure(FailureState( code=FailureCode.LLM_RESPONSE_MALFORMED, level=FailureLevel.ERROR, message=f“Unknown action type: {action[‘type’]}” )) # 运行结束,整理返回 return { “success”: result is not None, “result”: result, “final_state”: { “steps”: self.state_manager.current_step, “budget_spent”: self.budget_manager.get_spent_summary(), “failures”: [fs.__dict__ for fs in self.state_manager.failure_states] } } def _call_llm_for_decision(self, context): # 构建包含协议约束(如剩余预算、可用工具列表)的提示词 prompt = self._build_prompt(context) return self.llm_client.chat_completion(prompt) def _parse_llm_response(self, response): # 解析LLM返回的JSON或文本,提取动作指令 # 这里应有严格的校验逻辑 pass5. 实战应用与避坑指南
有了可控引擎,我们来看看如何在实际项目中应用,并分享一些关键的避坑经验。
5.1 应用场景:智能客服工单升级
假设我们有一个用于初步处理用户技术问题的客服 Agent。协议可以这样设计:
- 预算:限制每张工单最多进行 5 次 LLM 交互和 3 次知识库查询。
- 工具协议:
search_knowledge_base: 最多调用 3 次,每次间隔 1 秒。escalate_to_human: 这是一个特殊工具,调用后工单状态直接变为“待人工处理”。协议规定,当连续两次search_knowledge_base返回的结果置信度低于阈值时,必须调用此工具。
- 失败状态协议:
- 知识库连接失败 -> 重试1次后,若仍失败,状态置为
SYSTEM_ERROR,直接调用escalate_to_human。 - LLM 返回无法解析 -> 状态置为
AGENT_CONFUSED,同样触发升级。
- 知识库连接失败 -> 重试1次后,若仍失败,状态置为
这样,Agent 就能在明确的规则下运作:它会在有限的资源内尝试自助解决,一旦触及边界或遇到特定失败,就严格按照协议将问题转交给人,避免了 AI 在无法解决的问题上“空转”或给出错误承诺。
5.2 避坑心得与注意事项
- 协议不是银弹,设计需权衡:过于严格的协议会扼杀 Agent 的灵活性。例如,将每个工具的
max_calls设得太低,可能导致任务无法完成。我的经验是从宽松开始,逐步收紧。先在日志中监控工具的使用模式,观察哪些工具被频繁调用、哪些很少用,再基于数据来设定合理的限制。 - 预算估算的准确性:LLM 调用的成本随输入/输出 token 数变化,很难在协议中精确指定
max_per_call。一个实用的方法是使用“信用点”系统。例如,给 LLM 预算分配 1000 点,定义每 1000 个输入 token 消耗 1 点,每 1000 个输出 token 消耗 2 点。在执行前,根据当前上下文长度预估本次调用的点数消耗并进行检查。虽然不精确,但比固定金额更合理。 - 失败状态的处理是难点:定义清晰的状态码只是第一步。关键在于设计状态转移逻辑。一个
TOOL_CALL_TIMEOUT错误,是应该立即重试,还是标记该工具暂时不可用并尝试替代方案?这部分逻辑可以写在协议的on_failure段,也可以实现在StateManager中作为一个规则引擎。避免在核心引擎代码中写死大量的if-else。 - 协议的可观测性:必须建立完善的日志和监控。每一次预算扣除、工具调用(无论成功失败)、状态变更,都应该有结构化日志。这不仅能帮助调试,更是后期优化协议参数(如调整预算分配、修改重试策略)的唯一依据。可以考虑将运行时的状态和指标暴露给一个仪表盘。
- 与现有框架的集成:如果你在使用 LangChain 或 LlamaIndex 等框架,不必完全重写。可以将它们视为“工具实现层”和“LLM 调用层”。在其外层包裹我们实现的协议控制层。例如,LangChain 的 AgentExecutor 在执行前,先经过我们的
BudgetManager和StateManager检查;其工具调用的结果,先由我们的ToolExecutor进行协议合规性包装和失败处理,再返回给 Agent。这是一种非侵入式的增强。
6. 总结与展望
将 Agent 从while循环升级为协议驱动的系统,本质上是在引入控制论中的反馈与调节机制。预算、工具约束和失败状态协议,共同构成了一个多维度的“控制器”,持续监测 Agent 的运行状态(如花费、调用次数、错误码),并与预设的“设定点”(协议规则)进行比较,一旦偏差超出允许范围,就触发纠正动作(如停止、降级、上报)。
这种模式带来的最大好处是系统的可预测性和可运维性。作为开发者,你不再需要去浩如烟海的日志中猜测 Agent 为什么花了那么多钱或者卡住了。你查看最终的状态报告:BUDGET_EXCEEDED_LLM、TOOL_CALL_RATE_LIMITED,问题一目了然。作为系统管理者,你可以通过动态更新协议文件,来调整一群 Agent 的资源分配和策略,而不需要修改代码。
未来的方向,我认为协议会变得更加动态和智能。例如,协议本身可以由一个“元 Agent”根据历史运行数据和当前系统负载自动生成和调整。或者,Agent 在运行中遇到协议未定义的边缘情况时,可以主动发起一个“协议修订请求”,在人类监督下进行安全的学习和迭代。可控,是 AI Agent 走向大规模、高可靠生产应用的必经之路,而将其核心规则抽象为显式的协议,正是实现这种可控性的坚实一步。
