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 f"Tool 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(f"Thought: {thought}") if action: print(f"Action: {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 = f"""You 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 user's 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": f"Observation: {state.observation}"}) try: response = self.llm.chat.completions.create( model="gpt-3.5-turbo", # 或 gpt-4, claude 等 messages=messages, temperature=0.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(f"LLM did not return valid JSON: {llm_output}. Error: {e}") return "Failed to parse LLM response.", None except Exception as e: print(f"LLM call failed: {e}") return f"LLM 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 = f"Error: Tool '{tool_name}' not found." state.messages.append({"role": "system", "content": state.observation}) return print(f"Executing tool: {tool_name} with args: {args}") tool = self.tools[tool_name] # 执行工具 result = tool.run(**args) state.observation = result # 将工具执行结果记录到上下文中 state.messages.append({"role": "system", "content": f"Observation: {result}"}) print(f"Tool 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 f"Calculation error: {e}" # 2. 实例化工具对象 tools = [ Tool( name="get_current_time", func=get_current_time, description="Get the current date and time.", schema={"type": "object", "properties": {}, "additionalProperties": False} # 此工具无参数 ), Tool( name="calculator", func=calculator, description="Evaluate 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_key="your-api-key") # 请替换为你的API Key harness = Harness(llm_client=client, tools=tools) # 4. 运行一个任务 if __name__ == "__main__": # 任务1：简单问答，不需要工具 print("=== Task 1: Direct Q&A ===") result1 = harness.run("What is the capital of France?") print(f"Final 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(f"Final Answer: {result2}")

运行这个示例，你会看到类似以下的输出：

=== Task 1: Direct Q&A === === 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): # 从数据库加载并重建状态 pass

4.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()失败。
排查与解决：
1. 检查系统提示词：确保提示词中明确、强硬地要求返回 JSON。使用“You MUST respond in JSON format”等措辞。
2. 利用 API 特性：优先使用像 OpenAI 的response_format这样的原生参数，它比提示词约束力强得多。
3. 输出后处理：如果 API 不支持强制 JSON，可以尝试用正则表达式或一个轻量级的解析器（如json5库）从文本中提取可能的 JSON 块，并设置重试逻辑。
4. 降低 Temperature：将temperature设为 0 或接近 0（如 0.1），减少输出的随机性。

5.2 Agent 陷入无效循环或重复调用同一工具

问题：Agent 在一个简单问题上不断循环，或者反复调用同一个工具却不推进任务。
排查与解决：
1. 检查观察结果是否被正确传递：确保工具执行的结果observation被完整、清晰地放入了下一轮对话的上下文（messages）中。LLM 可能因为没看到结果而重复行动。
2. 增强提示词中的推理引导：在系统提示词中加入“分析上一步的结果”、“如果上一步失败了，尝试另一种方法”等引导。
3. 实现简单的循环检测：在Harness中维护一个最近几次行动的历史记录。如果检测到完全相同的(thought, action)组合重复出现，则中断循环并报错。
4. 引入“反思”步骤：在每轮循环后，让 LLM 对自己的进展做一个简短评估（“是否更接近目标了？”），这有时能帮助它跳出死胡同。

5.3 工具调用参数总是错误

问题：LLM 生成的参数格式不对，或者缺少必需参数。
排查与解决：
1. 优化工具描述和 Schema：确保description和schema清晰无歧义。对于复杂参数，可以在描述中给出示例。
2. 在 Prompt 中提供示例：在系统提示词里，直接给出一个完整的、正确的工具调用 JSON 示例。
3. 实现参数验证与修正：在_act方法中，如果参数验证失败，不要直接返回错误，可以将错误信息连同原始请求一起，再次发送给 LLM，要求它修正参数。这实现了一个简单的自我修正循环。

5.4 性能与成本问题

问题：任务运行慢，Token 消耗高。
实战技巧：
- 压缩历史消息：对于长对话，定期将旧消息总结成一段简短的摘要，替换掉冗长的原始历史。这能显著节省 Token 并保持上下文。
- 设置合理的max_turns：根据任务复杂度设定一个保守的默认值（如 5-10），防止简单问题无意义地消耗资源。
- 缓存：对频繁出现的、结果不变的子查询（如“今天的日期”）或工具调用结果进行缓存。
- 监控与告警：在 Harness 中集成 Token 计数和耗时统计，对异常高的消耗设置告警。

这个 50 行的最小内核，就像一张精确的解剖图，清晰地展示了 Agent Loop 与 Harness 的骨骼和肌腱。它可能没有商业框架那么肌肉发达、衣着光鲜，但它给了你最本质的理解和完全的控制权。当你下次再使用或评估一个复杂的 Agent 框架时，不妨在心里把它映射回这个简单的循环：它的_think在哪里？它的_act是如何实现的？它的状态又是如何管理的？有了这个内核打底，一切复杂都将变得透明和可理解。