从零构建智能体运行框架:基于Agent Harness的工程化实践

📅 2026/8/21 22:19:51
从零构建智能体运行框架:基于Agent Harness的工程化实践
如果你正在尝试将大语言模型LLM从“聊天机器人”升级为能自主完成复杂任务的“智能体”那么你很可能已经遇到了一个核心难题如何高效、稳定地管理智能体的生命周期是每次任务都手动编写冗长的提示词还是为每个智能体单独搭建一套运行环境当智能体数量增多、任务复杂度提升时你会发现简单的脚本调用变得难以维护错误难以追踪性能无法评估。这正是“智能体运行框架”要解决的根本问题。而Agent Harness正是这个领域中一个值得关注的概念和工具集。很多人误以为 Agent Harness 只是一个高级的 API 封装或任务调度器。实际上它的核心价值在于为智能体提供一套标准化的“测试与验证”体系。你可以把它想象成软件开发中的“持续集成/持续部署CI/CD”流水线但它是专门为 LLM 驱动的智能体设计的。它确保你的智能体不仅在理想环境下能工作在边界条件、异常输入和长期运行中也能保持可靠。本文将深入拆解 Agent Harness 的核心思想并基于这一思想为你呈现如何从零开始打造一个优秀的智能体运行框架。我们将不只讨论“是什么”更会聚焦“为什么重要”和“如何落地”。你将了解到智能体开发的核心痛点与框架要解决的真实问题。Agent Harness 的核心组件测试、评估、监控与编排。一个实战框架的架构设计包含清晰的模块划分。从零实现的完整代码示例涵盖智能体定义、任务执行与评估。生产环境的最佳实践与常见陷阱。无论你是想深入理解智能体工程化还是计划为自己的项目引入或开发一个运行框架这篇文章都将提供一条清晰的路径和可复用的代码。1. 这篇文章真正要解决的问题在 LLM 应用开发中我们经常面临一个困境原型验证很快但产品化极难。一个能在 Jupyter Notebook 里流畅对话的智能体一旦部署到真实、多变的环境中就可能出现各种问题回答偏离预期、无法处理复杂逻辑、性能不稳定、甚至产生有害内容。传统的软件开发有单元测试、集成测试和监控告警。但智能体的“行为”是非确定性的、基于自然语言的传统的测试方法几乎失效。这就是智能体工程化的最大障碍——缺乏可重复、可量化的质量保障机制。Agent Harness 的概念正是为此而生。它不是一个具体的工具而是一套方法论和工具集的统称旨在为智能体提供“缰绳”Harness对其进行约束、引导、测试和评估。它主要解决以下四个问题可测试性如何为基于自然语言交互的智能体编写“测试用例”如何断言它的输出是否符合预期可评估性如何量化智能体的表现除了准确率还有哪些维度如安全性、成本、延迟需要衡量可观测性智能体内部发生了什么它的思考过程Chain-of-Thought是怎样的调用了哪些工具消耗了多少 Token可编排性如何管理多个智能体的协作如何定义工作流、处理失败和进行重试本文的目标就是带你超越“单个智能体提示词优化”的层面从系统工程的角度理解并实践如何构建一个能解决上述问题的智能体运行框架。我们将从 Agent Harness 的理念出发最终落地为一个具备核心功能的简易框架。2. 基础概念与核心原理在深入代码之前我们需要统一几个关键概念这些概念是构建任何智能体框架的基石。2.1 LLM、智能体Agent与工具ToolLLM大语言模型如 GPT-4、Claude、LLaMA 等是智能体的“大脑”负责理解、推理和生成文本。智能体Agent一个由 LLM 驱动的、能够感知环境、进行决策并执行动作以完成目标的系统。一个智能体通常包含一个 LLM、一套可供调用的工具Tools、一个记忆Memory模块以及一个决策逻辑如 ReAct 框架。工具Tool智能体可以调用的函数或 API用于与外部世界交互。例如计算器、搜索引擎、数据库查询、代码执行器等。工具扩展了 LLM 的能力边界。2.2 什么是 Agent HarnessHarness 直译为“马具”或“安全带”在工程中引申为“控制系统”。Agent Harness 是为智能体设计的一套控制系统和评估体系。它的核心思想是将智能体视为一个“黑盒”或“灰盒”系统通过定义明确的输入、输出和评估标准来规范、测试和优化其行为。一个完整的 Agent Harness 通常包含以下层面测试套件Test Suite包含一系列针对不同技能和场景的测试用例。每个用例包括输入提示、上下文、期望的输出或行为规范。评估器Evaluator用于判断智能体输出质量的组件。评估可以是基于规则检查输出中是否包含特定关键词或格式。基于模型使用另一个 LLM评判员来评估输出质量、相关性和安全性。基于真实结果对于可执行的动作如运行代码、查询数据库验证其执行结果是否正确。运行环境Runtime负责加载智能体配置、管理工具调用、维护对话状态记忆、处理异常以及收集运行时指标如延迟、Token 消耗。编排器Orchestrator当任务需要多个智能体协作或顺序执行多个步骤时编排器负责管理工作流、传递数据和处理错误。2.3 智能体运行框架 vs. 普通 SDK很多 LLM 提供商如 OpenAI、Anthropic的 SDK 只提供了调用模型 API 的基础能力。LangChain、LlamaIndex 等框架前进了一步提供了智能体、工具链的抽象。而一个智能体运行框架是在此之上增加了Harness 层专注于生命周期管理、质量保障和运维支撑。我们可以用下表来对比特性LLM SDK (如 openai)应用框架 (如 LangChain)智能体运行框架 (本文目标)核心能力模型 API 调用组件抽象、链式编排智能体生命周期管理、测试评估、监控测试支持无有限通常需自行封装内置测试套件与评估器可观测性基础日志回调Callbacks集成的指标收集与追踪部署运维无无考虑配置管理、版本控制、回滚适用阶段原型验证应用开发产品化、持续集成/交付理解了这些概念我们就可以开始设计自己的框架了。3. 环境准备与前置条件我们将使用 Python 作为实现语言因为它拥有最丰富的 LLM 生态。以下是构建和运行本示例框架所需的环境。3.1 基础环境操作系统Windows 10/11, macOS, 或 Linux (Ubuntu 20.04 推荐)Python 版本3.9 或 3.103.11 也可注意某些包兼容性包管理工具pip(建议使用虚拟环境venv或conda)3.2 核心依赖库我们将基于langchain和langchain-openai来构建智能体基础因为它们提供了成熟的抽象。同时我们会引入pydantic用于数据验证pytest作为测试运行器用于启发我们的评估思路。创建一个requirements.txt文件# 核心LLM与智能体框架 langchain0.1.0 langchain-openai0.0.5 langchain-community0.0.10 # 包含更多社区工具 # OpenAI API (示例模型提供商) openai1.6.1 # 工具与工具调用示例 requests2.31.0 # 用于构建网络请求工具 # 框架核心数据验证、异步、配置 pydantic2.5.0 pydantic-settings2.1.0 # 测试与评估思想借鉴 pytest7.4.0 # 辅助日志、JSON python-dotenv1.0.0 # 管理环境变量如API密钥使用以下命令安装依赖# 创建并激活虚拟环境以venv为例 python -m venv .venv # Windows .venv\Scripts\activate # Linux/macOS source .venv/bin/activate # 安装依赖 pip install -r requirements.txt3.3 配置 API 密钥为了运行示例你需要一个 OpenAI API 密钥或其他兼容 API如 Azure OpenAI。将其设置在环境变量中。创建.env文件切勿提交到版本控制OPENAI_API_KEYsk-your-actual-api-key-here在代码中使用python-dotenv加载from dotenv import load_dotenv load_dotenv() # 现在可以通过 os.getenv(‘OPENAI_API_KEY’) 获取4. 核心框架架构设计我们的目标是构建一个轻量级但结构清晰的框架它应该包含以下核心模块agent-harness-framework/ ├── core/ │ ├── __init__.py │ ├── agent.py # 智能体基类与定义 │ ├── runtime.py # 智能体运行时环境 │ ├── evaluator.py # 评估器抽象与实现 │ └── orchestration.py # 简单工作流编排 ├── tools/ │ ├── __init__.py │ └── calculator.py # 示例工具 ├── tests/ │ ├── __init__.py │ ├── test_agent.py # 智能体测试用例 │ └── fixtures/ # 测试数据 ├── config/ │ └── settings.py # 配置管理 ├── examples/ # 使用示例 └── main.py # 入口点示例设计要点松耦合智能体定义、工具、评估逻辑相互独立。可扩展可以轻松添加新的工具、评估器或智能体类型。可观测运行时记录关键事件和指标。配置化智能体行为通过配置文件或代码可调。5. 从零实现框架核心模块代码让我们开始实现最关键的部分。我们将遵循从底层工具到上层评估的步骤。5.1 步骤一定义智能体基类与配置首先在core/agent.py中我们定义一个基础的智能体类。它封装了 LangChain 的智能体并添加了框架所需的元数据。# core/agent.py from typing import Any, Dict, List, Optional, Type from pydantic import BaseModel, Field from langchain.agents import AgentExecutor from langchain.tools import BaseTool class AgentConfig(BaseModel): 智能体配置模型 name: str Field(..., description智能体名称) description: str Field(..., description智能体功能描述) model_name: str Field(defaultgpt-3.5-turbo, description使用的LLM模型) temperature: float Field(default0.1, description模型温度参数) max_iterations: int Field(default5, description最大推理步数) verbose: bool Field(defaultFalse, description是否输出详细日志) class BaseAgent: 框架智能体基类 def __init__( self, config: AgentConfig, tools: List[BaseTool], **kwargs ): self.config config self.tools tools self._agent_executor: Optional[AgentExecutor] None self._initialize_agent(**kwargs) def _initialize_agent(self, **kwargs): 初始化底层的LangChain AgentExecutor。子类可重写此方法。 # 这是一个简化示例。实际中你需要根据智能体类型如ReAct, OpenAI Functions来创建。 from langchain.agents import create_openai_tools_agent from langchain_openai import ChatOpenAI from langchain.agents import AgentExecutor llm ChatOpenAI( modelself.config.model_name, temperatureself.config.temperature, **kwargs ) # 创建提示词模板 - 简化版 from langchain.prompts import ChatPromptTemplate, MessagesPlaceholder prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手。请使用提供的工具来回答问题。), MessagesPlaceholder(variable_namechat_history), (human, {input}), MessagesPlaceholder(variable_nameagent_scratchpad), ]) # 创建智能体 agent create_openai_tools_agent(llm, self.tools, prompt) # 创建执行器 self._agent_executor AgentExecutor( agentagent, toolsself.tools, verboseself.config.verbose, max_iterationsself.config.max_iterations, handle_parsing_errorsTrue, # 优雅处理解析错误 **kwargs ) async def ainvoke(self, input_data: Dict[str, Any]) - Dict[str, Any]: 异步调用智能体 if not self._agent_executor: raise RuntimeError(Agent not initialized.) return await self._agent_executor.ainvoke(input_data) def invoke(self, input_data: Dict[str, Any]) - Dict[str, Any]: 同步调用智能体 if not self._agent_executor: raise RuntimeError(Agent not initialized.) return self._agent_executor.invoke(input_data) def get_metrics(self) - Dict[str, Any]: 获取本次运行的指标示例可扩展为记录Token数、耗时等 # 此处为示例实际需要从_executor或回调中收集 return { agent_name: self.config.name, model: self.config.model_name, max_iterations: self.config.max_iterations }这个基类做了几件关键事1) 用 Pydantic 管理配置2) 封装了 LangChain 执行器3) 提供了同步/异步接口4) 预留了指标收集接口。5.2 步骤二实现工具与运行时环境工具是智能体的手脚。我们在tools/calculator.py中实现一个简单的计算器工具。# tools/calculator.py from typing import Optional, Type from pydantic import BaseModel, Field from langchain.tools import BaseTool class CalculatorInput(BaseModel): 计算器工具的输入模型 expression: str Field(description一个合法的数学表达式例如3 5 * 2) class CalculatorTool(BaseTool): name “calculator” description “用于计算一个数学表达式的值。输入应该是一个字符串格式的表达式如 ‘3 5 * 2’。” args_schema: Type[BaseModel] CalculatorInput return_direct: bool False # 结果交给Agent继续处理 def _run(self, expression: str) - str: 同步执行计算 try: # 警告使用eval存在安全风险仅用于演示。生产环境应用安全库如ast.literal_eval或自定义解析器。 result eval(expression, {“__builtins__”: None}, {}) return f”表达式 {expression} 的计算结果是{result}” except Exception as e: return f”计算失败{e}。请检查表达式格式。” async def _arun(self, expression: str) - str: 异步执行计算本例中直接调用同步方法 return self._run(expression)重要安全提示上述代码中的eval函数在真实生产环境中是极度危险的因为它会执行任意代码。此处仅用于最简单演示。真实工具必须进行严格的输入验证和沙箱处理或使用安全的计算库如numexpr、ast.literal_eval处理有限操作。接下来在core/runtime.py中我们创建一个简单的运行时环境负责加载配置、管理智能体实例和执行任务。# core/runtime.py import logging from typing import Dict, Any, Optional from .agent import BaseAgent, AgentConfig from langchain.tools import BaseTool logger logging.getLogger(__name__) class AgentRuntime: 智能体运行时管理器 def __init__(self): self._agents: Dict[str, BaseAgent] {} self._tools_registry: Dict[str, BaseTool] {} def register_tool(self, tool: BaseTool): 注册一个工具到运行时 self._tools_registry[tool.name] tool logger.info(f”Tool registered: {tool.name}”) def create_agent(self, agent_config: AgentConfig, tool_names: Optional[list] None) - BaseAgent: 根据配置创建智能体实例 # 选择工具 selected_tools [] if tool_names: for name in tool_names: if name in self._tools_registry: selected_tools.append(self._tools_registry[name]) else: logger.warning(f”Tool ‘{name}’ not found in registry, skipping.”) else: selected_tools list(self._tools_registry.values()) # 默认使用所有工具 # 创建智能体 agent BaseAgent(configagent_config, toolsselected_tools) self._agents[agent_config.name] agent logger.info(f”Agent created: {agent_config.name}”) return agent def get_agent(self, name: str) - Optional[BaseAgent]: 获取已创建的智能体 return self._agents.get(name) async def run_agent(self, agent_name: str, user_input: str) - Dict[str, Any]: 运行指定智能体 agent self.get_agent(agent_name) if not agent: raise ValueError(f”Agent ‘{agent_name}’ not found.”) logger.info(f”Running agent ‘{agent_name}’ with input: {user_input}”) result await agent.ainvoke({“input”: user_input, “chat_history”: []}) logger.info(f”Agent ‘{agent_name}’ finished. Result: {result.get(‘output’, ‘No output’)}”) return result运行时环境充当了容器和协调者的角色这是框架价值的重要体现。5.3 步骤三构建评估器Evaluator评估是 Harness 的核心。我们在core/evaluator.py中定义一个评估器接口和几种实现。# core/evaluator.py from abc import ABC, abstractmethod from typing import Any, Dict, List import asyncio class AgentEvaluator(ABC): 评估器抽象基类 abstractmethod async def evaluate(self, agent_response: Dict[str, Any], test_case: Dict[str, Any]) - Dict[str, Any]: 评估智能体的响应。 :param agent_response: 智能体invoke/ainvoke返回的结果字典 :param test_case: 测试用例包含输入、预期输出等信息 :return: 评估结果字典至少包含 {‘score’: float, ‘passed’: bool, ‘details’: str} pass class ExactMatchEvaluator(AgentEvaluator): 精确匹配评估器适用于有明确答案的任务 async def evaluate(self, agent_response: Dict[str, Any], test_case: Dict[str, Any]) - Dict[str, Any]: output agent_response.get(‘output’, ‘’).strip() expected test_case.get(‘expected_output’, ‘’).strip() passed (output expected) score 1.0 if passed else 0.0 return { ‘score’: score, ‘passed’: passed, ‘details’: f”Output: ‘{output}’ | Expected: ‘{expected}’” } class KeywordMatchEvaluator(AgentEvaluator): 关键词匹配评估器检查输出中是否包含特定关键词 def __init__(self, required_keywords: List[str], any_keywords: List[str] None): self.required_keywords required_keywords self.any_keywords any_keywords or [] async def evaluate(self, agent_response: Dict[str, Any], test_case: Dict[str, Any]) - Dict[str, Any]: output agent_response.get(‘output’, ‘’).lower() details [] score 0.0 # 检查必需关键词 req_missing [] for kw in self.required_keywords: if kw.lower() not in output: req_missing.append(kw) if req_missing: details.append(f”Missing required keywords: {req_missing}”) else: score 0.7 # 基础分 details.append(“All required keywords found.”) # 检查可选关键词加分项 found_any [] for kw in self.any_keywords: if kw.lower() in output: found_any.append(kw) if found_any: score 0.3 * (len(found_any) / max(len(self.any_keywords), 1)) details.append(f”Bonus keywords found: {found_any}”) passed (len(req_missing) 0) # 必需关键词全找到才算通过 return { ‘score’: min(score, 1.0), # 确保分数不超过1 ‘passed’: passed, ‘details’: ‘; ‘.join(details) } class LLMAsJudgeEvaluator(AgentEvaluator): 使用另一个LLM作为裁判进行评估适用于开放性任务 def __init__(self, llm_judge, evaluation_prompt): self.llm_judge llm_judge self.evaluation_prompt evaluation_prompt async def evaluate(self, agent_response: Dict[str, Any], test_case: Dict[str, Any]) - Dict[str, Any]: # 构建给裁判LLM的提示词 prompt self.evaluation_prompt.format( questiontest_case.get(‘input’), expectedtest_case.get(‘expected_output’, ‘N/A’), actualagent_response.get(‘output’, ‘’) ) # 调用裁判LLM judge_response await self.llm_judge.ainvoke(prompt) # 解析裁判的输出这里简化处理实际需要更复杂的解析逻辑 judge_text judge_response.content if hasattr(judge_response, ‘content’) else str(judge_response) # 简单判断是否包含“通过”、“正确”等词 passed any(word in judge_text for word in [‘通过’, ‘正确’, ‘符合’, ‘yes’, ‘correct’]) score 1.0 if passed else 0.5 # 简化评分 return { ‘score’: score, ‘passed’: passed, ‘details’: f”LLM Judge said: {judge_text[:200]}…” # 截断长文本 }评估器是框架的“质量关卡”你可以根据任务类型灵活组合或自定义评估器。5.4 步骤四编写测试用例与集成测试现在让我们在tests/目录下创建测试。这不仅是验证框架也是展示如何为智能体编写“测试套件”。# tests/test_agent.py import pytest import asyncio from core.runtime import AgentRuntime from core.agent import AgentConfig from tools.calculator import CalculatorTool from core.evaluator import ExactMatchEvaluator, KeywordMatchEvaluator pytest.fixture def runtime_with_tools(): 提供一个已注册计算器工具的运行时环境 rt AgentRuntime() rt.register_tool(CalculatorTool()) return rt pytest.fixture def math_agent_config(): 数学智能体配置 return AgentConfig( name“math_assistant”, description“一个能进行数学计算的助手”, model_name“gpt-3.5-turbo”, # 测试时可用模拟模型以节省成本 max_iterations3 ) pytest.mark.asyncio async def test_agent_calculation(runtime_with_tools, math_agent_config): 测试智能体能否正确使用计算器工具 # 1. 创建智能体 agent runtime_with_tools.create_agent(math_agent_config, tool_names[“calculator”]) # 2. 运行智能体 test_input “计算一下 15 加上 27 等于多少” result await runtime_with_tools.run_agent(“math_assistant”, test_input) # 3. 评估结果 evaluator KeywordMatchEvaluator(required_keywords[“42”]) # 期望输出包含结果42 evaluation await evaluator.evaluate(result, {“input”: test_input}) # 4. 断言 assert evaluation[‘passed’] True, f”Test failed. Details: {evaluation[‘details’]}” print(f”Test passed! Score: {evaluation[‘score’]}”) pytest.mark.asyncio async def test_agent_no_tool_call(runtime_with_tools, math_agent_config): 测试智能体在不需工具时的对话能力 agent runtime_with_tools.create_agent(math_agent_config, tool_names[“calculator”]) test_input “你好请介绍一下你自己。” result await runtime_with_tools.run_agent(“math_assistant”, test_input) # 评估不应触发工具调用可通过检查result中的intermediate_steps判断此处简化 assert ‘output’ in result assert len(result.get(‘output’, ‘’)) 0 print(f”Agent responded: {result[‘output’][:100]}…”) if __name__ “__main__”: # 方便直接运行测试 loop asyncio.get_event_loop() rt AgentRuntime() rt.register_tool(CalculatorTool()) cfg AgentConfig(name“test_agent”, description“Test”) loop.run_until_complete(test_agent_calculation(rt, cfg))这个测试文件展示了如何将智能体、运行时和评估器串联起来形成一个完整的测试流程。6. 运行结果与效果验证让我们编写一个完整的示例脚本examples/demo.py来演示框架的端到端使用。# examples/demo.py import asyncio import logging from core.runtime import AgentRuntime from core.agent import AgentConfig from tools.calculator import CalculatorTool from core.evaluator import KeywordMatchEvaluator # 配置日志 logging.basicConfig(levellogging.INFO) logger logging.getLogger(__name__) async def main(): logger.info(“Starting Agent Harness Framework Demo…”) # 1. 初始化运行时 runtime AgentRuntime() # 2. 注册工具 runtime.register_tool(CalculatorTool()) logger.info(“Tools registered.”) # 3. 创建智能体配置 agent_config AgentConfig( name“demo_math_agent”, description“演示用的数学计算智能体”, model_name“gpt-3.5-turbo”, temperature0.1, verboseTrue # 打开详细日志观察思考过程 ) # 4. 创建智能体 agent runtime.create_agent(agent_config, tool_names[“calculator”]) logger.info(f”Agent ‘{agent.config.name}’ created.”) # 5. 定义测试用例 test_cases [ {“input”: “123 乘以 456 等于多少”, “expected_keywords”: [“56088”]}, {“input”: “(15 27) / 2 的结果是什么”, “expected_keywords”: [“21”]}, {“input”: “请写一首关于春天的诗不要计算。”, “expected_keywords”: []}, # 开放性任务 ] # 6. 运行测试并评估 evaluator KeywordMatchEvaluator(required_keywords[]) all_passed True for i, test_case in enumerate(test_cases): logger.info(f”\n Running Test Case {i1}: {test_case[‘input’]} ”) try: # 运行智能体 result await runtime.run_agent(agent.config.name, test_case[‘input’]) # 评估根据用例调整评估器 case_evaluator KeywordMatchEvaluator(required_keywordstest_case.get(‘expected_keywords’, [])) evaluation await case_evaluator.evaluate(result, test_case) # 输出结果 status “PASS” if evaluation[‘passed’] else “FAIL” logger.info(f”Result: {status} | Score: {evaluation[‘score’]:.2f}”) logger.info(f”Agent Output: {result.get(‘output’, ‘N/A’)}”) logger.info(f”Evaluation Details: {evaluation[‘details’]}”) if not evaluation[‘passed’]: all_passed False except Exception as e: logger.error(f”Test case {i1} failed with error: {e}”) all_passed False # 7. 总结 logger.info(f”\n Demo Summary ) logger.info(f”All tests passed: {all_passed}”) if agent: logger.info(f”Agent metrics: {agent.get_metrics()}”) if __name__ “__main__”: asyncio.run(main())如何运行与验证确保已设置OPENAI_API_KEY环境变量。在项目根目录执行python examples/demo.py预期输出你将看到详细的日志包括智能体的思考过程因为verboseTrue、工具调用以及每个测试用例的评估结果。对于前两个数学问题智能体应成功调用计算器并返回正确答案。对于第三个问题它应正常进行对话而不调用工具。验证成功的关键控制台输出中包含Tool Call: calculator等日志。数学问题的输出中包含正确数字。测试用例的评估状态显示为PASS。没有抛出未处理的异常。7. 常见问题与排查思路在开发和运行此类框架时你可能会遇到以下典型问题问题现象可能原因排查方式解决方案智能体不调用工具1. 提示词未明确要求使用工具。2. 工具描述不清晰LLM 无法理解。3. 模型温度过高导致输出随机。1. 检查verboseTrue的日志看 Agent 的思考过程。2. 检查工具的描述 (description) 是否准确。3. 尝试更简单的输入。1. 优化系统提示词明确指令如“你必须使用可用工具”。2. 重写工具描述使其更匹配自然语言问题。3. 降低temperature(如设为 0.1)。工具调用出错如eval安全错误1. 工具内部代码有 bug。2. 输入格式不符合工具预期。1. 查看工具_run方法中的错误信息。2. 检查 Agent 传递给工具的参数字符串。1. 在工具内部添加更健壮的异常处理。2. 使用更安全的替代方案如ast.literal_eval或数学解析库。API 调用超时或失败1. 网络问题。2. API 密钥无效或额度不足。3. 请求速率超限。1. 检查网络连接。2. 检查 API 密钥和环境变量。3. 查看 OpenAI 控制台用量和错误信息。1. 添加重试机制和超时设置。2. 验证 API 密钥并确保有余额。3. 降低请求频率或升级账户。评估器评分不准1. 评估逻辑过于简单如精确匹配。2. LLM 作为裁判的提示词设计不佳。1. 手动检查几个失败案例的输出和预期。2. 分析裁判 LLM 的原始输出。1. 采用更鲁棒的评估策略如关键词匹配、嵌入相似度、或多维度评估。2. 设计更详细的评估提示词并让裁判 LLM 输出结构化结果如 JSON。多智能体协作混乱1. 智能体间通信协议未定义。2. 没有全局状态管理。3. 出现循环依赖或死锁。1. 记录每个智能体的输入输出。2. 检查工作流编排逻辑。1. 定义明确的消息格式如使用pydantic模型。2. 引入集中式的状态存储或消息总线。3. 为工作流设置超时和最大步数限制。8. 最佳实践与工程建议将智能体框架投入生产环境需要超越“能跑通”的层面考虑以下工程化实践8.1 配置管理与环境分离使用 Pydantic Settings将模型配置、API 端点、超时时间等抽象为配置类支持从环境变量、配置文件多级加载。环境隔离严格区分开发、测试、生产环境的配置如使用不同的 API 密钥、模型版本。8.2 可观测性与监控结构化日志使用structlog或logging的 JSON 格式化记录每次调用的agent_name,input,output,tool_calls,token_usage,latency等。指标收集集成像 Prometheus 这样的监控系统暴露关键指标如请求量、成功率、平均响应时间、Token 消耗。分布式追踪为每个用户会话或请求生成唯一trace_id串联起所有智能体和工具的调用链便于问题排查。8.3 测试策略分层测试单元测试单独测试每个工具、评估器的逻辑。集成测试测试智能体与工具的配合使用模拟MockLLM 以节省成本和保证确定性。端到端测试在预发布环境使用真实 LLM 运行关键用例评估整体效果。测试数据集构建一个涵盖核心功能、边界情况和对抗性输入的测试用例库并定期回归测试。8.4 安全与合规工具沙箱化对于执行代码、访问网络或文件系统的工具必须在严格的沙箱环境中运行限制其权限和资源。输入输出过滤对用户输入和智能体输出进行内容安全过滤防止注入攻击、隐私泄露或生成有害内容。权限控制根据用户角色动态决定智能体可以访问哪些工具和数据。8.5 性能与成本优化缓存对频繁出现的、确定性较高的查询结果进行缓存例如使用 Redis 缓存特定输入对应的 LLM 输出或工具结果。批处理对于非实时任务可以考虑将多个请求批量发送给 LLM API以降低成本。模型路由根据任务复杂度动态选择不同能力和成本的模型如简单任务用便宜模型复杂任务用强大模型。8.6 版本控制与部署智能体即代码将智能体的配置提示词、工具链、参数纳入版本控制系统如 Git。蓝绿部署/金丝雀发布新版本智能体上线时先引导少量流量进行测试验证效果后再全量发布。回滚机制确保能快速回退到上一个稳定版本的智能体配置。构建一个成熟的智能体运行框架是一个持续迭代的过程。本文提供的代码是一个起点展示了 Agent Harness 的核心思想——通过标准化、可测试、可观测的运行时环境来约束和提升智能体的可靠性。你可以在此基础上根据实际业务需求逐步添加更复杂的特性如工作流引擎、长期记忆、人类反馈集成等最终打造出能够支撑关键业务场景的智能体基础设施。