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(checkpointersaver)但你有没有想过一个问题每次写 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(checkpointersaver, storestore) # ─── 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_reviewTrue) 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(modelgpt-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( modelgpt-4o-mini, tools[query_database], )全功能模式agent MiniHarness( modelgpt-4o-mini, tools[query_database, search_web], system_prompt你是数据分析助手。, enable_planningTrue, # 开启任务分解 enable_human_reviewTrue, # 开启人工审批 memory_typeboth, # 短期长期记忆 max_iterations20, ) result agent.run(分析上个月的销售数据, user_idu001)设计目标用户不需要了解 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 运行时 │ │ (图构建 持久化) │ └──────────────────────┘每个模块的职责模块一句话对应期数StandardStateHarness 内置的标准状态定义覆盖绝大部分场景第 2、5 期AgentLoop预置的 LLM→Tool→LLM 循环图带条件路由和终止条件第 3、4 期MemoryManagerCheckpointer Store 的封装让用户一行代码开启记忆第 6、10 期PlannerLLM 驱动的任务分解把复杂问题拆成子任务逐个执行第 11 期BuiltinToolSet开箱即用的工具集用户不用自己写就能用第 4 期不用一次学完 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, temperatureself.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(checkpointerself.checkpointer) def _llm_node(self, state: StandardState) - dict: LLM 节点用模型 系统提示 对话历史生成回复 response self.model_with_tools.invoke([ SystemMessage(contentself.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( contentcontent, tool_call_idtool_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_idsession) print(f第 1 轮: {r1[messages][-1].content}) r2 agent.run(我叫什么名字, thread_idsession) 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(modelgpt-4o-mini, system_prompt你是一个助手) agent MiniHarness(configconfig)理由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_iterationsif 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_KEYsk-... # 运行 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。做了几个修改试试把max_iterations改成 1然后问一个需要调工具的问题——看 Agent 怎么被截断不加thread_id跑两轮对话——看 Agent 是否还记得传一个不存在的模型名——看报错信息是否清晰在工具节点里加time.sleep(2)模拟慢工具——看流式输出是否依然流畅