1. 项目概述从零理解 Agent Loop 与 Harness 内核最近在 AI 应用开发圈里一个叫 “harness” 的词热度挺高经常和 “Agent”、“Agent Loop” 这些概念一起出现。很多刚接触的朋友可能会有点懵这到底是个新框架还是个新工具和 LangChain、AutoGen 这些已有的 Agent 开发库又有什么区别我自己在尝试构建一些 AI 应用时也深感于现有框架的庞大和复杂——它们功能强大但学习曲线陡峭内部抽象层多有时候想实现一个简单的“思考-行动”循环却感觉被框架“绑架”了调试起来像在迷宫里找路。这个名为“50 行跑通一个 Agent Loopharness 的最小内核”的项目恰恰击中了这个痛点。它不是一个要取代现有框架的巨无霸而是一个概念验证和深度理解工具。它的目标很纯粹剥离所有外围的、花哨的功能用极简的代码目标50行左右揭示一个 AI Agent 最核心的运转机制——即那个著名的“感知-思考-行动”循环Agent Loop并展示harness在这一过程中扮演的基础设施层角色。通过亲手实现这个最小内核我们能透彻理解 Agent 如何与工具交互、如何管理状态、以及harness如何像“缰绳”一样规范而不束缚 Agent 的推理逻辑。这对于无论是想深入 Agent 原理的研究者还是希望获得更高可控性的应用开发者都是一次绝佳的“拆解”练习。2. 核心概念拆解Agent, Loop, Harness 分别是什么在动手写代码之前我们必须把几个关键术语掰开揉碎讲清楚。很多人混淆它们正是因为没理解各自的职责边界。2.1 Agent拥有推理能力的“大脑”在 AI 语境下一个 Agent智能体通常指一个能够感知环境、进行推理、并执行行动以达到目标的软件实体。它的核心是一个语言模型LLM比如 GPT-4、Claude 或者开源的 Llama 系列。这个 LLM 就是 Agent 的“大脑”负责理解输入用户问题、工具执行结果、分析现状、并做出决策下一步该调用哪个工具或者直接给出最终答案。但一个光秃秃的 LLM 并不是一个可用的 Agent。它需要被“装备”起来。比如你问 LLM“现在旧金山的天气怎么样” LLM 可能知道旧金山但它无法获取实时数据。它需要调用一个“查询天气”的工具Tool。Agent 的本质就是 LLM 加上它可用的工具集以及驱动它选择工具的逻辑。2.2 Agent Loop智能体运转的“心跳”Agent Loop智能体循环描述的就是上述“感知-思考-行动”过程的周而复始。它是 Agent 能够处理复杂、多步骤任务的关键。一个典型的循环步骤如下观察ObservationAgent 接收当前的环境状态。这可能是用户的初始问题也可能是上一个工具执行后返回的结果。思考ThoughtAgent 的“大脑”LLM基于当前观察进行推理。它需要决定任务完成了吗如果没完成下一步应该做什么应该调用哪个工具调用时需要什么参数行动Action根据思考的结果Agent 执行一个动作。最常见的就是调用一个工具如执行代码、搜索网络、查询数据库并传入必要的参数。获取结果Result工具执行完毕返回一个新的观察结果。这个结果连同历史记录一起成为下一轮循环的输入。这个 Loop 会一直运行直到 LLM 认为任务已经完成并输出最终的答案Final Answer。这个过程就像人的思考过程看到问题观察想想该怎么办思考动手去做行动看看效果如何新观察再继续思考...直到问题解决。2.3 Harness规范智能体的“基础设施”与“缰绳”这是最容易产生误解的地方。根据社区讨论和其设计哲学harness不是Agent 本身也不是替代 LLM 推理逻辑的东西。你可以把harness理解为“包裹在 AI Agent 核心推理逻辑之外的基础设施层”或“一套开发范式与约束框架”。它的核心思想是“规范而非替代”。想象一下驯马马LLM拥有强大的力量和奔跑能力推理能力但要让它按照既定路线安全、高效地拉车你需要一套缰绳、马鞍和马车harness。这套装备不代替马去决定往哪跑但它定义了马如何被控制、如何与车连接、以及行进的规则。具体到技术层面一个harness可能提供以下基础设施生命周期管理负责启动、暂停、重置 Agent 的一次任务会话。工具Tools的注册与发现提供一个标准化的方式让 Agent 知道它“手头”有哪些工具可用以及如何调用它们。状态State管理维护 Agent 在整个 Loop 运行过程中的上下文包括对话历史、中间结果、当前目标等。这通常比简单的聊天历史更结构化。执行引擎Orchestration驱动整个 Agent Loop 的运转。它负责在合适的时机调用 LLM 进行“思考”然后执行“行动”再将结果反馈回去形成循环。可观测性Observability钩子提供接口让我们能在 Loop 的关键节点思考前、行动后等插入日志、监控或干预逻辑方便调试和优化。错误处理与安全边界当工具调用失败或 LLM 输出不符合预期时提供 fallback 机制或安全中断防止 Agent 陷入死循环或执行危险操作。harness与 LangChain 等框架的关键区别在于关注点和抽象层级。LangChain 是一个“全家桶”它试图提供从数据加载、向量存储、链Chain构建到 Agent 创建的一切其 Agent 执行器Agent Executor已经是一个比较重的、封装好的harness实现。而harness的理念更底层、更聚焦它只关心如何定义和运行那个最核心的 Loop至于你的 LLM 来自 OpenAI 还是 Azure你的工具是自定义的还是第三方的它只提供接入标准不做过多的捆绑。这带来了更高的灵活性和可控性。3. 最小内核设计如何用 50 行代码勾勒出核心理解了概念我们来看如何用极简的代码实现这个内核。我们的设计目标是清晰而非功能完备。我们会用 Python 来实现。3.1 定义核心数据模型首先我们需要定义在 Loop 中流转的核心数据对象。这就像定义游戏规则。from typing import Any, Dict, List, Optional, Callable from dataclasses import dataclass from enum import Enum class AgentState: Agent 的运行时状态。 def __init__(self): self.observation: str # 当前的观察输入或工具结果 self.thought: str # LLM 产生的思考 self.action: str # 要执行的动作通常是工具名 self.action_input: Dict[str, Any] {} # 动作的输入参数 self.history: List[Dict[str, str]] [] # 历史记录用于构造LLM上下文 class Tool: 工具的抽象。 def __init__(self, name: str, func: Callable, description: str): self.name name self.func func self.description description def run(self, **kwargs) - str: return str(self.func(**kwargs))设计解析AgentState是一个简单的类持有 Loop 中每一步的关键信息。history列表用于积累所有的(thought, action, observation)三元组这是构造给 LLM 的提示词Prompt的关键材料。Tool类是对一个可调用功能的封装。name和description至关重要因为它们会被拼接到给 LLM 的提示词中帮助 LLM 理解这个工具是干什么的、该怎么用。3.2 构建 Harness 骨架接下来我们构建Harness类它是整个基础设施的核心。class Harness: Agent Loop 的基础设施层最小实现。 def __init__(self, llm_client: Any, tools: List[Tool]): self.llm llm_client # 一个LLM调用客户端任何能生成文本的接口都行 self.tools {tool.name: tool for tool in tools} self.state AgentState() def _build_prompt(self) - str: 构建给 LLM 的提示词。这是 harness 的核心逻辑之一。 # 1. 系统指令定义 Agent 的角色和行为规范 prompt 你是一个有帮助的 AI 助手。你可以使用以下工具 # 2. 工具描述告诉 LLM 它能做什么 for tool in self.tools.values(): prompt f\n- {tool.name}: {tool.description} # 3. 使用格式指令严格约束 LLM 的输出格式便于解析 prompt 请严格按照以下格式回应 思考[你对当前情况的思考和分析] 行动[要调用的工具名必须是上述工具之一] 行动输入[一个 JSON 字符串包含调用该工具所需的参数] 或者如果任务已经完成直接给出最终答案 最终答案[你的回答] # 4. 历史上下文让 LLM 知道之前发生了什么 if self.state.history: prompt \n\n以下是之前的对话历史 for step in self.state.history[-3:]: # 只保留最近几步防止上下文过长 prompt f\n思考{step.get(thought)} prompt f\n行动{step.get(action)} prompt f\n观察{step.get(observation)} # 5. 当前观察新的用户输入或上一个工具的结果 prompt f\n\n当前观察{self.state.observation} prompt \n\n你的回应 return prompt def _parse_llm_response(self, response: str) - Dict[str, str]: 解析 LLM 的响应提取思考、行动或最终答案。 lines response.strip().split(\n) result {} for line in lines: if line.startswith(思考): result[thought] line[3:].strip() elif line.startswith(行动): result[action] line[3:].strip() elif line.startswith(行动输入): # 这里简化处理实际应用中需要解析 JSON result[action_input] line[5:].strip() elif line.startswith(最终答案): result[final_answer] line[5:].strip() return result设计解析__init__方法接收llm_client和tools。这表明harness不关心 LLM 的具体实现只要它提供一个生成文本的接口比如llm_client.generate(prompt)。这种依赖注入的设计保证了灵活性。_build_prompt方法是灵魂所在。它定义了 Agent 的“世界观”和“行为准则”。通过精心设计提示词我们引导 LLM 按照我们设定的 Loop 格式进行输出。这里包含了工具描述、历史上下文和当前观察构成了 LLM 推理的全部依据。_parse_llm_response是一个简单的解析器。它依赖于 LLM 严格遵守输出格式。在实际复杂场景中你可能需要使用更鲁棒的方法比如让 LLM 输出 JSON或者使用正则表达式。3.3 实现核心的 Loop 驱动引擎最后我们实现驱动整个循环的run方法。def run(self, initial_input: str) - str: 启动并运行 Agent Loop直到得到最终答案。 self.state.observation initial_input max_steps 5 # 安全限制防止无限循环 for step in range(max_steps): # 1. 思考阶段调用 LLM prompt self._build_prompt() llm_response self.llm.generate(prompt) # 假设 llm.generate 返回字符串 parsed self._parse_llm_response(llm_response) # 2. 检查是否已得出最终答案 if final_answer in parsed: print(f步骤 {step1}: 任务完成) print(f最终答案{parsed[final_answer]}) return parsed[final_answer] # 3. 行动阶段执行工具调用 thought parsed.get(thought, ) action_name parsed.get(action, ) action_input_str parsed.get(action_input, {}) print(f步骤 {step1}:) print(f 思考{thought}) print(f 行动{action_name}) print(f 输入{action_input_str}) if action_name not in self.tools: self.state.observation f错误未知工具 {action_name}。请从可用工具中选择。 # 将错误信息记录到历史以便LLM在下轮知晓 self.state.history.append({ thought: thought, action: action_name, observation: self.state.observation }) continue try: # 简化这里假设 action_input_str 是合法的 JSON 字符串 import json action_input json.loads(action_input_str) tool self.tools[action_name] # 4. 获取结果阶段执行工具 tool_result tool.run(**action_input) self.state.observation tool_result print(f 观察{tool_result}) except Exception as e: self.state.observation f工具执行出错{e} print(f 观察错误{self.state.observation}) # 5. 更新历史状态为下一轮循环准备 self.state.history.append({ thought: thought, action: action_name, observation: self.state.observation }) return 错误达到最大步数限制任务未完成。Loop 流程解析初始化将用户输入设为初始观察。构建提示并调用 LLM将当前状态历史观察构建成提示词交给 LLM“思考”。解析决策解析 LLM 的输出判断是给出了“最终答案”还是决定“行动”。执行行动如果决定行动则找到对应的工具并执行。观察结果将工具执行的结果或错误作为新的观察。状态更新将本轮循环的思考行动观察存入历史。循环判断回到第2步除非得到最终答案或超过最大步数。这个run方法就是一个最简化的Agent Executor它体现了harness作为“执行引擎”的职责。4. 实战演示用最小内核解决一个实际问题让我们用一个具体的例子让这个内核“活”起来。我们将创建一个能进行简单计算和获取当前时间的 Agent。4.1 准备工具和模拟 LLM首先我们定义两个简单的工具并模拟一个 LLM 客户端。为了绝对简化我们用一个字典来模拟 LLM 的响应。# 定义工具 def calculator(a: float, b: float, op: str) - float: if op : return a b elif op -: return a - b elif op *: return a * b elif op /: return a / b if b ! 0 else 错误除数不能为零 else: return f错误未知运算符 {op} def get_current_time() - str: from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S) tools [ Tool(namecalculator, funccalculator, description执行数学计算。输入参数: a (数字), b (数字), op (运算符, 如 , -, *, /)), Tool(nameget_time, funcget_current_time, description获取当前系统时间。无需输入参数。) ] # 模拟一个简单的 LLM 客户端 class MockLLM: def __init__(self): # 这里用一个简单的规则来模拟LLM的“思考”过程仅用于演示。 # 真实场景中这里会调用 OpenAI API 等。 self.response_rules { 计算: 思考用户需要进行数学计算。我需要使用计算器工具。\n行动calculator\n行动输入{\a\: 10, \b\: 5, \op\: \\}, 时间: 思考用户想知道现在的时间。我需要调用获取时间的工具。\n行动get_time\n行动输入{}, 最终: 思考我已经计算出了结果可以给出最终答案了。\n最终答案计算结果是 15。 } def generate(self, prompt: str) - str: # 这是一个极其简化的模拟。真实情况下prompt 会被发送到 LLM。 if 105 in prompt: return self.response_rules[计算] elif 时间 in prompt: return self.response_rules[时间] elif 观察15 in prompt: # 假设工具返回了结果15 return self.response_rules[最终] else: return 思考我无法理解这个请求。\n最终答案抱歉我无法处理这个问题。 # 初始化 harness llm_client MockLLM() my_harness Harness(llm_clientllm_client, toolstools)4.2 运行 Agent Loop现在让我们运行这个 Agent 来处理问题“10加5等于多少”print( 开始 Agent Loop 演示 ) result my_harness.run(105等于多少) print(f 演示结束最终返回结果{result} )预期的控制台输出会类似于 开始 Agent Loop 演示 步骤 1: 思考用户需要进行数学计算。我需要使用计算器工具。 行动calculator 输入{a: 10, b: 5, op: } 观察15 步骤 2: 任务完成 最终答案计算结果是 15。 演示结束最终返回结果计算结果是 15。 过程解读用户输入105等于多少成为初始观察。Harness 构建提示词包含工具列表、格式指令、当前观察交给模拟 LLM。模拟 LLM 根据规则返回“思考”和“行动”指令。Harness 解析出要调用calculator工具参数是a10, b5, op。执行calculator工具得到结果15作为新的观察。Harness 更新历史并开始下一轮循环。新的提示词中包含了上一步的“观察15”。模拟 LLM 看到上一步的结果判断任务完成返回“最终答案”。Harness 解析到最终答案结束循环并返回结果。这个演示虽然使用了模拟 LLM但完整地展示了从输入、思考、行动、观察到最终输出的整个闭环。将MockLLM替换为真实的 OpenAI 或 Anthropic API 调用它就能处理真实的、开放域的问题。5. 从最小内核到工程实践关键考量与扩展方向这个50行的内核是一个完美的起点和教学工具但要用于实际生产还有很长的路要走。以下是几个关键的扩展方向和工程化考量。5.1 提升可靠性与鲁棒性最小内核非常脆弱工程化首先要解决稳定性问题。LLM 输出解析的强化我们简单的行解析器太脆弱了。LLM 的输出可能不严格遵守格式可能有额外解释。更好的做法是使用结构化输出在提示词中要求 LLM 直接输出 JSON并指定 JSON Schema。许多现代 LLM API 原生支持此功能。使用输出解析库例如可以利用 LangChain 的OutputParser或 Pydantic 模型来定义和解析输出结构。加入重试与降级逻辑当解析失败时可以尝试让 LLM 重新生成或者回退到更简单的解析方式。工具调用的错误处理与验证参数验证在调用工具前验证action_input的参数类型、数量是否与工具定义匹配。异常捕获与友好反馈工具执行可能抛出各种异常网络超时、权限错误、业务逻辑错误。harness需要捕获这些异常并将其转化为 LLM 能理解的、结构化的错误观察信息帮助 Agent 进行下一步决策例如重试或选择其他工具。循环控制与超时防止幻觉与死循环LLM 可能会陷入“幻觉循环”反复调用不存在的工具或重复无效操作。除了设置最大步数max_steps还可以引入更复杂的策略如监控历史重复模式、设置超时时间、或在特定错误出现多次后强制终止。状态检查点对于长时间运行的任务需要能将状态序列化保存并在中断后恢复。5.2 优化提示工程与上下文管理提示词是 Agent 的“操作系统”其设计直接决定 Agent 的能力上限。动态上下文窗口管理我们的示例简单截取了最近3步历史。真实场景中LLM 有上下文长度限制。harness需要实现智能的上下文窗口管理策略关键信息摘要将过长的历史对话或工具输出总结成简短的要点保留核心信息。重要性筛选根据当前任务目标筛选出最相关的历史片段。向量检索将历史信息存入向量数据库根据当前查询动态检索最相关的上下文这是实现长期记忆的常用方法。工具描述的优化工具的描述 (description) 需要精心编写不仅要说明功能最好还能包含清晰的调用示例以及可能出错的场景。这能极大提升 LLM 选择和使用工具的准确性。思维链Chain-of-Thought的集成鼓励 LLM 在“思考”部分进行更详细的推理而不仅仅是得出结论。这不仅能提升最终行动的质量也为调试提供了宝贵线索。harness可以设计专门的提示词模板来引导这种思考。5.3 增强可观测性与调试能力当 Agent 行为不符合预期时强大的调试工具至关重要。结构化日志harness应该在每个关键节点收到输入、调用 LLM 前、解析输出后、调用工具前、收到工具结果后输出结构化的日志。这些日志应包含时间戳、步骤号、输入/输出数据、耗时等信息方便追踪整个 Loop 的流程。可视化与追踪可以集成像 LangSmith、Arize Phoenix 这样的 AI 应用监控平台或者自己实现一个简单的界面将一次 Agent 运行的完整轨迹Thought-Action-Observation 链可视化出来。这对于理解 Agent 的决策路径、定位问题步骤有巨大帮助。中间状态检查与干预在开发阶段harness可以提供“断点”功能允许开发者在每一轮循环后暂停检查当前的AgentState甚至可以手动修改状态或跳过某一步再继续运行。这是一种高效的交互式调试手段。5.4 架构扩展与生态集成随着业务复杂化最小内核需要演进为更健壮的架构。多 Agent 协作harness可以设计成能管理多个 Agent 实例并定义它们之间的通信协议例如通过共享状态或消息队列。一个 Agent 的输出可以作为另一个 Agent 的输入形成工作流。外部系统集成harness作为基础设施层需要提供清晰的接口来集成外部的知识库RAG、工作流引擎、数据库等。例如可以设计一种“插件”机制让新的工具或数据源能够方便地注册进来。配置化与策略模式将提示词模板、工具选择策略、循环终止条件等抽象为可配置的“策略”。这样无需修改核心Harness类就能通过更换策略来改变 Agent 的行为模式例如从一个保守的、步步为营的 Agent 切换为一个更激进、尝试性更强的 Agent。实现这个50行的最小内核就像亲手搭建了一个简易的发动机让你看清了每个气缸如何工作。而后续的工程化扩展则是为这台发动机加上燃油系统、冷却系统、变速箱和底盘让它能稳定、高效、安全地驱动一辆真正的汽车。理解了这个内核你再去看 LangChain Agent、AutoGen 或是其他新兴的 Agent 框架就会有一种“一览众山小”的感觉——你能清晰地分辨出哪些是核心的 Loop 逻辑哪些是它们各自实现的harness基础设施从而能够更好地选择、使用甚至定制适合自己需求的工具。