最近在推进 Agent 类项目时,我发现最让人头疼的其实不是 Agent 本身怎么写,而是底层的 Infra 到底应该怎么搭。今天选一套自研 Agent 执行引擎,明天团队又想切更抽象的多 Agent 编排框架,后天产品又要求给每个 Agent 挂上记忆和工具集。Agent 的形态更新频率比业务需求还快,下游的存储、模型网关、工具执行、日志链路全都要跟着变。这篇文章想围绕“Agent 形态频变,Infra 到底该为谁而建”这个问题,梳理一套不绑定具体框架的 Agent 基础设施设计思路,包含概念拆解、最小可运行示例和常见问题排查,适合正在做 Agent 落地或准备搭建 Agent 工程基座的开发者阅读。
1. 背景与核心概念
1.1 先回答一个基础问题:Agent 是什么
Agent 通常指能够感知环境、做出决策并执行动作的智能程序。在 LLM 语境下,Agent 一般具备以下能力:
- 接收自然语言目标。
- 调用大模型进行推理。
- 通过工具调用访问外部系统。
- 根据执行结果调整下一步计划。
- 必要时维护短期或长期记忆。
和传统“一问一答”的 Chatbot 不同,Agent 的核心特征是有“执行闭环”。大模型只负责规划,真正的落地动作由工具完成。这就决定了 Agent 项目不仅仅是写 Prompt,还需要处理模型路由、上下文管理、工具协议、权限控制、任务追踪等一系列工程问题。
1.2 为什么说 Agent 形态一天一个样
从技术演进看,Agent 工程化的步伐非常快,形态不断变化:
- 早先的 Agent 以“Function Calling + 单轮工具调用”为主。
- 随后出现 ReAct 风格的“思考—行动—观察”循环。
- 再往后是记忆管理、多 Agent 协作、Agent 编排框架。
- 现在又引入了 Skill、Harness、MCP 等概念,让 Agent 的能力边界从“单模型对话”扩展到“模型 + 工具 + 记忆 + 策略”的组合体。
每个阶段都会出现新的框架和术语,比如 Agent 框架、Agent 编排、Agent 记忆、Agent 安全、Agent 部署、多 Agent 协作、Agent Loop、Agent Skill、Harness、MCP。这些概念之间既有重叠,也有差异化定位。对一线开发来说,最直接的感受就是:今天基于某框架写的 Agent,明天可能就要迁移到另一个框架。如果 Infra 和框架强绑定,改造成本就会非常高。
1.3 Infra 在 Agent 语境下指什么
Infra 不只是服务器和 Kubernetes,在 Agent 项目中更值得关注的是“面向智能应用的软件基础设施”,例如:
- 模型网关和统一 API 接入层。
- 上下文存储、向量数据库、长期记忆组件。
- 工具注册与调用协议。
- 任务调度与执行环境。
- 可观测性、日志追踪、评估体系。
- 安全与权限控制。
这些能力应当被看成 Agent 的“底座”,它服务于任意形态的 Agent,而不是服务于某一个具体框架。
1.4 Agent 形态和 Infra 的关系
Agent 是上层的业务逻辑,形态可以频繁变化;Infra 是下层的通用能力,应当保持相对稳定。
如果把 Infra 也绑在某一种 Agent 形态上,那么当 Agent 形态切换时,Infra 也要跟着重构。更合理的做法是:将 Agent 的“能力要素”抽象成独立的模块,让上层 Agent 通过统一接口调用能力,而不是直接操作底层实现。
打个比方,Agent 形态像不断更新的手机应用,Infra 像手机操作系统和硬件能力。应用可以换,但系统不能每次都重写。
2. Agent 形态演进的四个阶段
2.1 阶段一:单轮工具调用
最早期也是最简单的 Agent 形态,是让模型决定是否调用某个工具,并把工具结果回填给模型,然后生成最终回答。
关键流程:
- 用户输入。
- 模型识别意图。
- 代码调用对应函数。
- 模型基于函数结果生成回复。
这种形态适合“天气查询、计算器、表单填充”等单步骤任务。它的优点是实现简单,缺点是模型只能做一次决策,无法处理多步骤任务。
2.2 阶段二:Agent Loop 与 ReAct
真正意义上的 Agent 开始于循环执行。模型不再“一次决定”,而是重复执行“推理—行动—观察”的循环,直到问题解决或达到最大轮数。
这里会引入几个典型工程点:
- 最大迭代次数限制。
- 上下文裁剪。
- 工具报错后的重试策略。
- 循环中止条件。
常见报错如:
the agent execution provider did not respond in time. this may indicate the agent execution provider is overloaded or deadlocked.以及:
agent terminated due to error. you can prompt the model to try again or start a new conversation.这类错误通常发生在 Agent 循环执行过程中,根因可能是指令死循环、工具超时、上下文过长,也可能是执行环境内存耗尽。后面会在常见问题部分展开。
2.3 阶段三:记忆与多 Agent 协作
当任务复杂到需要长期保持状态,Agent 就需要记忆能力。记忆通常分为:
- 短期记忆:当前会话内的上下文。
- 长期记忆:跨会话保留的用户偏好、事实信息、历史结果。
- 工作记忆:当前任务执行过程中的临时状态。
多 Agent 协作则把一个大任务拆分为多个子任务,由不同的 Agent 分别负责。这种形态在提升复杂任务完成率的同时,也带来新的 Infra 挑战:
- Agent 之间如何通信。
- 是否需要消息队列。
- 共享上下文如何并发读写。
- 全局状态的一致性如何保证。
2.4 阶段四:Skill、Harness 与 MCP
最近讨论较多的几个概念需要区分:
- Agent Skill:面向 Agent 的能力包,通常描述“什么场景下使用什么能力、怎么调用”。
- Agent Scope:Agent 可以访问的资源范围或权限边界。
- Harness:Agent 的运行框架,负责把模型、工具、记忆、策略组合成一个可执行的整体。
- MCP(Model Context Protocol):一种模型上下文协议,目标是让工具和服务以标准化方式接入 Agent。
通俗理解:Skill 偏向“能力定义”,Harness 偏向“运行时框架”,MCP 偏向“工具交互协议”。它们解决的是 Agent 生态的连接问题,目的是让同一套 Infra 能力被不同 Agent 复用。
2.5 形态变化背后的不变要素
无论 Agent 形态怎么变,以下能力始终需要存在:
- 大模型统一访问。
- 上下文管理。
- 工具注册与调用。
- 任务执行状态。
- 日志追踪。
- 权限与安全控制。
Infra 的建设核心,就是把这六类能力沉淀为通用服务。这样上层 Agent 无论采用什么框架,接入时都只需要关注业务逻辑,而不是重新建设底层能力。
| Agent 形态 | 核心关注点 | Infra 重点关注 |
|---|---|---|
| 单轮工具调用 | 函数调用准确率 | 模型网关、工具返回协议 |
| ReAct 循环 | 推理步骤控制 | 执行循环、上下文窗口管理 |
| 记忆型 Agent | 长期状态保持 | 向量库、缓存、记忆读写接口 |
| 多 Agent 协作 | 任务拆分与通信 | 消息队列、状态一致性、并发控制 |
| Skill / MCP 生态 | 能力标准化接入 | 工具注册中心、协议适配、权限隔离 |
3. Infra 到底该为谁而建
3.1 不应该为某个框架建
如果 Infra 只面向某一个 Agent 框架,框架升级时 Infra 就要跟着改,Agent 迁移时 Infra 也要重写。框架迭代速度快,甚至可能半年内就出现新的替代方案,Infra 如果绑死在框架上,等于把地基建在流沙上。
3.2 也不应该为某一个大模型建
模型 API 在快速迭代,不同模型的能力、上下文长度、工具调用方式都有差异。Infra 如果直接依赖某家模型 SDK 的数据结构,模型切换成本会很高。更合理的方式是定义一套统一的模型调用接口,底层适配不同模型。
3.3 应该为 Agent 的“能力要素”建
Infra 的真正服务对象,是 Agent 运行过程中反复需要的能力要素:
- 模型接入:让不同 Agent 都能方便地访问 LLM。
- 上下文管理:让 Agent 能读取、更新、裁剪上下文。
- 工具调用:让 Agent 能发现工具、调用工具、获取结果。
- 状态存储:让 Agent 能保存会话和长期记忆。
- 可观测性:让开发者能看到 Agent 在“想什么、调用了什么、卡在哪里”。
- 安全与权限:让 Agent 只能访问授权范围内的资源。
只要 Infra 围绕这些能力要素建设,Agent 上层形态无论怎么变,底座都能稳定支撑。
3.4 从具体场景反推 Infra 需求
假设你要做一个自动运维助手,Agent 通过调用监控 API 排查问题。如果只是写一个 Demo,直接在代码里请求监控接口就行。但如果要做成产品,你会遇到:
- 切换模型时,所有 Prompt 和 Tool 定义重写。
- 工具从 5 个扩展到 50 个,注册逻辑变得混乱。
- 会话状态存在内存里,服务重启就丢。
- 多个 Agent 并发执行任务,日志互相覆盖。
- 某个 Agent 误操作触发高危命令,缺少熔断机制。
这些问题都不是 Agent 业务代码本身能解决的,而是 Infra 层面需要提供的通用能力。
3.5 化繁为简:Infra 优先做“接口稳定”
Infra 建设的核心目标,是让上层变化可被隔离,让下层能力可被复用。用接口抽象替代直接实现,用配置声明替代代码硬编码,用协议标准化替代各写一套。
4. 环境准备与版本说明
4.1 基础环境
本文的示例使用 Python 编写,重点演示 Agent 运行时与底层能力的解耦思路。如果你所在团队使用 Java 或 Go,设计思路同样适用,只是代码语言不同。
示例运行环境大致如下:
- Python 3.10 或更高版本。
- 一个 OpenAI 兼容的模型 API 服务,地址和密钥通过配置传入。
- 本地数据库或消息队列并非强制要求,但建议准备一个 Redis 实例用于验证记忆和状态存储。
- 依赖管理工具:建议使用 venv 或 poetry。
版本需要根据你的项目实际情况调整,本文示例以常见环境为例,重点演示配置思路。不同依赖之间的版本兼容性需要以官方文档为准,不要直接照搬老旧教程中的版本组合。
4.2 项目结构
先规划一个最小项目结构:
agent-infra-demo/ ├── config.yaml ├── requirements.txt ├── agent_infra/ │ ├── __init__.py │ ├── model.py │ ├── context.py │ ├── tools.py │ ├── memory.py │ └── runtime.py ├── main.py └── tools/ ├── __init__.py └── echo_tool.py其中:
- model.py:统一模型访问接口。
- context.py:上下文管理。
- tools.py:工具注册与调用。
- memory.py:记忆存储抽象。
- runtime.py:Agent 执行循环。
这个结构刻意把“能力要素”拆成独立模块,上层 main.py 只写业务逻辑,不再关心具体模型和工具实现。
4.3 requirements.txt
以下依赖只用于演示:
openai>=1.0.0 PyYAML>=6.0 redis>=5.0.0 pydantic>=2.0.0安装命令:
pip install -r requirements.txt实际项目中请根据你的模型服务商和存储组件选择对应 SDK,不要盲目保持与本文一致。
5. 从零搭建一个可演进的 Agent Infra 最小示例
5.1 设计目标
本示例要体现以下能力:
- 模型接入可替换。
- 工具注册可扩展。
- 上下文和记忆可管理。
- Agent 执行循环与具体模型解耦。
也就是说,当你把 OpenAI 兼容接口换成其他模型服务时,只需要改配置;当你新增工具时,只需要注册一个新函数;当你的 Agent 形态从单 Agent 变成多 Agent 时,底层的模型和工具模块可以被多个执行器复用。
5.2 配置管理
先编写 config.yaml:
model: provider: openai-compatible base_url: https://api.example.com/v1 api_key: ${MODEL_API_KEY} model_name: demo-agent-model temperature: 0.2 max_tokens: 2048 runtime: max_steps: 10 request_timeout: 60 memory: type: redis host: localhost port: 6379 db: 0这里有一个关键设计:所有敏感信息通过环境变量注入,而不是直接写在配置文件里。api_key 使用${MODEL_API_KEY}占位,运行时从环境变量读取。
5.3 统一模型访问接口
在 model.py 中定义统一的模型调用方法:
# 文件路径:agent_infra/model.py import os from dataclasses import dataclass try: from openai import OpenAI except ImportError: OpenAI = None @dataclass class ModelConfig: base_url: str api_key: str model_name: str temperature: float = 0.2 max_tokens: int = 2048 class BaseModelClient: """统一模型客户端,上层 Agent 只依赖这个接口。""" def __init__(self, config: ModelConfig): self.config = config def chat(self, messages: list) -> str: raise NotImplementedError def chat_with_tools(self, messages: list, tools: list) -> dict: raise NotImplementedError class OpenAIClient(BaseModelClient): """OpenAI 兼容协议实现。""" def __init__(self, config: ModelConfig): super().__init__(config) if OpenAI is None: raise RuntimeError("openai SDK is not installed") self.client = OpenAI( base_url=config.base_url, api_key=config.api_key, ) def chat(self, messages: list) -> str: resp = self.client.chat.completions.create( model=self.config.model_name, messages=messages, temperature=self.config.temperature, max_tokens=self.config.max_tokens, ) return resp.choices[0].message.content or "" def chat_with_tools(self, messages: list, tools: list) -> dict: resp = self.client.chat.completions.create( model=self.config.model_name, messages=messages, temperature=self.config.temperature, max_tokens=self.config.max_tokens, tools=tools, ) message = resp.choices[0].message return { "content": message.content or "", "tool_calls": message.tool_calls, } def create_model_client(config: ModelConfig) -> BaseModelClient: """简单工厂方法,后续可以按配置扩展其他实现。""" if config.base_url.lower().startswith("http"): return OpenAIClient(config) raise ValueError(f"unsupported model provider: {config.base_url}")这段代码解决的核心问题是:上层 Agent 不直接调用 OpenAI SDK,而是调用 BaseModelClient。如果后续需要接入其他模型服务,只需要新增一个子类,并在 create_model_client 中增加分支。
5.4 上下文管理
Agent 循环中,上下文会不断累积。如果不加管理,几轮循环后 tokens 就会超出模型限制。
实现一个简单版本:
# 文件路径:agent_infra/context.py class AgentContext: """管理单个 Agent 执行过程中的消息列表和总长度估算。""" def __init__(self, system_prompt: str, max_tokens: int = 8000): self.system_prompt = system_prompt self.max_tokens = max_tokens self.messages = [ {"role": "system", "content": system_prompt} ] def add_user_message(self, content: str): self.messages.append({"role": "user", "content": content}) def add_assistant_message(self, content: str): self.messages.append({"role": "assistant", "content": content}) def add_tool_result(self, tool_call_id: str, result: str): self.messages.append({ "role": "tool", "tool_call_id": tool_call_id, "content": result, }) def get_trimmed_messages(self, max_messages: int = 30): """简单裁剪:保留 system 和最近的消息。""" if len(self.messages) <= max_messages: return self.messages tail = self.messages[-(max_messages - 1):] return [self.messages[0]] + tail @property def current_messages(self) -> list: return self.messages注意,这里的 token 估算比较粗略,只是演示“有上下文管理”这一个概念。生产环境应该使用 tokenizer 精确统计,并根据模型窗口动态裁剪。
5.5 工具注册与调用
工具注册是 Agent Infra 的核心模块。我建议用装饰器模式把工具定义和函数实现放在一起,减少重复代码。
# 文件路径:agent_infra/tools.py import inspect import json from typing import Callable, Dict class ToolRegistry: """工具注册中心:统一管理 Agent 可调用的工具。""" def __init__(self): self._tools: Dict[str, Callable] = {} def register(self, func: Callable, description: str = ""): name = func.__name__ self._tools[name] = func doc = description or inspect.getdoc(func) or "" # 从函数签名生成简单参数信息,便于模型理解。 parameters = { "type": "object", "properties": {}, "required": [], } for key, value in inspect.signature(func).parameters.items(): if value.annotation is not inspect.Parameter.empty: parameters["properties"][key] = { "type": "string", "description": str(value.annotation), } else: parameters["properties"][key] = { "type": "string", "description": key, } if value.default is inspect.Parameter.empty: parameters["required"].append(key) self._tools_meta[name] = { "type": "function", "function": { "name": name, "description": doc, "parameters": parameters, }, } return func def add_tool(self, func: Callable, meta: dict): name = func.__name__ self._tools[name] = func self._tools_meta[name] = meta return func def tool_schema(self) -> list: return list(self._tools_meta.values()) def execute(self, name: str, arguments: dict): if name not in self._tools: raise ValueError(f"unknown tool: {name}") return self._tools[name](**arguments) # 初始化时创建实例 tool_registry = ToolRegistry() tools_meta = {} def tool(description: str = ""): """装饰器:注册 Agent 工具。""" def decorator(func): tool_registry.add_tool(func, { "type": "function", "function": { "name": func.__name__, "description": description or inspect.getdoc(func) or "", "parameters": { "type": "object", "properties": { name: {"type": "string", "description": name} for name in inspect.signature(func).parameters }, "required": [ name for name, param in inspect.signature(func).parameters.items() if param.default is inspect.Parameter.empty ], }, }, }) return func return decorator这里需要说明一个实现细节:初始化tool_registry = ToolRegistry()时,其实应该同步初始化_tools_meta。上面的示例中tool_registry依赖ToolRegistry内部状态。为了简洁,可以这样写:
class ToolRegistry: def __init__(self): self._tools = {} self._tools_meta = {} def add_tool(self, func, meta): self._tools[func.__name__] = func self._tools_meta[func.__name__] = meta return func实际使用中,装饰器会把工具定义和实现绑定在一起,新增工具非常方便。例如创建一个简单的回显工具:
# 文件路径:tools/echo_tool.py from agent_infra.tools import tool @tool(description="回显输入内容,用于测试工具调用链路。") def echo_tool(message: str) -> str: return f"echo: {message}"5.6 记忆存储抽象
长期记忆是 Agent 演进到记忆型形态的关键能力。这里做一个抽象层,默认提供 Redis 实现:
# 文件路径:agent_infra/memory.py import json from typing import Optional class BaseMemoryStore: def get(self, key: str) -> Optional[dict]: raise NotImplementedError def set(self, key: str, value: dict): raise NotImplementedError class RedisMemoryStore(BaseMemoryStore): def __init__(self, host="localhost", port=6379, db=0): import redis self.client = redis.Redis(host=host, port=port, db=db) def get(self, key: str) -> Optional[dict]: raw = self.client.get(key) if raw is None: return None return json.loads(raw) def set(self, key: str, value: dict): self.client.set(key, json.dumps(value, ensure_ascii=False)) class InMemoryStore(BaseMemoryStore): def __init__(self): self._data = {} def get(self, key: str) -> Optional[dict]: return self._data.get(key) def set(self, key: str, value: dict): self._data[key] = value def create_memory_store(config: dict) -> BaseMemoryStore: if config.get("type") == "redis": return RedisMemoryStore( host=config.get("host", "localhost"), port=config.get("port", 6379), db=config.get("db", 0), ) return InMemoryStore()记忆模块的接口要尽量小,这样无论是 Redis、数据库还是对象存储,都可以通过适配器接入。
5.7 Agent 执行循环
现在把模型、上下文、工具组合成执行循环。这里故意不引入任何特定 Agent 框架,只用手写循环展示 Agent Loop 的本质:
# 文件路径:agent_infra/runtime.py import time import json from typing import Optional from .context import AgentContext from .model import BaseModelClient from .tools import ToolRegistry class AgentRuntime: """轻量级 Agent 执行引擎。""" def __init__( self, model_client: BaseModelClient, tool_registry: ToolRegistry, system_prompt: str, max_steps: int = 10, request_timeout: int = 60, ): self.model_client = model_client self.tool_registry = tool_registry self.system_prompt = system_prompt self.max_steps = max_steps self.request_timeout = request_timeout self.context = AgentContext(system_prompt) def run(self, user_input: str) -> str: self.context.add_user_message(user_input) for step in range(self.max_steps): try: response = self.model_client.chat_with_tools( self.context.get_trimmed_messages(), self.tool_registry.tool_schema(), ) except Exception as exc: return f"agent execution failed: {exc}" tool_calls = response.get("tool_calls") if not tool_calls: # 没有工具调用,说明 Agent 已完成任务。 return response.get("content", "") for tool_call in tool_calls: function_name = tool_call.function.name raw_arguments = tool_call.function.arguments or "{}" try: arguments = json.loads(raw_arguments) except json.JSONDecodeError: arguments = {} self.context.add_assistant_message("") # 执行工具 try: result = self.tool_registry.execute( function_name, arguments ) result_text = json.dumps( {"result": result}, ensure_ascii=False ) except Exception as exc: result_text = json.dumps( {"error": str(exc)}, ensure_ascii=False ) # 把工具结果放回上下文 tool_call_id = getattr(tool_call, "id", str(step)) self.context.add_tool_result(tool_call_id, result_text) time.sleep(0.1) return "agent terminated due to error: max steps exceeded"这个循环和 ReAct 思路一致:
- 模型接收历史消息和工具定义。
- 模型决定是调用工具还是直接回答。
- 如果调用工具,执行并回填结果。
- 继续下一次循环,直到模型直接回答或达到最大步数。
注意,代码中time.sleep(0.1)只是为了让日志更直观,真实项目中不一定需要。超时报错提示“the agent execution provider did not respond in time”时,可以从请求超时设置、模型服务负载、循环死锁三个方向排查。
5.8 主程序编排
main.py 负责读取配置、组装组件、执行 Agent:
# 文件路径:main.py import os import yaml from agent_infra.model import ModelConfig, create_model_client from agent_infra.tools import tool_registry from agent_infra.memory import create_memory_store from agent_infra.runtime import AgentRuntime from tools.echo_tool import echo_tool # noqa: F401 def load_config(path="config.yaml"): with open(path, "r", encoding="utf-8") as f: config = yaml.safe_load(f) # 支持环境变量替换 ${MODEL_API_KEY} api_key = os.getenv("MODEL_API_KEY", "") config["model"]["api_key"] = config["model"].get("api_key", api_key) or api_key return config def main(): config = load_config() model_config = ModelConfig( base_url=config["model"]["base_url"], api_key=config["model"]["api_key"], model_name=config["model"]["model_name"], temperature=config["model"].get("temperature", 0.2), max_tokens=config["model"].get("max_tokens", 2048), ) model_client = create_model_client(model_config) # 注册所有工具(通过 import 已经触发装饰器) # 初始化记忆存储 memory_store = create_memory_store(config.get("memory", {})) # 从记忆读取历史,如果存在则恢复上下文 memory_key = "agent:session:demo" saved = memory_store.get(memory_key) if saved: print("load memory:", saved) runtime = AgentRuntime( model_client=model_client, tool_registry=tool_registry, system_prompt="你是一个拥有工具调用能力的助手,请根据用户问题调用合适工具。", max_steps=config["runtime"].get("max_steps", 10), request_timeout=config["runtime"].get("request_timeout", 60), ) result = runtime.run("请使用 echo 工具输出 hello agent") print("agent result:", result) # 保存记忆 memory_store.set(memory_key, { "last_input": "hello agent", "last_result": result, "created_at": "2025-01-01T00:00:00", }) if __name__ == "__main__": main()5.9 运行与验证
启动前确保:
- 模型 API 地址可用。
- 环境变量 MODEL_API_KEY 已设置。
- 如果配置了 Redis,确保 Redis 服务可用;没有 Redis 时,config.yaml 中 memory.type 设置为 memory 或直接使用默认 InMemoryStore。
运行命令:
export MODEL_API_KEY="your-api-key" python main.py预期输出大致如下(具体取决于模型行为):
load memory: {'last_input': 'hello agent', 'last_result': 'hello agent', 'created_at': '2025-01-01T00:00:00'} agent result: echo: hello agent如果你的模型不支持工具调用,agent result 可能是正常的文本回复,而不是工具调用结果。这属于模型能力差异,不是 Infra 代码问题。
6. 常见问题与排查思路
6.1 高频报错汇总
下面整理 Agent 开发中经常遇到的问题:
| 问题现象 | 常见原因 | 解决思路 |
|---|---|---|
| 模型调用超时 | 模型服务负载高、网络延迟、超时时间过短 | 加大 request_timeout,增加重试策略,检查服务端性能 |
| the agent execution provider did not respond in time | 执行器被阻塞或死锁;请求积压 | 检查执行线程池、队列积压、模型调用超时设置 |
| agent terminated due to error | 工具执行异常、循环卡死、上下文超限 | 查看日志定位卡在哪一步,给工具调用加兜底异常处理 |
| max steps exceeded | Agent 循环次数过多,模型一直在调用工具 | 优化 Prompt,增加终止条件,或限制最大步数 |
| 工具参数解析失败 | 模型返回的 JSON 参数不合法 | 使用 json 解析兜底,用 pydantic 校验参数 |
| 上下文窗口溢出 | 工具结果太长、历史消息累积 | 做消息裁剪、摘要化、控制每次工具返回内容长度 |
| 记忆写入失败 | Redis 连接失败、序列化错误 | 检查 Redis 服务,使用 try-except 包裹,提供降级方案 |
| 模型不支持工具调用 | 选择的模型没有 function calling 能力 | 更换支持工具调用的模型,或用提示词模拟工具协议 |
6.2 执行器超时排查步骤
如果你遇到“the agent execution provider did not respond in time”这类错误,建议按以下顺序排查:
- 确认是哪个环节超时:模型请求、工具执行,还是整个 Agent 循环。
- 检查模型服务 CPU、内存、并发请求数是否过高。
- 检查工具调用中是否有可能阻塞的操作,比如同步请求外部系统且没有设置超时。
- 检查 Agent 循环是否有死循环,例如工具调用参数固定导致永远无法完成。
- 在运行时中加入每步日志,观察最后停在哪一步。
示例做法是在 execute 工具时记录耗时:
start = time.time() result = self.tool_registry.execute(function_name, arguments) elapsed = time.time() - start print(f"tool {function_name} executed in {elapsed:.2f}s")6.3 工具调用死循环问题
工具调用死循环是 Agent 项目最典型的坑。常见场景:
- 模型反复调用同一个工具,每次得到的中间结果都让它继续调用。
- 工具返回的结果没有真正推进任务,导致 Agent 无法收敛。
解决方案:
- 为每个工具结果增加时效性描述。
- 在 Prompt 中说明“如果工具结果未能推进目标,请基于已有信息回答,不要重复调用”。
- 在 Runtime 中限制连续调用同一工具的次数,超过阈值直接终止。
6.4 上下文爆掉的预防
上下文管理不能只靠“裁剪”。更工程化的做法:
- 限制每个工具返回长度。
- 长文本先做摘要,再放入上下文。
- 设置全局 token 预算,超预算时触发告警。
- 对于超大工具结果,改写到对象存储,上下文里只保留引用。
7. 最佳实践与工程建议
7.1 接口抽象优先于框架绑定
无论你的项目使用什么 Agent 框架,底层能力模块都应该通过接口定义。模型接入、记忆存储、工具注册这三类能力尤其要抽象。哪怕初期只有一个实现,也要保留接口层。这样 Agent 框架升级时,你只需要写适配器,而不需要重写底层。
7.2 配置与密钥分离
Agent 项目通常会涉及模型 API Key、数据库地址、Redis 密码等敏感配置。建议:
- 配置文件只写非敏感项。
- 密钥通过环境变量或密钥管理服务注入。
- 不要在代码中硬编码 API Key。
- 对生产环境使用独立的密钥配置,不要复用本地配置。
7.3 工具调用必须有超时和熔断
外部工具可能不稳定,网络可能超时,第三方系统可能限流。工具执行应该统一封装超时逻辑,并提供熔断机制。当一个工具连续失败多次,应暂时停用该工具,防止 Agent 反复调用故障工具导致任务卡死。
7.4 日志和可观测性要前置
Agent 运行的每一步都值得记录:
- 模型输入输出。
- 工具调用参数和结果。
- 每一步耗时。
- 上下文长度。
- 错误堆栈。
生产环境建议接入分布式链路追踪,为每个 Agent 任务生成唯一 trace_id,方便排查多 Agent 协作时的消息传递问题。
7.5 安全边界要明确
Agent 可以调用工具,意味着它拥有执行动作的能力。Infra 层面必须做权限控制:
- Agent Scope:明确每个 Agent 能访问的资源和工具范围。
- 敏感操作审批:危险操作需要二次确认。
- 输入校验:工具参数必须做合法性校验,防止注入或误操作。
- 审计日志:记录所有工具调用行为和对象变更。
7.6 为多 Agent 协作预留能力
即使现在只做单 Agent,也建议把“任务 ID”“Agent ID”“请求追踪”这些字段纳入数据结构。后续扩展多 Agent 协作时,这些字段就是消息路由和状态隔离的基石。
7.7 评估意识要建立
Agent 不像传统接口那样可以用断言验证。建议从三个维度建立评估:
- 任务完成率:Agent 是否达到用户目标。
- 调用效率:是否用最少步骤完成任务,是否出现重复调用。
- 安全合规:是否有越权工具调用,是否泄漏敏感信息。
这些评估可以从运行时日志中统计,也可以单独建立评测集。
8. 收尾:先建底座,再等形态
Agent 形态还会继续变化,但模型接入、上下文管理、工具调用、记忆存储、可观测性、安全控制这六类能力不会消失。与其每天追着新框架跑,不如先把能力底座建设好,让上层 Agent 像搭积木一样自由切换。
本文用一个最小示例展示了“不绑定框架”的 Agent Infra 设计思路:模型客户端用接口隔离,工具用注册中心管理,上下文和记忆分别抽象存储层,Runtime 只负责执行循环。这个示例不是生产级方案,但足够让你理解 Infra 与 Agent 形态之间的关系。
如果你正在做 Agent 选型或重构,建议先按这个思路梳理现有系统的能力要素:哪些能力被业务代码硬编码了,哪些能力可以下沉为公共模块,哪些配置分散在多个服务里。把这些理清楚,再决定引入什么新框架也不迟。