1. 项目概述从零理解 Agent Loop 与 Harness 的工程实践最近在 AI 应用开发领域特别是围绕大语言模型构建智能体时一个高频出现的词是“Agent Loop”。很多朋友觉得这个概念很玄乎好像涉及复杂的调度、记忆、工具调用等一系列高级功能。同时另一个词“Harness”也开始频繁出现常被描述为“包裹在 Agent 核心推理逻辑之外的基础设施层”。这听起来更抽象了它到底是干嘛的和 Agent 又是什么关系今天我们就用一个极简的实战项目来彻底搞懂这两个核心概念。我们的目标非常明确用大约 50 行清晰、核心的 Python 代码实现一个可运行、可扩展的 Agent Loop 最小内核并在此过程中深刻理解 Harness 扮演的角色。这不是一个玩具而是一个高度浓缩的工程原型。通过拆解它你会明白那些复杂的 Agent 框架如 LangChain、AutoGen 的底层思维是如何运转起来的以及为什么我们需要 Harness 这样的“基础设施层”。无论你是想自己动手搭建 AI 应用还是仅仅想理解市面上的 Agent 产品这个最小内核都能为你提供一个坚实、清晰的认知起点。简单来说你可以把这个项目看作是一次“心脏外科手术”——我们不关心皮肤、肌肉和骨骼那些花哨的 UI、复杂的管理功能而是直接剖开胸膛观察那个驱动一切跳动的“心脏”Agent Loop以及维持它稳定跳动的“起搏器和监控系统”Harness。理解了这颗最小的心脏你就能看懂任何复杂的身体结构。2. 核心概念拆解Agent Loop 与 Harness 到底是什么在动手写代码之前我们必须把几个关键概念掰开揉碎达成共识。这是避免后续讨论变成“鸡同鸭讲”的关键。2.1 Agent Loop智能体的“思考-行动”循环Agent Loop翻译过来是“智能体循环”。它描述的是一个自主智能体AI Agent与世界或用户交互的基本模式。这个循环通常包含以下几个核心阶段观察Observe智能体从环境中获取输入。这可能是用户的提问、传感器的数据、一段代码的执行结果等。思考Think智能体基于观察到的信息结合自身的记忆和目标进行内部推理决定下一步要做什么。在大模型时代这一步通常由 LLM 驱动。行动Act智能体执行思考后决定的动作。这个动作可以是调用一个工具如计算器、搜索引擎、API生成一段文本回复或者修改内部状态。观察结果Observe Result行动会对环境产生影响产生新的观察。这个结果被反馈回智能体开启下一个循环。这个过程会一直持续直到智能体认为任务完成达到终止条件。一个经典的例子是让 AI 写一份市场报告观察用户指令“写一份关于新能源汽车的市场报告。”思考LLM 分析任务决定需要先搜索最新数据、整理政策、分析竞争对手。行动调用“网络搜索”工具搜索“2024 新能源汽车销量”。观察结果获取到搜索结果的文本。思考LLM 分析搜索结果发现还需要政策信息。行动调用“网络搜索”工具搜索“2024 新能源汽车补贴政策”。… 如此循环直到报告生成。所以Agent Loop 的本质是一个状态机。它管理着智能体在“感知-决策-执行”这个循环中的状态流转。我们代码要实现的就是这个状态机的驱动引擎。2.2 Harness智能体的“赛车安全带与仪表盘”现在来说 Harness。这个词原意是“马具”或“安全带”在工程中引申为“控制系统”或“装备”。在 AI Agent 的语境下它的定义非常精准一套包裹在 AI Agent 核心推理逻辑之外的基础设施层。理解这一点至关重要Harness 不是 Agent也不替代 Agent 的思考LLM能力。你可以把它想象成 F1 赛车的“座舱系统”核心引擎Agent是车手LLM和发动机核心算法负责做出超车、刹车的决策和提供动力。座舱系统Harness包括安全带安全约束、仪表盘状态监控、方向盘和踏板输入控制、车载电脑逻辑调度、遥测系统数据记录。Harness 的作用它把强大的引擎和车手安全、可靠、高效地“绑”在一起确保车手能清晰看到转速、油温能精准控制方向并且一旦发生异常如油温过高系统能介入保护。没有它再好的车手和引擎也可能在第一个弯道就失控。对应到我们的 Agent 系统Harness 通常提供以下基础设施能力生命周期管理启动、暂停、停止、重启一个 Agent 任务。状态管理与持久化保存和加载 Agent 运行中的上下文、记忆、中间结果。工具调用与编排以标准化、安全的方式注册、管理和执行外部工具函数、API。流程控制与调度决定何时调用 LLM何时执行工具如何处理循环中的分支和异常。可观测性Observability记录详细的运行日志、追踪每个步骤的输入输出、监控耗时和 Token 使用量。安全与约束限制工具调用的权限过滤不安全的用户输入或模型输出设置超时和重试机制。Harness 和 Agent 的关系是“包裹”与“被包裹”是“基础设施”与“业务逻辑”的关系。一个设计良好的 Harness 能让开发者专注于设计 Agent 的“思考策略”Prompt 和逻辑而不用重复造轮子去处理工具调用、错误恢复、状态保存这些繁琐但必需的工程问题。2.3 为什么需要“最小内核”市面上成熟的框架如 LangChain功能强大但同时也非常复杂抽象层级多对于初学者来说就像面对一个黑箱难以理解其核心原理。“最小内核”的意义在于剥离所有非核心的、锦上添花的功能只保留最本质的驱动逻辑。通过亲手实现这个不足 50 行的内核你将获得穿透性理解不再被框架的复杂 API 所迷惑直接掌握 Agent 系统运转的“第一性原理”。完全的掌控感每一行代码你都清楚其作用可以轻松地修改、调试和扩展。定制的起点这个内核本身就是一个可用的原型你可以基于它只为自己的特定需求添加功能避免引入庞大框架的冗余和开销。3. 50 行代码实现最小 Agent Loop 内核下面我们将分步实现这个最小内核。我将使用纯 Python仅依赖openai库或兼容的 API 库来调用大模型确保极致的简洁和清晰。你可以用任何你熟悉的 LLM API 来替换。3.1 环境准备与核心类定义首先我们需要定义几个最核心的类。整个系统的骨架就在这里。# agent_core.py from typing import Any, Dict, List, Callable, Optional import json class Tool: 工具类代表 Agent 可以调用的一个外部函数或API。 def __init__(self, name: str, func: Callable, description: str, schema: Dict): self.name name self.func func self.description description # 给LLM看的工具描述 self.schema schema # 工具的参数JSON Schema用于引导LLM生成正确调用 def run(self, **kwargs) - str: 执行工具并返回字符串格式的结果。 try: result self.func(**kwargs) return str(result) except Exception as e: return fTool execution error: {e} class AgentState: Agent 的状态容器记录当前循环的上下文。 def __init__(self, initial_input: str): self.messages: List[Dict] [{role: user, content: initial_input}] self.observation: Optional[str] None # 上一轮行动的结果 self.final_answer: Optional[str] None # 最终答案如果已得出 self.is_finished: bool False # 循环是否应终止 class Harness: Harness 核心类驱动 Agent Loop 运转。 def __init__(self, llm_client, tools: List[Tool]): self.llm llm_client # 例如 OpenAI client self.tools {tool.name: tool for tool in tools} # 工具字典 self.max_turns 10 # 最大循环次数防止无限循环 def run(self, user_input: str) - str: 主运行方法输入用户问题返回最终答案。 state AgentState(user_input) for turn in range(self.max_turns): print(f\n Turn {turn 1} ) # 1. 思考让LLM基于当前状态决定下一步 thought, action self._think(state) print(fThought: {thought}) if action: print(fAction: {action}) # 2. 行动执行LLM决定的动作调用工具或生成回答 self._act(state, action) # 3. 检查是否应结束循环 if state.is_finished: print(Agent has finished its task.) break return state.final_answer or Agent reached max turns without final answer.代码解读与设计理由Tool类这是 Agent 与外界交互的桥梁。schema字段至关重要它用 JSON Schema 格式定义了工具的输入参数这将是引导 LLM 正确生成工具调用参数的关键。我们让run方法始终返回字符串是为了简化与 LLM 文本交互的接口。AgentState类这是一个简单的数据容器采用messages列表来维护与 LLM 的对话历史这是最通用和有效的方式。observation和final_answer分离是为了清晰区分中间结果和最终输出。Harness类这是大脑和总控。它持有 LLM 客户端和工具集。run方法实现了最核心的循环逻辑思考 - 行动 - 更新状态 - 判断终止。我们设置了max_turns作为一个重要的安全阀这是生产系统中防止成本失控和无限循环的必备措施。3.2 实现“思考”阶段让 LLM 做决策“思考”阶段是 Agent 的“大脑”所在。我们需要精心设计 Prompt让 LLM 能够理解当前状态、可用工具并做出合理的决策是调用工具还是直接回答。# 在 Harness 类中添加 _think 方法 class Harness: # ... __init__ 等方法 ... def _think(self, state: AgentState) - (str, Optional[Dict]): 基于当前状态调用LLM生成‘思考’和‘行动指令’。 # 构建系统提示词定义Agent的角色、能力和规则 system_prompt fYou are a helpful AI assistant that can use tools. Available tools: {self._format_tools_for_prompt()} You must respond in a strict JSON format with two keys: thought and action. - thought: Your reasoning about the current situation and what to do next. - action: Can be null if you have the final answer, or a JSON object with: - tool: tool name (must be one of the available tools) - args: arguments for the tool (must match its schema) If you have enough information to answer the users question directly, set action to null and put your final answer in the thought field. Your response must be valid JSON only. # 构建包含历史对话和最新观察的对话上下文 messages [{role: system, content: system_prompt}] messages.extend(state.messages) if state.observation: # 将上一轮的工具执行结果作为系统消息插入告知Agent messages.append({role: system, content: fObservation: {state.observation}}) try: response self.llm.chat.completions.create( modelgpt-3.5-turbo, # 或 gpt-4, claude 等 messagesmessages, temperature0.1, # 低温度保证决策的稳定性 response_format{type: json_object} # 强制JSON输出关键 ) llm_output response.choices[0].message.content decision json.loads(llm_output) thought decision.get(thought, ) action decision.get(action) return thought, action except json.JSONDecodeError as e: print(fLLM did not return valid JSON: {llm_output}. Error: {e}) return Failed to parse LLM response., None except Exception as e: print(fLLM call failed: {e}) return fLLM error: {e}, None def _format_tools_for_prompt(self) - str: 将工具列表格式化为LLM可读的字符串描述。 tool_descriptions [] for name, tool in self.tools.items(): desc f- {name}: {tool.description}. Args schema: {json.dumps(tool.schema)} tool_descriptions.append(desc) return \n.join(tool_descriptions)关键点与避坑指南结构化输出是生命线我们通过response_format{type: json_object}OpenAI API强制 LLM 返回 JSON。这是实现可靠程序化解析的关键避免了从自由文本中抽取信息的复杂性和不稳定性。对于不支持该参数的模型需要在 Prompt 中更严格地强调。系统提示词是“宪法”系统提示词定义了 Agent 的行为准则。这里明确规定了输出格式、工具使用规则和终止条件。描述越清晰Agent 行为越可控。将观察结果作为系统消息这是一个实用技巧。将工具执行结果以role: system的形式插入对话历史可以清晰地将其与用户输入和助理回复区分开帮助 LLM 理解这是“环境反馈”。错误处理必须对 LLM 调用失败和 JSON 解析失败进行处理。在实际系统中这里可能需要重试、降级策略或更复杂的错误恢复。3.3 实现“行动”阶段执行与状态更新“行动”阶段根据思考的结果要么执行工具要么生成最终答案并更新 Agent 的状态。# 在 Harness 类中添加 _act 方法 class Harness: # ... __init__, _think 等方法 ... def _act(self, state: AgentState, action: Optional[Dict]): 执行行动更新Agent状态。 if action is None: # LLM决定直接给出最终答案 # 我们假设_think方法返回的thought里包含了答案 state.final_answer state.messages[-1][content] # 这是一个简化处理 # 更健壮的做法在_think的返回中明确区分“推理过程”和“最终答案文本” # 这里为了极简我们用一个技巧将最后一条LLM的“thought”作为最终消息存入历史 if state.messages[-1][role] assistant: # 构造一个包含最终答案的助理消息 final_message {role: assistant, content: state.messages[-1][content]} state.messages.append(final_message) state.is_finished True print(Action: Give Final Answer.) return # 执行工具调用 tool_name action.get(tool) args action.get(args, {}) if not isinstance(args, dict): args {} if tool_name not in self.tools: state.observation fError: Tool {tool_name} not found. state.messages.append({role: system, content: state.observation}) return print(fExecuting tool: {tool_name} with args: {args}) tool self.tools[tool_name] # 执行工具 result tool.run(**args) state.observation result # 将工具执行结果记录到上下文中 state.messages.append({role: system, content: fObservation: {result}}) print(fTool Result: {result})行动逻辑的精髓分支判断action为None是循环的终止信号表示 Agent 认为可以给出最终答案了。这里的状态更新逻辑需要根据你的具体设计调整。上述代码是一个简化版本。工具执行与反馈执行工具后将结果存入state.observation并立即将其作为一条系统消息追加到state.messages中。这确保了在下一轮“思考”时LLM 能感知到这次行动的结果形成闭环。错误处理检查工具是否存在是必要的。更完善的实现还应验证参数是否符合 schema并处理工具执行时的异常。3.4 组装与运行一个完整的示例现在让我们创建几个简单的工具并运行一个完整的例子。# main.py from agent_core import Tool, Harness import openai import math from datetime import datetime # 1. 定义几个工具 def get_current_time(**kwargs) - str: 获取当前日期和时间。 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def calculator(expression: str, **kwargs) - str: 计算一个数学表达式。警告使用eval有安全风险仅用于演示。 try: # 严重警告在生产环境中绝对不要用eval直接执行用户或LLM提供的字符串。 # 这里仅为演示应替换为安全的数学表达式解析器如ast.literal_eval限制操作。 result eval(expression, {__builtins__: None}, {math: math}) return str(result) except Exception as e: return fCalculation error: {e} # 2. 实例化工具对象 tools [ Tool( nameget_current_time, funcget_current_time, descriptionGet the current date and time., schema{type: object, properties: {}, additionalProperties: False} # 此工具无参数 ), Tool( namecalculator, funccalculator, descriptionEvaluate a mathematical expression. Provide an expression string like 3 5 * 2., schema{ type: object, properties: { expression: {type: string, description: The mathematical expression to evaluate.} }, required: [expression], additionalProperties: False } ) ] # 3. 初始化LLM客户端和Harness client openai.OpenAI(api_keyyour-api-key) # 请替换为你的API Key harness Harness(llm_clientclient, toolstools) # 4. 运行一个任务 if __name__ __main__: # 任务1简单问答不需要工具 print( Task 1: Direct QA ) result1 harness.run(What is the capital of France?) print(fFinal Answer: {result1}\n) # 任务2需要调用工具的任务 print( Task 2: Task requiring tools ) result2 harness.run(What time is it now, and what is 15 raised to the power of 2?) print(fFinal Answer: {result2})运行这个示例你会看到类似以下的输出 Task 1: Direct QA Turn 1 Thought: The user is asking for a factual piece of information. I have the knowledge to answer this directly without using any tools. Action: None Action: Give Final Answer. Final Answer: The capital of France is Paris. Task 2: Task requiring tools Turn 1 Thought: The user is asking two questions: the current time and a calculation. I need to use tools. First, I should get the current time using the get_current_time tool. Then, I can calculate 15^2 using the calculator tool. Action: {tool: get_current_time, args: {}} Executing tool: get_current_time with args: {} Tool Result: 2024-05-27 10:30:15 Turn 2 Thought: I have the current time. Now I need to calculate 15 raised to the power of 2. The expression is 15 ** 2. Action: {tool: calculator, args: {expression: 15 ** 2}} Executing tool: calculator with args: {expression: 15 ** 2} Tool Result: 225 Turn 3 Thought: I have gathered all necessary information. The current time is 2024-05-27 10:30:15, and 15 raised to the power of 2 is 225. I can now provide the final answer. Action: None Action: Give Final Answer. Final Answer: The current time is 2024-05-27 10:30:15, and 15 raised to the power of 2 is 225.看一个能够自主规划、调用工具、并整合信息给出答案的 Agent其最核心的驱动循环就这样在 50 行左右的代码里跑通了。Harness我们的Harness类负责管理整个流程准备 Prompt、调用 LLM、解析决策、执行工具、维护状态、控制循环。而 Agent 的“智能”则来源于我们设计的 Prompt 和 LLM 本身。4. 从最小内核到生产系统Harness 的扩展方向我们实现的这个最小内核具备了最核心的“循环驱动”能力。但要将其用于实际生产Harness 还需要在以下几个关键方向上进行大幅增强这也正是成熟框架所提供的价值所在。4.1 增强状态管理与记忆当前的状态只是简单的消息列表。一个完整的系统可能需要长短期记忆分离将对话历史短期记忆与知识库、用户画像长期记忆分开管理。状态持久化支持将运行状态AgentState保存到数据库或文件以便任务中断后恢复。这对于运行时间长的任务至关重要。上下文窗口管理当对话历史超过 LLM 的上下文长度限制时需要智能地总结、压缩或选择性遗忘旧消息。这通常被称为“ConversationSummaryBufferMemory”或类似功能。# 扩展状态管理的简单示例 class PersistentAgentState(AgentState): def __init__(self, initial_input: str, session_id: str): super().__init__(initial_input) self.session_id session_id self.long_term_memory [] # 可持久化到外部的记忆片段 self.created_at datetime.now() def save_to_db(self): # 将状态序列化后存入数据库 pass classmethod def load_from_db(cls, session_id): # 从数据库加载并重建状态 pass4.2 强化工具调用与安全性当前的工具调用非常基础存在安全风险如eval。参数验证与类型转换严格根据schema验证 LLM 生成的参数并尝试进行类型转换如将字符串5转为整数5。工具权限与沙箱为工具划分权限等级对高风险工具如执行代码、访问网络进行沙箱隔离或二次确认。异步与并行工具调用支持同时调用多个不依赖的工具提升效率。工具组合与流程定义更复杂的工具组合逻辑如循环调用、条件分支这部分可以整合进 Prompt也可以通过 Harness 的流程控制来实现。4.3 完善流程控制与可观测性更精细的终止条件除了最大轮次还应支持基于 LLM 输出中的特定标记、用户中断信号、任务成功判定等多种终止条件。错误恢复与重试当工具调用失败或 LLM 返回格式错误时不应直接崩溃而应尝试重试、简化问题或向用户求助。全面的日志与追踪记录每一轮循环的输入、输出、耗时、Token 用量、工具调用详情。这是调试、优化和计费的基石。可以集成像OpenTelemetry这样的标准。超时控制为整个任务或单个 LLM 调用、工具调用设置超时防止 hanging。4.4 优化提示工程与模型交互动态提示构建根据任务类型、历史表现动态调整系统提示词例如在 Agent 多次尝试失败后加入“让我们一步步思考”的引导。多模型路由与降级根据任务难度、成本预算自动选择不同的模型如 GPT-4 用于复杂规划GPT-3.5 用于简单回复。输出解析与后处理在将 LLM 输出交给业务逻辑前进行清洗、格式化或敏感信息过滤。5. 常见问题与实战调试技巧在实际操作中你一定会遇到各种问题。以下是一些典型问题及其排查思路以及我从实战中总结的技巧。5.1 LLM 不遵循指令不返回 JSON 格式问题LLM 返回了自由文本导致json.loads()失败。排查与解决检查系统提示词确保提示词中明确、强硬地要求返回 JSON。使用“You MUST respond in JSON format”等措辞。利用 API 特性优先使用像 OpenAI 的response_format这样的原生参数它比提示词约束力强得多。输出后处理如果 API 不支持强制 JSON可以尝试用正则表达式或一个轻量级的解析器如json5库从文本中提取可能的 JSON 块并设置重试逻辑。降低 Temperature将temperature设为 0 或接近 0如 0.1减少输出的随机性。5.2 Agent 陷入无效循环或重复调用同一工具问题Agent 在一个简单问题上不断循环或者反复调用同一个工具却不推进任务。排查与解决检查观察结果是否被正确传递确保工具执行的结果observation被完整、清晰地放入了下一轮对话的上下文messages中。LLM 可能因为没看到结果而重复行动。增强提示词中的推理引导在系统提示词中加入“分析上一步的结果”、“如果上一步失败了尝试另一种方法”等引导。实现简单的循环检测在Harness中维护一个最近几次行动的历史记录。如果检测到完全相同的(thought, action)组合重复出现则中断循环并报错。引入“反思”步骤在每轮循环后让 LLM 对自己的进展做一个简短评估“是否更接近目标了”这有时能帮助它跳出死胡同。5.3 工具调用参数总是错误问题LLM 生成的参数格式不对或者缺少必需参数。排查与解决优化工具描述和 Schema确保description和schema清晰无歧义。对于复杂参数可以在描述中给出示例。在 Prompt 中提供示例在系统提示词里直接给出一个完整的、正确的工具调用 JSON 示例。实现参数验证与修正在_act方法中如果参数验证失败不要直接返回错误可以将错误信息连同原始请求一起再次发送给 LLM要求它修正参数。这实现了一个简单的自我修正循环。5.4 性能与成本问题问题任务运行慢Token 消耗高。实战技巧压缩历史消息对于长对话定期将旧消息总结成一段简短的摘要替换掉冗长的原始历史。这能显著节省 Token 并保持上下文。设置合理的max_turns根据任务复杂度设定一个保守的默认值如 5-10防止简单问题无意义地消耗资源。缓存对频繁出现的、结果不变的子查询如“今天的日期”或工具调用结果进行缓存。监控与告警在 Harness 中集成 Token 计数和耗时统计对异常高的消耗设置告警。这个 50 行的最小内核就像一张精确的解剖图清晰地展示了 Agent Loop 与 Harness 的骨骼和肌腱。它可能没有商业框架那么肌肉发达、衣着光鲜但它给了你最本质的理解和完全的控制权。当你下次再使用或评估一个复杂的 Agent 框架时不妨在心里把它映射回这个简单的循环它的_think在哪里它的_act是如何实现的它的状态又是如何管理的有了这个内核打底一切复杂都将变得透明和可理解。 50行代码实现Agent Loop与Harness：AI智能体核心架构实战解析
1. 项目概述从零理解 Agent Loop 与 Harness 的工程实践最近在 AI 应用开发领域特别是围绕大语言模型构建智能体时一个高频出现的词是“Agent Loop”。很多朋友觉得这个概念很玄乎好像涉及复杂的调度、记忆、工具调用等一系列高级功能。同时另一个词“Harness”也开始频繁出现常被描述为“包裹在 Agent 核心推理逻辑之外的基础设施层”。这听起来更抽象了它到底是干嘛的和 Agent 又是什么关系今天我们就用一个极简的实战项目来彻底搞懂这两个核心概念。我们的目标非常明确用大约 50 行清晰、核心的 Python 代码实现一个可运行、可扩展的 Agent Loop 最小内核并在此过程中深刻理解 Harness 扮演的角色。这不是一个玩具而是一个高度浓缩的工程原型。通过拆解它你会明白那些复杂的 Agent 框架如 LangChain、AutoGen 的底层思维是如何运转起来的以及为什么我们需要 Harness 这样的“基础设施层”。无论你是想自己动手搭建 AI 应用还是仅仅想理解市面上的 Agent 产品这个最小内核都能为你提供一个坚实、清晰的认知起点。简单来说你可以把这个项目看作是一次“心脏外科手术”——我们不关心皮肤、肌肉和骨骼那些花哨的 UI、复杂的管理功能而是直接剖开胸膛观察那个驱动一切跳动的“心脏”Agent Loop以及维持它稳定跳动的“起搏器和监控系统”Harness。理解了这颗最小的心脏你就能看懂任何复杂的身体结构。2. 核心概念拆解Agent Loop 与 Harness 到底是什么在动手写代码之前我们必须把几个关键概念掰开揉碎达成共识。这是避免后续讨论变成“鸡同鸭讲”的关键。2.1 Agent Loop智能体的“思考-行动”循环Agent Loop翻译过来是“智能体循环”。它描述的是一个自主智能体AI Agent与世界或用户交互的基本模式。这个循环通常包含以下几个核心阶段观察Observe智能体从环境中获取输入。这可能是用户的提问、传感器的数据、一段代码的执行结果等。思考Think智能体基于观察到的信息结合自身的记忆和目标进行内部推理决定下一步要做什么。在大模型时代这一步通常由 LLM 驱动。行动Act智能体执行思考后决定的动作。这个动作可以是调用一个工具如计算器、搜索引擎、API生成一段文本回复或者修改内部状态。观察结果Observe Result行动会对环境产生影响产生新的观察。这个结果被反馈回智能体开启下一个循环。这个过程会一直持续直到智能体认为任务完成达到终止条件。一个经典的例子是让 AI 写一份市场报告观察用户指令“写一份关于新能源汽车的市场报告。”思考LLM 分析任务决定需要先搜索最新数据、整理政策、分析竞争对手。行动调用“网络搜索”工具搜索“2024 新能源汽车销量”。观察结果获取到搜索结果的文本。思考LLM 分析搜索结果发现还需要政策信息。行动调用“网络搜索”工具搜索“2024 新能源汽车补贴政策”。… 如此循环直到报告生成。所以Agent Loop 的本质是一个状态机。它管理着智能体在“感知-决策-执行”这个循环中的状态流转。我们代码要实现的就是这个状态机的驱动引擎。2.2 Harness智能体的“赛车安全带与仪表盘”现在来说 Harness。这个词原意是“马具”或“安全带”在工程中引申为“控制系统”或“装备”。在 AI Agent 的语境下它的定义非常精准一套包裹在 AI Agent 核心推理逻辑之外的基础设施层。理解这一点至关重要Harness 不是 Agent也不替代 Agent 的思考LLM能力。你可以把它想象成 F1 赛车的“座舱系统”核心引擎Agent是车手LLM和发动机核心算法负责做出超车、刹车的决策和提供动力。座舱系统Harness包括安全带安全约束、仪表盘状态监控、方向盘和踏板输入控制、车载电脑逻辑调度、遥测系统数据记录。Harness 的作用它把强大的引擎和车手安全、可靠、高效地“绑”在一起确保车手能清晰看到转速、油温能精准控制方向并且一旦发生异常如油温过高系统能介入保护。没有它再好的车手和引擎也可能在第一个弯道就失控。对应到我们的 Agent 系统Harness 通常提供以下基础设施能力生命周期管理启动、暂停、停止、重启一个 Agent 任务。状态管理与持久化保存和加载 Agent 运行中的上下文、记忆、中间结果。工具调用与编排以标准化、安全的方式注册、管理和执行外部工具函数、API。流程控制与调度决定何时调用 LLM何时执行工具如何处理循环中的分支和异常。可观测性Observability记录详细的运行日志、追踪每个步骤的输入输出、监控耗时和 Token 使用量。安全与约束限制工具调用的权限过滤不安全的用户输入或模型输出设置超时和重试机制。Harness 和 Agent 的关系是“包裹”与“被包裹”是“基础设施”与“业务逻辑”的关系。一个设计良好的 Harness 能让开发者专注于设计 Agent 的“思考策略”Prompt 和逻辑而不用重复造轮子去处理工具调用、错误恢复、状态保存这些繁琐但必需的工程问题。2.3 为什么需要“最小内核”市面上成熟的框架如 LangChain功能强大但同时也非常复杂抽象层级多对于初学者来说就像面对一个黑箱难以理解其核心原理。“最小内核”的意义在于剥离所有非核心的、锦上添花的功能只保留最本质的驱动逻辑。通过亲手实现这个不足 50 行的内核你将获得穿透性理解不再被框架的复杂 API 所迷惑直接掌握 Agent 系统运转的“第一性原理”。完全的掌控感每一行代码你都清楚其作用可以轻松地修改、调试和扩展。定制的起点这个内核本身就是一个可用的原型你可以基于它只为自己的特定需求添加功能避免引入庞大框架的冗余和开销。3. 50 行代码实现最小 Agent Loop 内核下面我们将分步实现这个最小内核。我将使用纯 Python仅依赖openai库或兼容的 API 库来调用大模型确保极致的简洁和清晰。你可以用任何你熟悉的 LLM API 来替换。3.1 环境准备与核心类定义首先我们需要定义几个最核心的类。整个系统的骨架就在这里。# agent_core.py from typing import Any, Dict, List, Callable, Optional import json class Tool: 工具类代表 Agent 可以调用的一个外部函数或API。 def __init__(self, name: str, func: Callable, description: str, schema: Dict): self.name name self.func func self.description description # 给LLM看的工具描述 self.schema schema # 工具的参数JSON Schema用于引导LLM生成正确调用 def run(self, **kwargs) - str: 执行工具并返回字符串格式的结果。 try: result self.func(**kwargs) return str(result) except Exception as e: return fTool execution error: {e} class AgentState: Agent 的状态容器记录当前循环的上下文。 def __init__(self, initial_input: str): self.messages: List[Dict] [{role: user, content: initial_input}] self.observation: Optional[str] None # 上一轮行动的结果 self.final_answer: Optional[str] None # 最终答案如果已得出 self.is_finished: bool False # 循环是否应终止 class Harness: Harness 核心类驱动 Agent Loop 运转。 def __init__(self, llm_client, tools: List[Tool]): self.llm llm_client # 例如 OpenAI client self.tools {tool.name: tool for tool in tools} # 工具字典 self.max_turns 10 # 最大循环次数防止无限循环 def run(self, user_input: str) - str: 主运行方法输入用户问题返回最终答案。 state AgentState(user_input) for turn in range(self.max_turns): print(f\n Turn {turn 1} ) # 1. 思考让LLM基于当前状态决定下一步 thought, action self._think(state) print(fThought: {thought}) if action: print(fAction: {action}) # 2. 行动执行LLM决定的动作调用工具或生成回答 self._act(state, action) # 3. 检查是否应结束循环 if state.is_finished: print(Agent has finished its task.) break return state.final_answer or Agent reached max turns without final answer.代码解读与设计理由Tool类这是 Agent 与外界交互的桥梁。schema字段至关重要它用 JSON Schema 格式定义了工具的输入参数这将是引导 LLM 正确生成工具调用参数的关键。我们让run方法始终返回字符串是为了简化与 LLM 文本交互的接口。AgentState类这是一个简单的数据容器采用messages列表来维护与 LLM 的对话历史这是最通用和有效的方式。observation和final_answer分离是为了清晰区分中间结果和最终输出。Harness类这是大脑和总控。它持有 LLM 客户端和工具集。run方法实现了最核心的循环逻辑思考 - 行动 - 更新状态 - 判断终止。我们设置了max_turns作为一个重要的安全阀这是生产系统中防止成本失控和无限循环的必备措施。3.2 实现“思考”阶段让 LLM 做决策“思考”阶段是 Agent 的“大脑”所在。我们需要精心设计 Prompt让 LLM 能够理解当前状态、可用工具并做出合理的决策是调用工具还是直接回答。# 在 Harness 类中添加 _think 方法 class Harness: # ... __init__ 等方法 ... def _think(self, state: AgentState) - (str, Optional[Dict]): 基于当前状态调用LLM生成‘思考’和‘行动指令’。 # 构建系统提示词定义Agent的角色、能力和规则 system_prompt fYou are a helpful AI assistant that can use tools. Available tools: {self._format_tools_for_prompt()} You must respond in a strict JSON format with two keys: thought and action. - thought: Your reasoning about the current situation and what to do next. - action: Can be null if you have the final answer, or a JSON object with: - tool: tool name (must be one of the available tools) - args: arguments for the tool (must match its schema) If you have enough information to answer the users question directly, set action to null and put your final answer in the thought field. Your response must be valid JSON only. # 构建包含历史对话和最新观察的对话上下文 messages [{role: system, content: system_prompt}] messages.extend(state.messages) if state.observation: # 将上一轮的工具执行结果作为系统消息插入告知Agent messages.append({role: system, content: fObservation: {state.observation}}) try: response self.llm.chat.completions.create( modelgpt-3.5-turbo, # 或 gpt-4, claude 等 messagesmessages, temperature0.1, # 低温度保证决策的稳定性 response_format{type: json_object} # 强制JSON输出关键 ) llm_output response.choices[0].message.content decision json.loads(llm_output) thought decision.get(thought, ) action decision.get(action) return thought, action except json.JSONDecodeError as e: print(fLLM did not return valid JSON: {llm_output}. Error: {e}) return Failed to parse LLM response., None except Exception as e: print(fLLM call failed: {e}) return fLLM error: {e}, None def _format_tools_for_prompt(self) - str: 将工具列表格式化为LLM可读的字符串描述。 tool_descriptions [] for name, tool in self.tools.items(): desc f- {name}: {tool.description}. Args schema: {json.dumps(tool.schema)} tool_descriptions.append(desc) return \n.join(tool_descriptions)关键点与避坑指南结构化输出是生命线我们通过response_format{type: json_object}OpenAI API强制 LLM 返回 JSON。这是实现可靠程序化解析的关键避免了从自由文本中抽取信息的复杂性和不稳定性。对于不支持该参数的模型需要在 Prompt 中更严格地强调。系统提示词是“宪法”系统提示词定义了 Agent 的行为准则。这里明确规定了输出格式、工具使用规则和终止条件。描述越清晰Agent 行为越可控。将观察结果作为系统消息这是一个实用技巧。将工具执行结果以role: system的形式插入对话历史可以清晰地将其与用户输入和助理回复区分开帮助 LLM 理解这是“环境反馈”。错误处理必须对 LLM 调用失败和 JSON 解析失败进行处理。在实际系统中这里可能需要重试、降级策略或更复杂的错误恢复。3.3 实现“行动”阶段执行与状态更新“行动”阶段根据思考的结果要么执行工具要么生成最终答案并更新 Agent 的状态。# 在 Harness 类中添加 _act 方法 class Harness: # ... __init__, _think 等方法 ... def _act(self, state: AgentState, action: Optional[Dict]): 执行行动更新Agent状态。 if action is None: # LLM决定直接给出最终答案 # 我们假设_think方法返回的thought里包含了答案 state.final_answer state.messages[-1][content] # 这是一个简化处理 # 更健壮的做法在_think的返回中明确区分“推理过程”和“最终答案文本” # 这里为了极简我们用一个技巧将最后一条LLM的“thought”作为最终消息存入历史 if state.messages[-1][role] assistant: # 构造一个包含最终答案的助理消息 final_message {role: assistant, content: state.messages[-1][content]} state.messages.append(final_message) state.is_finished True print(Action: Give Final Answer.) return # 执行工具调用 tool_name action.get(tool) args action.get(args, {}) if not isinstance(args, dict): args {} if tool_name not in self.tools: state.observation fError: Tool {tool_name} not found. state.messages.append({role: system, content: state.observation}) return print(fExecuting tool: {tool_name} with args: {args}) tool self.tools[tool_name] # 执行工具 result tool.run(**args) state.observation result # 将工具执行结果记录到上下文中 state.messages.append({role: system, content: fObservation: {result}}) print(fTool Result: {result})行动逻辑的精髓分支判断action为None是循环的终止信号表示 Agent 认为可以给出最终答案了。这里的状态更新逻辑需要根据你的具体设计调整。上述代码是一个简化版本。工具执行与反馈执行工具后将结果存入state.observation并立即将其作为一条系统消息追加到state.messages中。这确保了在下一轮“思考”时LLM 能感知到这次行动的结果形成闭环。错误处理检查工具是否存在是必要的。更完善的实现还应验证参数是否符合 schema并处理工具执行时的异常。3.4 组装与运行一个完整的示例现在让我们创建几个简单的工具并运行一个完整的例子。# main.py from agent_core import Tool, Harness import openai import math from datetime import datetime # 1. 定义几个工具 def get_current_time(**kwargs) - str: 获取当前日期和时间。 return datetime.now().strftime(%Y-%m-%d %H:%M:%S) def calculator(expression: str, **kwargs) - str: 计算一个数学表达式。警告使用eval有安全风险仅用于演示。 try: # 严重警告在生产环境中绝对不要用eval直接执行用户或LLM提供的字符串。 # 这里仅为演示应替换为安全的数学表达式解析器如ast.literal_eval限制操作。 result eval(expression, {__builtins__: None}, {math: math}) return str(result) except Exception as e: return fCalculation error: {e} # 2. 实例化工具对象 tools [ Tool( nameget_current_time, funcget_current_time, descriptionGet the current date and time., schema{type: object, properties: {}, additionalProperties: False} # 此工具无参数 ), Tool( namecalculator, funccalculator, descriptionEvaluate a mathematical expression. Provide an expression string like 3 5 * 2., schema{ type: object, properties: { expression: {type: string, description: The mathematical expression to evaluate.} }, required: [expression], additionalProperties: False } ) ] # 3. 初始化LLM客户端和Harness client openai.OpenAI(api_keyyour-api-key) # 请替换为你的API Key harness Harness(llm_clientclient, toolstools) # 4. 运行一个任务 if __name__ __main__: # 任务1简单问答不需要工具 print( Task 1: Direct QA ) result1 harness.run(What is the capital of France?) print(fFinal Answer: {result1}\n) # 任务2需要调用工具的任务 print( Task 2: Task requiring tools ) result2 harness.run(What time is it now, and what is 15 raised to the power of 2?) print(fFinal Answer: {result2})运行这个示例你会看到类似以下的输出 Task 1: Direct QA Turn 1 Thought: The user is asking for a factual piece of information. I have the knowledge to answer this directly without using any tools. Action: None Action: Give Final Answer. Final Answer: The capital of France is Paris. Task 2: Task requiring tools Turn 1 Thought: The user is asking two questions: the current time and a calculation. I need to use tools. First, I should get the current time using the get_current_time tool. Then, I can calculate 15^2 using the calculator tool. Action: {tool: get_current_time, args: {}} Executing tool: get_current_time with args: {} Tool Result: 2024-05-27 10:30:15 Turn 2 Thought: I have the current time. Now I need to calculate 15 raised to the power of 2. The expression is 15 ** 2. Action: {tool: calculator, args: {expression: 15 ** 2}} Executing tool: calculator with args: {expression: 15 ** 2} Tool Result: 225 Turn 3 Thought: I have gathered all necessary information. The current time is 2024-05-27 10:30:15, and 15 raised to the power of 2 is 225. I can now provide the final answer. Action: None Action: Give Final Answer. Final Answer: The current time is 2024-05-27 10:30:15, and 15 raised to the power of 2 is 225.看一个能够自主规划、调用工具、并整合信息给出答案的 Agent其最核心的驱动循环就这样在 50 行左右的代码里跑通了。Harness我们的Harness类负责管理整个流程准备 Prompt、调用 LLM、解析决策、执行工具、维护状态、控制循环。而 Agent 的“智能”则来源于我们设计的 Prompt 和 LLM 本身。4. 从最小内核到生产系统Harness 的扩展方向我们实现的这个最小内核具备了最核心的“循环驱动”能力。但要将其用于实际生产Harness 还需要在以下几个关键方向上进行大幅增强这也正是成熟框架所提供的价值所在。4.1 增强状态管理与记忆当前的状态只是简单的消息列表。一个完整的系统可能需要长短期记忆分离将对话历史短期记忆与知识库、用户画像长期记忆分开管理。状态持久化支持将运行状态AgentState保存到数据库或文件以便任务中断后恢复。这对于运行时间长的任务至关重要。上下文窗口管理当对话历史超过 LLM 的上下文长度限制时需要智能地总结、压缩或选择性遗忘旧消息。这通常被称为“ConversationSummaryBufferMemory”或类似功能。# 扩展状态管理的简单示例 class PersistentAgentState(AgentState): def __init__(self, initial_input: str, session_id: str): super().__init__(initial_input) self.session_id session_id self.long_term_memory [] # 可持久化到外部的记忆片段 self.created_at datetime.now() def save_to_db(self): # 将状态序列化后存入数据库 pass classmethod def load_from_db(cls, session_id): # 从数据库加载并重建状态 pass4.2 强化工具调用与安全性当前的工具调用非常基础存在安全风险如eval。参数验证与类型转换严格根据schema验证 LLM 生成的参数并尝试进行类型转换如将字符串5转为整数5。工具权限与沙箱为工具划分权限等级对高风险工具如执行代码、访问网络进行沙箱隔离或二次确认。异步与并行工具调用支持同时调用多个不依赖的工具提升效率。工具组合与流程定义更复杂的工具组合逻辑如循环调用、条件分支这部分可以整合进 Prompt也可以通过 Harness 的流程控制来实现。4.3 完善流程控制与可观测性更精细的终止条件除了最大轮次还应支持基于 LLM 输出中的特定标记、用户中断信号、任务成功判定等多种终止条件。错误恢复与重试当工具调用失败或 LLM 返回格式错误时不应直接崩溃而应尝试重试、简化问题或向用户求助。全面的日志与追踪记录每一轮循环的输入、输出、耗时、Token 用量、工具调用详情。这是调试、优化和计费的基石。可以集成像OpenTelemetry这样的标准。超时控制为整个任务或单个 LLM 调用、工具调用设置超时防止 hanging。4.4 优化提示工程与模型交互动态提示构建根据任务类型、历史表现动态调整系统提示词例如在 Agent 多次尝试失败后加入“让我们一步步思考”的引导。多模型路由与降级根据任务难度、成本预算自动选择不同的模型如 GPT-4 用于复杂规划GPT-3.5 用于简单回复。输出解析与后处理在将 LLM 输出交给业务逻辑前进行清洗、格式化或敏感信息过滤。5. 常见问题与实战调试技巧在实际操作中你一定会遇到各种问题。以下是一些典型问题及其排查思路以及我从实战中总结的技巧。5.1 LLM 不遵循指令不返回 JSON 格式问题LLM 返回了自由文本导致json.loads()失败。排查与解决检查系统提示词确保提示词中明确、强硬地要求返回 JSON。使用“You MUST respond in JSON format”等措辞。利用 API 特性优先使用像 OpenAI 的response_format这样的原生参数它比提示词约束力强得多。输出后处理如果 API 不支持强制 JSON可以尝试用正则表达式或一个轻量级的解析器如json5库从文本中提取可能的 JSON 块并设置重试逻辑。降低 Temperature将temperature设为 0 或接近 0如 0.1减少输出的随机性。5.2 Agent 陷入无效循环或重复调用同一工具问题Agent 在一个简单问题上不断循环或者反复调用同一个工具却不推进任务。排查与解决检查观察结果是否被正确传递确保工具执行的结果observation被完整、清晰地放入了下一轮对话的上下文messages中。LLM 可能因为没看到结果而重复行动。增强提示词中的推理引导在系统提示词中加入“分析上一步的结果”、“如果上一步失败了尝试另一种方法”等引导。实现简单的循环检测在Harness中维护一个最近几次行动的历史记录。如果检测到完全相同的(thought, action)组合重复出现则中断循环并报错。引入“反思”步骤在每轮循环后让 LLM 对自己的进展做一个简短评估“是否更接近目标了”这有时能帮助它跳出死胡同。5.3 工具调用参数总是错误问题LLM 生成的参数格式不对或者缺少必需参数。排查与解决优化工具描述和 Schema确保description和schema清晰无歧义。对于复杂参数可以在描述中给出示例。在 Prompt 中提供示例在系统提示词里直接给出一个完整的、正确的工具调用 JSON 示例。实现参数验证与修正在_act方法中如果参数验证失败不要直接返回错误可以将错误信息连同原始请求一起再次发送给 LLM要求它修正参数。这实现了一个简单的自我修正循环。5.4 性能与成本问题问题任务运行慢Token 消耗高。实战技巧压缩历史消息对于长对话定期将旧消息总结成一段简短的摘要替换掉冗长的原始历史。这能显著节省 Token 并保持上下文。设置合理的max_turns根据任务复杂度设定一个保守的默认值如 5-10防止简单问题无意义地消耗资源。缓存对频繁出现的、结果不变的子查询如“今天的日期”或工具调用结果进行缓存。监控与告警在 Harness 中集成 Token 计数和耗时统计对异常高的消耗设置告警。这个 50 行的最小内核就像一张精确的解剖图清晰地展示了 Agent Loop 与 Harness 的骨骼和肌腱。它可能没有商业框架那么肌肉发达、衣着光鲜但它给了你最本质的理解和完全的控制权。当你下次再使用或评估一个复杂的 Agent 框架时不妨在心里把它映射回这个简单的循环它的_think在哪里它的_act是如何实现的它的状态又是如何管理的有了这个内核打底一切复杂都将变得透明和可理解。