Harness 是什么——从“会用“到“会设计

第 1 课：Harness 是什么——从"会用"到"会设计"

如果你已经读完前 13 期的 LangGraph 系列，你现在应该能熟练地搭一个 Agent 了：

builder = StateGraph(MyState) builder.add_node("llm", llm_node) builder.add_node("tools", tool_node) builder.add_edge(START, "llm") builder.add_conditional_edges("llm", should_continue, {"tools": "tools", END: END}) builder.add_edge("tools", "llm") agent = builder.compile(checkpointer=saver)

但你有没有想过一个问题：每次写 Agent，你都在重复上面的代码。

换个模型要改、加个工具要改、加记忆又要改。最烦的是——你想加的特性越多，样板代码就越膨胀。

这门课的目标很直接：基于 LangGraph，封装一套你自己能用的 Harness。

第 1 课，我们先搞清楚 Harness 到底是什么、长什么样、怎么把它拆成模块。

1.1 问题：LangGraph 的"最后一公里"

LangGraph 本身已经很底层了——它给了你状态、节点、边、持久化、中断、流式。但正因为它太底层，从"有了这些积木"到"搭一个完整的 Agent 产品"之间，还有一段路。

来看看一个"比较完整"的 Agent 要多少代码：

# 这个 Agent 要支持：多轮对话 + 记忆 + 工具 + 审批 + 流式 # ─── 1. State 定义 ─── 约 10 行 class MyState(TypedDict): messages: Annotated[list, add_messages] user_id: str needs_review: bool iteration: int # ─── 2. Node 定义 ─── 约 30 行 def llm_node(state: MyState) -> dict: ... def tool_node(state: MyState) -> dict: ... def review_node(state: MyState) -> dict: ... # ─── 3. 路由函数 ─── 约 10 行 def route(state: MyState) -> str: ... # ─── 4. 构建图 ─── 约 20 行 builder = StateGraph(MyState) builder.add_node("llm", llm_node) builder.add_node("tools", tool_node) builder.add_node("review", review_node) builder.add_edge(START, "llm") builder.add_conditional_edges("llm", route, {"tools": "tools", "review": "review", END: END}) builder.add_edge("tools", "llm") builder.add_edge("review", END) # ─── 5. 编译 + 记忆 ─── 约 10 行 saver = SqliteSaver.from_conn_string("agent.db") store = InMemoryStore() graph = builder.compile(checkpointer=saver, store=store) # ─── 6. 调用 ─── 约 10 行 def chat(message: str, thread_id: str): config = {"configurable": {"thread_id": thread_id}} return graph.invoke({"messages": [("user", message)], ...}, config)

90 行代码，换一个 Agent 就要重新写一遍。大部分是重复劳动。

这就是Harness 要解决的问题——把这 90 行缩减成 3 行：

agent = MiniHarness(tools=[my_tool], enable_human_review=True) agent.run("帮我查天气")

听起来不错？我们来设计它。

1.2 Harness 的核心思想

一句话：Harness = 预置的 LangGraph 骨架 + 可配置的功能模块。

它不是一个新的框架，它是一层"胶水代码"，帮你把 LangGraph 的积木拼成一套默认好用的模式。

做一个类比：

搭 Agent 就像做饭 ┌─────────────┬───────────────┬──────────────────┐ │ │ LangGraph │ Harness │ ├─────────────┼───────────────┼──────────────────┤ │ 食材 │ State、Node、Edge │ StandardState │ │ 厨具 │ compile() │ AgentLoop │ │ 冰箱 │ Checkpointer │ MemoryManager │ │ 菜谱 │ 需要自己写 │ 预置模式 │ │ 成品 │ 每次从头做 │ 按需配置即可 │ └─────────────┴───────────────┴──────────────────┘

Harness 不替代 LangGraph——它站在 LangGraph 上面，让开发者不再重复写样板代码。

1.3 从外部看：Harness 的用户视角

先不管内部怎么实现，先看用户想怎么用它：

极简模式

# 用户只想快速搭一个能用的 Agent from mini_harness import MiniHarness agent = MiniHarness(model="gpt-4o-mini") result = agent.run("查一下北京的天气") print(result["messages"][-1].content)

带自定义工具

from mini_harness import MiniHarness from langchain.tools import tool @tool def query_database(sql: str) -> str: """执行 SQL 查询""" return str(execute(sql)) agent = MiniHarness( model="gpt-4o-mini", tools=[query_database], )

