在实际 AI 应用开发中将大型语言模型LLM的能力与外部工具、API 和数据源连接起来构建一个能够自主执行复杂任务的智能体Agent是当前技术探索的热点。然而从零开始搭建一个稳定、可扩展的 Agent 运行时环境往往涉及复杂的架构设计、工具调用Tool Call的标准化处理、状态管理以及错误恢复机制这对于许多开发者来说是一个不小的挑战。本文将以zditor作为一个快速构建 Agent 的框架或平台示例探讨如何在 24 小时内构建一个类似 Codex 风格的 Harness Agent。我们将聚焦于理解 Agent 的核心运行机制特别是 Agent Runtime 和 Tool Call 这两个关键概念并提供一个从环境准备到核心功能实现的实践路径。无论你是希望快速验证一个 AI 应用想法还是希望深入理解智能体系统的内部工作原理本文都将提供一个清晰的、可操作的指南。1. 理解核心概念Agent Runtime 与 Tool Call在开始动手之前必须厘清几个核心概念这决定了我们构建的 Agent 是否具备真正的“智能执行”能力而不仅仅是一个聊天接口。1.1 什么是 Harness Agent“Harness”在这里可以理解为“驾驭”或“控制”。一个 Harness Agent 的核心思想是它能够“驾驭”或“编排”一系列外部工具Tools来完成用户指定的任务。它不仅仅是生成文本而是根据目标动态地决定调用哪个工具、传入什么参数、如何处理工具的返回结果并可能进行多轮交互直至任务完成或无法继续。例如用户请求“查询北京明天的天气然后告诉我是否需要带伞”。一个简单的聊天模型可能只会回复“我可以帮你查天气但我需要调用天气API”。而一个 Harness Agent 则会自动执行以下流程理解意图需要天气查询工具。工具调用调用配置好的天气 API传入参数city北京。结果解析收到 API 返回的天气数据如晴最高温 25°C。决策与执行基于“是否需要带伞”的规则如判断是否有雨得出结论“不需要带伞”。生成回复组织自然语言回复用户。1.2 Agent Runtime智能体的执行引擎Agent Runtime 是智能体的“运行时环境”或“执行引擎”。它负责管理 Agent 的整个生命周期和决策循环。一个典型的 Agent Runtime 需要处理以下事务状态管理维护 Agent 与用户对话的历史Memory以及当前任务执行的上下文Context。工具发现与绑定让 Agent 知道它可以使用哪些工具以及每个工具的调用规范。决策制定基于用户输入、历史上下文和可用工具决定下一步是直接回复、调用工具还是结束对话。这通常由 LLM 本身根据提示词或一个专门的规划器Planner来完成。工具调用执行将 LLM 输出的结构化工具调用请求如{“tool”: “get_weather”, “args”: {“city”: “北京”}}转化为实际的 API 调用或函数执行。错误处理与重试处理工具调用失败、网络超时、返回结果格式异常等情况并决定重试或向用户报告错误。流式输出控制管理最终回复的生成和输出方式。你可以把 Agent Runtime 想象成一个操作系统内核它调度资源工具管理进程任务步骤并处理异常。1.3 Tool Call智能体与世界的交互接口Tool Call 是 Agent 与外部世界交互的标准方式。它本质上是一个标准化的请求-响应协议。现代 LLM如 GPT-4, Claude 3通常支持在聊天补全接口中以特定的 JSON 格式声明工具并接收模型返回的工具调用请求。一个完整的 Tool Call 流程包括工具定义以 JSON Schema 的形式描述工具包括工具名、描述、所需参数及其类型。{ type: function, function: { name: get_weather, description: 获取指定城市的天气信息, parameters: { type: object, properties: { city: { type: string, description: 城市名称例如北京、上海 } }, required: [city] } } }模型决策LLM 根据对话上下文判断是否需要调用工具。如果需要它会生成一个结构化的工具调用请求。{ role: assistant, content: null, tool_calls: [ { id: call_abc123, type: function, function: { name: get_weather, arguments: {\city\: \北京\} } } ] }工具执行Runtime 解析这个请求找到本地对应的函数或远程 API 并执行。结果返回将工具执行的结果成功或失败以特定格式返回给 LLM供其生成下一步回复。{ role: tool, content: {\weather\: \晴\, \temperature\: 25}, tool_call_id: call_abc123 }2. 环境准备与项目初始化我们假设使用一个类似zditor的抽象框架来加速开发。在现实中你可以选择 LangChain、LlamaIndex、Semantic Kernel 或自主开发。这里我们以构建一个最小化自主 Agent Runtime 为目标。2.1 技术栈选择与依赖为了在24小时内完成我们选择 Python 作为开发语言因为它有最丰富的 AI 生态。我们将使用openai库作为与 LLM 交互的基础并自行构建 Runtime 的核心逻辑。基础环境要求Python 3.9pip 包管理工具一个可用的 OpenAI API 密钥或其他兼容 OpenAI API 的模型服务密钥创建项目目录并初始化虚拟环境# 创建项目目录 mkdir harness_agent_quickstart cd harness_agent_quickstart # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Linux/Mac: source venv/bin/activate # 创建核心文件 touch main.py agent_runtime.py tools.py config.py requirements.txt编辑requirements.txt添加基础依赖openai1.0.0 pydantic2.0.0 # 用于数据验证和设置管理 httpx0.25.0 # 用于异步HTTP请求调用外部API python-dotenv1.0.0 # 用于管理环境变量安装依赖pip install -r requirements.txt2.2 配置管理在config.py中管理配置避免将密钥硬编码在代码中。# config.py import os from pydantic_settings import BaseSettings from dotenv import load_dotenv load_dotenv() # 从 .env 文件加载环境变量 class Settings(BaseSettings): # OpenAI 配置 openai_api_key: str os.getenv(OPENAI_API_KEY, ) openai_base_url: str os.getenv(OPENAI_BASE_URL, https://api.openai.com/v1) # 支持自定义端点 openai_model: str os.getenv(OPENAI_MODEL, gpt-3.5-turbo) # 默认模型 # Agent 配置 agent_max_iterations: int 10 # Agent 最大执行轮次防止死循环 agent_temperature: float 0.1 # 较低的温度使输出更确定适合工具调用 class Config: env_file .env settings Settings()创建.env文件务必加入.gitignore# .env OPENAI_API_KEYsk-your-openai-api-key-here # OPENAI_BASE_URLhttps://your-custom-endpoint.com/v1 # 如果需要使用其他兼容API OPENAI_MODELgpt-4-turbo-preview # 建议使用更强大的模型以获得更好的工具调用能力3. 构建核心 Agent Runtime我们的 Runtime 需要完成以下核心功能管理对话历史、处理工具定义、执行 LLM 调用、解析工具调用、执行工具、管理循环。3.1 定义数据结构首先在agent_runtime.py中定义核心的数据模型。# agent_runtime.py from typing import Dict, Any, List, Callable, Optional, Union from pydantic import BaseModel import json class Tool(BaseModel): 工具定义模型 name: str description: str parameters: Dict[str, Any] # JSON Schema function: Callable # 实际执行的函数 class ToolCall(BaseModel): 工具调用请求模型 id: str type: str “function” function: Dict[str, str] # 包含 name 和 arguments class Message(BaseModel): 消息模型兼容OpenAI格式 role: str # system, user, assistant, tool content: Optional[str] None tool_calls: Optional[List[ToolCall]] None tool_call_id: Optional[str] None # 当 roletool 时使用 class AgentState(BaseModel): Agent运行时状态 messages: List[Message] [] # 完整的对话历史 available_tools: Dict[str, Tool] {} # 可用的工具字典 iteration_count: int 0 # 当前迭代次数3.2 实现 Agent Runtime 主类接下来实现 Runtime 的主类它封装了主要的循环逻辑。# agent_runtime.py (续) from openai import OpenAI from config import settings import asyncio from typing import Tuple class HarnessAgentRuntime: def __init__(self, system_prompt: str “你是一个有帮助的AI助手可以调用工具来解决问题。”): self.client OpenAI( api_keysettings.openai_api_key, base_urlsettings.openai_base_url ) self.system_prompt system_prompt self.state AgentState() # 初始化系统消息 self.state.messages.append(Message(role“system”, contentsystem_prompt)) def register_tool(self, tool: Tool): 向Runtime注册一个工具 self.state.available_tools[tool.name] tool def _prepare_openai_tools(self) - List[Dict]: 将内部Tool对象转换为OpenAI API所需的格式 openai_tools [] for tool in self.state.available_tools.values(): openai_tools.append({ “type”: “function”, “function”: { “name”: tool.name, “description”: tool.description, “parameters”: tool.parameters } }) return openai_tools async def execute_tool_call(self, tool_call: ToolCall) - Tuple[str, bool]: 执行单个工具调用返回结果和是否成功 tool_name tool_call.function[“name”] if tool_name not in self.state.available_tools: return f“Error: Tool ‘{tool_name}’ not found.”, False try: # 解析参数 arguments json.loads(tool_call.function[“arguments”]) tool_obj self.state.available_tools[tool_name] # 执行工具函数 result await tool_obj.function(**arguments) if asyncio.iscoroutinefunction(tool_obj.function) else tool_obj.function(**arguments) return str(result), True except json.JSONDecodeError as e: return f“Error: Invalid arguments JSON. {e}”, False except Exception as e: return f“Error during execution: {e}”, False async def process_user_input(self, user_input: str) - str: 处理用户输入驱动Agent循环返回最终回复 # 1. 添加用户消息 self.state.messages.append(Message(role“user”, contentuser_input)) while self.state.iteration_count settings.agent_max_iterations: self.state.iteration_count 1 print(f“\n[Iteration {self.state.iteration_count}]”) # 2. 调用LLM传入当前消息历史和可用工具 openai_tools self._prepare_openai_tools() try: response self.client.chat.completions.create( modelsettings.openai_model, messages[m.dict(exclude_noneTrue) for m in self.state.messages], toolsopenai_tools if openai_tools else None, tool_choice“auto” if openai_tools else None, temperaturesettings.agent_temperature, ) except Exception as e: # 处理API调用错误如网络问题、鉴权失败 error_msg f“LLM API call failed: {e}” self.state.messages.append(Message(role“assistant”, contenterror_msg)) return error_msg message response.choices[0].message assistant_msg Message( role“assistant”, contentmessage.content, tool_calls[ToolCall(**tc.model_dump()) for tc in message.tool_calls] if message.tool_calls else None ) self.state.messages.append(assistant_msg) # 3. 检查LLM是否要求调用工具 if not assistant_msg.tool_calls: # 没有工具调用直接返回助理的回复 final_reply assistant_msg.content or “(No content)” return final_reply # 4. 执行工具调用 tool_messages [] for tool_call in assistant_msg.tool_calls: print(f“ Executing tool: {tool_call.function[‘name’]} with args: {tool_call.function[‘arguments’]}”) tool_result, success await self.execute_tool_call(tool_call) # 将工具执行结果添加到消息历史 tool_msg Message( role“tool”, contenttool_result, tool_call_idtool_call.id ) tool_messages.append(tool_msg) self.state.messages.extend(tool_messages) # 循环继续将工具结果作为上下文让LLM进行下一步决策 # 超出最大迭代次数 return f“Agent stopped after reaching max iterations ({settings.agent_max_iterations}).”4. 实现工具与运行验证Runtime 搭建好了现在需要为其“注入”实际的能力即定义和注册具体的工具。4.1 定义示例工具在tools.py中我们定义几个简单的工具。# tools.py import httpx import asyncio from datetime import datetime from agent_runtime import Tool from pydantic import BaseModel, Field # 示例1一个简单的计算器工具 async def calculator(a: float, b: float, operation: str Field(description“操作类型: add, subtract, multiply, divide”)) - str: “”“执行简单的数学计算。”“” if operation “add”: result a b elif operation “subtract”: result a - b elif operation “multiply”: result a * b elif operation “divide”: if b 0: return “Error: Division by zero.” result a / b else: return f“Error: Unknown operation ‘{operation}’. Supported: add, subtract, multiply, divide” return f“The result of {a} {operation} {b} is {result}” # 示例2一个模拟的网络查询工具获取当前时间 async def get_current_time(timezone: str Field(default“UTC”, description“时区例如 UTC, Asia/Shanghai”)) - str: “”“获取指定时区的当前时间。这是一个模拟的外部API调用。”“” # 模拟网络延迟 await asyncio.sleep(0.5) now datetime.utcnow() # 这里简化处理实际应使用pytz等库进行时区转换 if “shanghai” in timezone.lower(): from datetime import timedelta now now timedelta(hours8) return f“The current time in {timezone} is approximately {now.strftime(‘%Y-%m-%d %H:%M:%S’)}.” # 示例3一个需要真实网络请求的工具需要安装httpx async def search_web(query: str) - str: “”“模拟网络搜索实际调用一个公共API。”“” # 注意这是一个示例实际使用请遵守目标API的使用条款。 # 这里使用一个返回JSON的公共API作为演示。 url f“https://api.agify.io?name{query}” try: async with httpx.AsyncClient() as client: resp await client.get(url, timeout10.0) if resp.status_code 200: data resp.json() # 对结果进行解释 return f“Based on public data, the name ‘{query}’ is associated with an estimated age of {data.get(‘age’, ‘N/A’)} and {data.get(‘count’, 0)} occurrences in the dataset.” else: return f“Search API returned an error: HTTP {resp.status_code}” except Exception as e: return f“Network error during search: {e}” # 工具定义集合 TOOLS [ Tool( name“calculator”, description“执行加、减、乘、除运算。”, parameters{ “type”: “object”, “properties”: { “a”: {“type”: “number”, “description”: “第一个数字”}, “b”: {“type”: “number”, “description”: “第二个数字”}, “operation”: {“type”: “string”, “description”: “操作类型: add, subtract, multiply, divide”, “enum”: [“add”, “subtract”, “multiply”, “divide”]} }, “required”: [“a”, “b”, “operation”] }, functioncalculator ), Tool( name“get_current_time”, description“获取指定时区的当前时间。”, parameters{ “type”: “object”, “properties”: { “timezone”: {“type”: “string”, “description”: “时区名称例如 UTC, Asia/Shanghai”, “default”: “UTC”} }, “required”: [] }, functionget_current_time ), Tool( name“search_web”, description“根据查询词进行简单的网络信息检索。”, parameters{ “type”: “object”, “properties”: { “query”: {“type”: “string”, “description”: “搜索关键词”} }, “required”: [“query”] }, functionsearch_web ), ]4.2 编写主程序并运行最后在main.py中将所有部分组合起来创建一个可交互的 CLI 程序。# main.py import asyncio from agent_runtime import HarnessAgentRuntime from tools import TOOLS import sys async def main(): # 1. 初始化Agent Runtime system_prompt “”” 你是一个强大的AI助手可以调用工具来解决用户的问题。 请遵循以下规则 1. 仔细分析用户请求判断是否需要调用工具。 2. 如果需要请精确地调用最合适的工具并提供准确的参数。 3. 根据工具返回的结果组织清晰、有用的回复给用户。 4. 如果工具调用失败或结果不明确请向用户说明情况。 “”” agent HarnessAgentRuntime(system_promptsystem_prompt) # 2. 注册所有工具 for tool in TOOLS: agent.register_tool(tool) print(f“Agent initialized with {len(TOOLS)} tools: {[t.name for t in TOOLS]}”) # 3. 交互循环 print(“\n Harness Agent Started ) print(“Type ‘quit’ or ‘exit’ to end the conversation.\n”) while True: try: user_input input(“You: “).strip() if user_input.lower() in [“quit”, “exit”]: print(“Goodbye!”) break if not user_input: continue # 4. 处理用户输入 final_response await agent.process_user_input(user_input) print(f“\nAgent: {final_response}\n”) # 5. 可选重置迭代计数以备下次对话或不清除保持上下文 agent.state.iteration_count 0 # 注意这里没有清空 messages所以对话有历史记忆。 # 如果需要开启新对话可以执行agent.state.messages [agent.state.messages[0]] except KeyboardInterrupt: print(“\n\nInterrupted by user.”) break except Exception as e: print(f“\nAn unexpected error occurred: {e}”) # 可以选择重置Agent状态或退出 # agent.state.messages [agent.state.messages[0]] # agent.state.iteration_count 0 if __name__ “__main__”: asyncio.run(main())4.3 运行与验证启动程序python main.py进行对话测试测试工具选择与调用输入“计算 123 乘以 456 是多少”。观察控制台输出Agent 应识别出需要调用calculator工具并正确解析参数a123, b456, operationmultiply然后返回计算结果。测试多轮工具调用输入“先查一下‘Michael’这个名字的年龄估计然后告诉我现在上海的时间”。这是一个需要按顺序调用两个工具search_web和get_current_time的复杂请求。Agent 应在第一轮调用搜索工具在第二轮根据搜索结果和上下文调用时间工具。测试错误处理输入“用0除以5然后再加上10”。Agent 应成功调用计算器完成除法a0, b5, operationdivide得到结果0然后可能根据上下文决定是否需要再次调用计算器进行加法或者直接给出最终答案。测试纯对话输入“你好请介绍一下你自己。” Agent 应不调用任何工具直接基于系统提示词生成回复。预期成功运行的标志程序正常启动打印出已注册的工具列表。输入问题后控制台会显示[Iteration N]和Executing tool: ...的日志。最终能返回一个结合了工具执行结果的、连贯的自然语言答案。对于无法处理或工具出错的请求能返回明确的错误信息而不是崩溃。5. 关键配置、参数与常见问题排查构建并运行起一个基础 Agent 后需要理解其关键控制点并知道如何排查问题。5.1 关键配置参数说明下表列出了影响 Agent 行为的核心参数及其调整建议参数位置参数名含义默认值/示例调整建议与影响config.pyopenai_model使用的 LLM 模型gpt-3.5-turbo强烈建议使用gpt-4-turbo-preview或更高版本。GPT-4 系列在工具调用准确性和复杂推理上远优于 GPT-3.5。config.pyagent_temperature生成文本的随机性0.1工具调用场景下建议设为较低值0.1-0.3使模型输出更确定、更遵循指令减少“幻觉”调用。config.pyagent_max_iterationsAgent 最大循环次数10防止 Agent 陷入死循环。对于简单任务可设为 5复杂多步任务可设为 15-20。如果任务未完成就提前退出可适当调高。tools.pyTooldescription工具的功能描述“获取当前时间”至关重要。LLM 仅凭此描述决定是否及如何调用工具。描述应清晰、准确包含输入输出示例更佳。tools.pyToolparameters工具参数的 JSON Schema见上文示例定义必须精确。type,description,required,enum等字段能极大帮助 LLM 生成正确的参数。main.pysystem_prompt系统提示词见上文示例引导 Agent 的行为风格。可在此强调“必须调用工具”、“逐步思考”、“向用户解释过程”等。5.2 常见问题与排查路径在开发和运行过程中你可能会遇到以下典型问题问题1Agent 不调用工具直接回复现象对于明显需要工具的问题如“计算11”Agent 直接给出了文本答案控制台没有Executing tool日志。可能原因与排查工具描述不清检查tools.py中工具的description是否足够清晰能让 LLM 理解其用途。系统提示词引导不足在system_prompt中明确要求“请优先使用工具解决问题”。模型能力不足尝试切换到gpt-4-turbo-preview。GPT-3.5 的工具调用能力较弱。API 调用未传tools参数检查agent_runtime.py中_prepare_openai_tools方法是否正确生成列表并在client.chat.completions.create调用中传入了tools参数。问题2工具调用参数解析错误现象控制台显示调用了工具但日志显示Invalid arguments JSON或工具函数内部报参数类型错误。可能原因与排查JSON 格式错误LLM 生成的arguments字符串可能不是合法 JSON。检查agent_runtime.py中的json.loads是否被异常捕获。参数类型不匹配工具函数期望int但 LLM 传了string。在 Tool 的parametersJSON Schema 中严格定义type并使用enum限制可选值。缺少必需参数确保required字段列表正确并且 LLM 有能力推断出该参数例如从用户问题“计算圆的面积”推断出需要“半径”参数可能比较困难需要更精确的提示。问题3网络或外部 API 错误现象工具函数如search_web执行时报超时或 HTTP 错误。排查在工具函数内部添加更详细的错误日志。使用try...except包裹网络请求并返回友好的错误信息给 Agent使其能向用户解释。检查网络连接和代理设置如果公司网络需要。注意代码中不应出现任何相关配置或说明。问题4达到最大迭代次数后退出现象Agent 回复 “Agent stopped after reaching max iterations (10).”排查任务过于复杂当前任务可能需要超过 10 步的规划。适当增加agent_max_iterations。陷入循环Agent 可能在某一步出不来。检查工具返回的结果是否清晰能否让 LLM 做出下一步决策。可以在system_prompt中加入“如果你认为任务无法继续或已足够请直接给出最终答案并停止调用工具。”查看完整日志检查每一轮迭代中 LLM 的回复和工具调用的输入输出定位卡住的环节。6. 生产环境最佳实践与扩展方向上述代码是一个用于学习和验证的概念验证PoC实现。要将其用于更严肃的项目或生产环境需要考虑以下方面6.1 生产环境加固建议配置与密钥管理绝对不要将密钥提交到代码仓库。使用.env文件并通过环境变量或专业的密钥管理服务如 AWS Secrets Manager, HashiCorp Vault加载。将配置分类如LLM_CONFIG,TOOL_CONFIG,AGENT_CONFIG。增强错误处理与韧性重试机制对 LLM API 调用和外部工具调用添加指数退避重试。超时控制为每个工具调用和 LLM 调用设置合理的超时时间避免整个 Agent 挂起。熔断与降级如果某个工具持续失败应能暂时将其禁用熔断或提供降级方案。可观测性结构化日志使用structlog或loggingJSON Formatter 记录关键事件如对话开始、工具调用、LLM 请求、错误发生等并包含会话 ID、工具名、耗时等字段。监控与告警监控 Agent 的响应延迟、工具调用成功率、LLM Token 消耗等指标。状态持久化当前的AgentState保存在内存中服务器重启即丢失。生产环境需要将会话状态messages持久化到数据库如 Redis, PostgreSQL中支持多实例部署和会话恢复。工具管理动态注册/注销工具而无需重启服务。为工具添加权限控制和速率限制。6.2 架构扩展方向引入规划器Planner对于复杂任务可以让一个专门的 LLM 或模块先进行任务分解Plan生成步骤列表再由执行器Executor逐步调用工具完成。这比让单个 LLM 在循环中自行规划更可控。工具学习与发现构建一个工具库让 Agent 能够根据任务描述自动检索和组合最相关的工具而不是硬编码在代码中。长期记忆Memory除了对话历史引入向量数据库存储和检索过往对话的关键信息使 Agent 拥有跨会话的记忆能力。多模态工具扩展工具类型使其不仅能处理文本还能调用图像生成、语音合成、文档解析等多模态能力。评估与测试建立自动化测试框架用一系列标准问题测试 Agent 的工具调用准确性、回复质量和任务完成率确保迭代更新不会导致性能回退。构建一个成熟的 Harness Agent 系统是一个持续迭代的过程。本文提供的 24 小时快速启动方案旨在帮助你打通从想法到可运行原型的关键路径。理解 Agent Runtime 和 Tool Call 这两个核心支柱后你可以在此基础上根据实际业务需求逐步完善其健壮性、可扩展性和智能化水平最终打造出真正能够驾驭复杂工作流的 AI 智能体。