全功能模式

agent = MiniHarness( model="gpt-4o-mini", tools=[query_database, search_web], system_prompt="你是数据分析助手。", enable_planning=True, # 开启任务分解 enable_human_review=True, # 开启人工审批 memory_type="both", # 短期+长期记忆 max_iterations=20, ) result = agent.run("分析上个月的销售数据", user_id="u001")

设计目标：用户不需要了解 LangGraph 的 State、Node、Edge、Reducer、Checkpointer——他们只需要知道"加工具"和"调参"。

1.4 从内部看：Harness 的模块拆解

一个 Harness 由 5 个模块组成：

┌──────────────────────────────────────────────────────┐ │ MiniHarness │ │ │ │ ┌────────────┐ ┌───────────┐ ┌────────────────┐ │ │ │ AgentLoop │ │ Planner │ │ MemoryManager │ │ │ │ (编排引擎) │ │ (任务分解) │ │ (短期+长期记忆) │ │ │ └─────┬──────┘ └─────┬─────┘ └───────┬────────┘ │ │ │ │ │ │ │ ┌─────▼───────────────▼────────────────▼─────────┐ │ │ │ StandardState │ │ │ │ messages + plan + task_list + metadata │ │ │ └────────────────────────────────────────────────┘ │ │ │ │ ┌────────────────────────────────────────────────┐ │ │ │ BuiltinToolSet │ │ │ │ read_file / calculate / get_current_time │ │ │ └────────────────────────────────────────────────┘ │ └──────────────────────────────────────────────────────┘ │ ▼ ┌──────────────────────┐ │ LangGraph 运行时 │ │ (图构建 + 持久化) │ └──────────────────────┘

每个模块的职责：

不用一次学完 5 个模块。第 1 课（本节）只做 AgentLoop + StandardState，其他模块在后面几课逐个加上。

1.5 写代码：MiniHarness 骨架

理论讲够了，开始写代码。这是第 1 课结束时的完整产出——一个能跑的、最小可用的 Harness。

创建一个文件mini_harness_v1.py：

""" mini_harness_v1.py — 第 1 课：MiniHarness 骨架 这个版本能干什么： - 用一行代码创建 Agent - 自动绑定内置工具 + 用户自定义工具 - 支持多轮对话（Checkpointer） - 流式输出 - 自动限制最大迭代次数 不能干什么（后面几课加）： - 任务规划 - 人类审批 - 长期记忆 """ from typing import Annotated, Literal, Optional from typing_extensions import TypedDict from langgraph.graph import StateGraph, START, END, add_messages from langgraph.checkpoint.memory import InMemorySaver from langgraph.store.memory import InMemoryStore from langchain.chat_models import init_chat_model from langchain_core.messages import SystemMessage, ToolMessage from langchain.tools import tool from dataclasses import dataclass import os # ═══════════════════════════════════════════════ # 1. 配置类 # ═══════════════════════════════════════════════ @dataclass class HarnessConfig: """所有配置集中管理，一目了然""" model: str = "gpt-4o-mini" # LLM 模型 system_prompt: str = "你是一个智能助手" # 系统提示词 max_iterations: int = 10 # 最大 LLM 调用次数 temperature: float = 0 # 模型温度 # ═══════════════════════════════════════════════ # 2. 标准状态 # ═══════════════════════════════════════════════ class StandardState(TypedDict): """ 内置的标准状态，覆盖 95% 的 Agent 场景 字段说明： - messages: 对话历史，用 add_messages 实现追加 - iteration_count: 当前已执行的 LLM 调用次数 """ messages: Annotated[list, add_messages] iteration_count: int # ═══════════════════════════════════════════════ # 3. MiniHarness 主类 # ═══════════════════════════════════════════════ class MiniHarness: """ MiniHarness — 基于 LangGraph 的最小 Agent Harness 用法： agent = MiniHarness() result = agent.run("查天气") """ def __init__( self, config: Optional[HarnessConfig] = None, tools: Optional[list] = None, ): """ 参数： config: HarnessConfig 对象，或 None（使用默认配置） tools: 用户自定义工具列表 """ self.config = config or HarnessConfig() # 合并内置工具 + 用户工具 self.builtin_tools = self._create_builtin_tools() self.user_tools = tools or [] self.all_tools = self.builtin_tools + self.user_tools self.tools_by_name = {t.name: t for t in self.all_tools} # 初始化检查点和存储 self.checkpointer = InMemorySaver() # 初始化模型 self.model = init_chat_model( self.config.model, temperature=self.config.temperature, ) self.model_with_tools = self.model.bind_tools(self.all_tools) # 构建 LangGraph 图 self.graph = self._build_graph() # ── 公开方法 ── def run( self, input_text: str, thread_id: Optional[str] = None, ) -> dict: """ 运行 Agent，返回最终结果 参数： input_text: 用户输入 thread_id: 会话 ID（同一 ID 延续对话，不同 ID 新对话） 返回： State 字典，包含 messages 和 iteration_count """ import uuid tid = thread_id or str(uuid.uuid4()) config = { "configurable": {"thread_id": tid}, } return self.graph.invoke( { "messages": [("user", input_text)], "iteration_count": 0, }, config, ) def stream(self, input_text: str, thread_id: Optional[str] = None): """ 流式运行 Agent，逐 Node 返回中间状态 用法： for event in agent.stream("查天气"): print(event) """ import uuid tid = thread_id or str(uuid.uuid4()) config = { "configurable": {"thread_id": tid}, } return self.graph.stream( { "messages": [("user", input_text)], "iteration_count": 0, }, config, ) # ── 内部方法 ── def _build_graph(self) -> StateGraph: """构建 Agent 图——这是 LangGraph 的核心""" builder = StateGraph(StandardState) # 添加节点 builder.add_node("llm", self._llm_node) builder.add_node("tools", self._tool_node) # 连接：START → LLM builder.add_edge(START, "llm") # 条件边：LLM 决定下一步 builder.add_conditional_edges( "llm", self._route, {"tools": "tools", END: END}, ) # 连接：工具执行完 → 回到 LLM builder.add_edge("tools", "llm") return builder.compile(checkpointer=self.checkpointer) def _llm_node(self, state: StandardState) -> dict: """LLM 节点：用模型 + 系统提示 + 对话历史生成回复""" response = self.model_with_tools.invoke([ SystemMessage(content=self.config.system_prompt), *state["messages"], ]) return { "messages": [response], "iteration_count": state.get("iteration_count", 0) + 1, } def _tool_node(self, state: StandardState) -> dict: """工具节点：执行 LLM 请求的所有工具调用""" last_message = state["messages"][-1] results = [] for tool_call in last_message.tool_calls: tool_fn = self.tools_by_name.get(tool_call["name"]) if tool_fn is None: content = f"未知工具: {tool_call['name']}" else: try: content = str(tool_fn.invoke(tool_call["args"])) except Exception as e: content = f"工具执行失败: {type(e).__name__}: {e}" results.append(ToolMessage( content=content, tool_call_id=tool_call["id"], )) return {"messages": results} def _route(self, state: StandardState) -> Literal["tools", END]: """路由函数：有工具调用就去 tools，不然就结束""" last_message = state["messages"][-1] # 有工具调用 → 先去 tools 节点 if last_message.tool_calls: return "tools" # 超过最大迭代次数 → 强制结束 if state.get("iteration_count", 0) >= self.config.max_iterations: print(f" [达到最大迭代次数 {self.config.max_iterations}，强制结束]") return END # LLM 直接回复 → 结束 return END def _create_builtin_tools(self) -> list: """内置工具集：不用任何额外依赖""" @tool def calculate(expression: str) -> str: """执行数学计算。传入数学表达式，如 "2 + 2" 或 "sqrt(16)" """ import math allowed = set("0123456789+-*/(). sqrtpow,%") if not all(c in allowed or c.isspace() for c in expression): return "表达式包含非法字符" try: result = eval( expression, {"__builtins__": {}}, {"sqrt": math.sqrt, "pow": pow, "pi": math.pi}, ) return str(result) except Exception as e: return f"计算错误: {e}" @tool def get_time() -> str: """获取当前日期和时间""" from datetime import datetime return datetime.now().strftime("%Y-%m-%d %H:%M:%S") @tool def echo(text: str) -> str: """原样返回输入内容（调试用）""" return text return [calculate, get_time, echo] # ═══════════════════════════════════════════════ # 4. 使用示例 # ═══════════════════════════════════════════════ if __name__ == "__main__": # ── 演示 1：最小化使用 ── print("=" * 50) print("演示 1：最基本使用") print("=" * 50) agent = MiniHarness() result = agent.run("2 + 3 * 4 等于多少？") print(f"回答: {result['messages'][-1].content}") print(f"迭代次数: {result['iteration_count']}") # ── 演示 2：多轮对话 ── print("\n" + "=" * 50) print("演示 2：多轮对话（同一 thread_id）") print("=" * 50) agent = MiniHarness() session = "demo-session" r1 = agent.run("你好！我叫三丰", thread_id=session) print(f"第 1 轮: {r1['messages'][-1].content}") r2 = agent.run("我叫什么名字？", thread_id=session) print(f"第 2 轮: {r2['messages'][-1].content}") # ── 演示 3：自定义工具 ── print("\n" + "=" * 50) print("演示 3：自定义工具") print("=" * 50) from langchain.tools import tool @tool def get_weather(city: str) -> str: """查询指定城市的天气""" data = {"北京": "晴 25°C", "上海": "多云 28°C", "深圳": "雷阵雨 30°C"} return data.get(city, f"暂无 {city} 的天气数据") agent = MiniHarness(tools=[get_weather]) result = agent.run("北京天气怎么样？") print(f"回答: {result['messages'][-1].content}") # ── 演示 4：流式输出 ── print("\n" + "=" * 50) print("演示 4：流式输出") print("=" * 50) agent = MiniHarness() for event in agent.stream("现在几点了？"): for node_name, update in event.items(): if "messages" in update: msg = update["messages"][-1] if hasattr(msg, "content") and msg.content: print(f" [{node_name}] {msg.content}")

1.6 逐行解读设计决策

这段代码虽然短，但每一行都有设计上的取舍。我们把关键点过一遍。

为什么用HarnessConfig而不是散落的参数？

# ❌ 不好：参数散落在各处 agent = MiniHarness("gpt-4o-mini", "你是一个助手", 10, 0) # ✅ 好：配置集中管理 config = HarnessConfig(model="gpt-4o-mini", system_prompt="你是一个助手") agent = MiniHarness(config=config)

理由：Agent 的参数会越来越多（后面几课会加 enable_planning、enable_human_review、memory_type 等）。用 dataclass 统一管理，IDE 自动补全、文档化、序列化都方便。

为什么 StandardState 只定义了两个字段？

class StandardState(TypedDict): messages: Annotated[list, add_messages] iteration_count: int

理由：最小可行。后面几课会逐步加上 plan、task_list、context、metadata。第一课先跑通，不贪多。

为什么 combine 内置工具和用户工具？

self.all_tools = self.builtin_tools + self.user_tools

理由：用户不应该为了"查时间"这种基础操作自己去写工具。内置工具集覆盖常见需求，用户只需追加特定领域的工具。

为什么_route考虑了 max_iterations？

if state.get("iteration_count", 0) >= self.config.max_iterations: return END

理由：没有终止条件的循环是灾难。LLM 有时会陷入"调工具→发回 LLM→再调工具"的死循环。max_iterations是安全阀。

1.7 动手试试

把上面的代码保存为mini_harness_v1.py，运行一下：

# 安装依赖 pip install -U langgraph langchain langchain-openai # 设置 API Key（或等它报错后设置） export OPENAI_API_KEY="sk-..." # 运行 python mini_harness_v1.py

你会看到类似这样的输出：

================================================== 演示 1：最基本使用 ================================================== 回答: 2 + 3 * 4 = 14 迭代次数: 1 ================================================== 演示 2：多轮对话（同一 thread_id） ================================================== 第 1 轮: 你好！很高兴认识你，三丰！ 第 2 轮: 你刚才告诉我你叫三丰！ ================================================== 演示 3：自定义工具 ================================================== 回答: 北京今天天气晴朗，25°C。 ================================================== 演示 4：流式输出 ================================================== [llm] 当前时间是 2026-07-07 23:55:00。

做了几个修改试试：

1. 把max_iterations改成 1，然后问一个需要调工具的问题——看 Agent 怎么被截断
2. 不加thread_id跑两轮对话——看 Agent 是否还记得
3. 传一个不存在的模型名——看报错信息是否清晰
4. 在工具节点里加time.sleep(2)模拟慢工具——看流式输出是否依然流畅