GEOZ

Agent Harness 到底是什么?拆解 LLM 从「能跑」到「可靠」的基础设施层

2026/9/22
Agent Harness 到底是什么?拆解 LLM 从「能跑」到「可靠」的基础设施层

AIAI Summary (BLUF)

本文系统讲解 Agent Harness 的概念与核心架构,涵盖会话管理、工具注册、规则引擎、上下文注入和可观测性五大组件,并深入解析 Skill 系统设计。最后使用智谱 GLM-5-Turbo 模型(OpenAI 兼容接口)手把手搭建完整 Agent,包含 6 个渐进式示例,代码可直接运行。

核心洞察

这篇文章最有意思的地方在于它把 Agent Harness 这个听起来很玄的概念拆得特别实在。作者用"马具"来比喻 Harness,一下子就把 LLM 和基础设施的关系讲清楚了。不过我要提醒一句:文中给的架构图虽然完整,但实际项目里你大概率不需要一上来就搭这么全,先跑通 Agent Loop 再说。


核心结论

  1. Agent Harness 是包裹 LLM 的基础设施层,核心作用是让 Agent 从"能跑"变成"可靠",主要解决调试困难、行为不可控、无法评估、难以复用、多模型切换五大问题。

  2. Harness、Framework、Runtime 三者职责不同:Framework(如 LangChain、CrewAI)负责定义 Agent 结构与行为,Runtime(如 Vercel AI SDK)提供运行环境,Harness 负责管理、控制和评估 Agent。

  3. Agent 技术经历了六个发展阶段:Prompt Engineering(2022)→ Chain of Thought(2023)→ Tool Use/Function Calling(2023)→ ReAct/Agent Loop(2023)→ Multi-Agent(2024)→ Agent Harness(2024-2025)。

  4. Agent Loop 的核心是 ReAct 循环:思考(Think)→ 行动(Act)→ 观察(Observe)→ 判断(Decide),未完成则回到思考,完成则返回结果,通常设置最大迭代次数(如 5 次)防止死循环。

  5. Harness 架构包含五大核心组件:会话管理(维护对话历史与上下文窗口)、工具注册中心(工具定义与 schema)、规则与安全层、LLM 交互层(模型适配器)、可观测性层(轨迹记录、指标收集、评估系统),其中可观测性层贯穿所有层。

一、什么是 Agent Harness?

1.1 从 LLM 到 Agent

普通的 LLM 调用长这样:

用户输入 → LLM → 文本回复

这就是个问答机器。要让它变成真正的 Agent,得给它配几样东西:

  • 记住对话历史(会话管理)
  • 调用外部工具(工具调用)
  • 遵守规则约束(安全策略)
  • 追踪执行过程(可观测性)

这些基础设施加在一起,就是 Agent Harness。

1.2 Harness 的比喻

Harness = 马具

马(LLM)本身有力量,但需要马具来:
- 控制方向(规则引擎)
- 连接马车(工具系统)
- 记录行程(追踪系统)
- 管理骑手(会话管理)

1.3 为什么需要 Harness?

没有 Harness 有 Harness
每次对话都是独立的 记住上下文和历史
只能输出文本 可以调用工具执行操作
无法控制行为 有规则和约束
出错无法追溯 完整轨迹记录
无法测试评估 自动化评估框架

1.4 Harness vs Framework vs Runtime

这三个概念经常被搞混,理清它们的区别很重要:

层级 职责 解决的核心问题 举例
Framework 定义 Agent 结构和行为 如何组织 Agent 的逻辑 LangChain, CrewAI
Runtime 提供运行环境 如何执行 Agent 代码 Vercel AI SDK
Harness 包裹 Agent 的基础设施 如何管理、控制、评估 Agent 会话管理、工具注册
LLM 推理和生成 如何理解和生成内容 GLM-5-Turbo, GPT-4

一句话:Framework 是设计图纸,Runtime 是施工平台,Harness 是建成后的物业管理系统。


二、Agent 技术演进

2.1 六个发展阶段

阶段 1: Prompt Engineering (2022)
  └── 精心设计的提示词让 LLM 完成特定任务
  └── 局限:无记忆、无工具、单次交互

阶段 2: Chain of Thought (2023)
  └── 让 LLM 逐步推理,提高复杂任务能力
  └── 局限:仍然无外部交互能力

阶段 3: Tool Use / Function Calling (2023)
  └── LLM 可以调用外部工具
  └── 突破:从"说话"到"做事"

阶段 4: ReAct / Agent Loop (2023)
  └── 思考 → 行动 → 观察 → 思考 的循环
  └── 突破:自主决策和多步执行

阶段 5: Multi-Agent (2024)
  └── 多个 Agent 协作完成复杂任务
  └── 突破:分工协作、角色 specialization

阶段 6: Agent Harness (2024-2025)
  └── 完整的 Agent 基础设施
  └── 突破:标准化、可评估、可观测、可复用

2.2 Agent Loop(智能体循环)

Agent 的核心是一个循环过程,叫 ReAct Loop:

┌─────────┐
│  思考    │ ← 分析当前状态,决定下一步行动
│ Think   │
└────┬────┘
     ↓
┌─────────┐
│  行动    │ ← 调用工具或生成回复
│  Act    │
└────┬────┘
     ↓
┌─────────┐
│  观察    │ ← 获取工具执行结果或用户反馈
│ Observe │
└────┬────┘
     ↓
┌─────────┐
│  判断    │ ← 任务是否完成?
│ Decide  │
└────┬────┘
     ↓
┌──────────┬──────────┐
│  未完成   │   完成    │
│  回到思考  │  返回结果  │
└──────────┴──────────┘

代码视角的 Agent Loop

def agent_loop(user_input, max_iterations=5):
    messages = [{"role": "user", "content": user_input}]
    
    for i in range(max_iterations):
        # 1. 思考:调用 LLM 决定下一步
        response = llm(messages, tools=available_tools)
        
        # 2. 判断:LLM 返回了什么?
        if response.tool_calls:
            # 3. 行动:执行工具
            for tool_call in response.tool_calls:
                result = execute_tool(tool_call)
                # 4. 观察:将结果加入对话
                messages.append({"role": "tool", "content": result})
        else:
            # 任务完成,返回最终回复
            return response.content
    
    return "达到最大迭代次数"

2.3 为什么 Harness 是必然趋势?

Agent 越来越复杂之后,下面这些问题就藏不住了:

问题 没有 Harness 有 Harness
调试困难 Agent 出错不知原因 完整轨迹记录,可回放
行为不可控 LLM 可能做危险操作 规则引擎拦截
无法评估 不知道 Agent 好不好 自动化测试和评分
难以复用 每个项目重新造轮子 标准化组件可插拔
多模型切换 代码耦合特定 SDK 统一接口,热切换

Harness 要解决的核心问题是:让 Agent 从"能跑"变成"可靠"。


三、核心架构解析

3.1 完整架构图

┌─────────────────────────────────────────────────────────────────────────────┐
│                              Agent Harness                                   │
│                                                                             │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                        输入处理层 (Input Layer)                      │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────────┐  │   │
│  │  │  用户输入     │  │  事件触发     │  │  外部系统回调            │  │   │
│  │  └──────┬───────┘  └──────┬───────┘  └───────────┬──────────────┘  │   │
│  │         └─────────────────┼──────────────────────┘                 │   │
│  │                           ↓                                        │   │
│  │              ┌────────────────────────┐                            │   │
│  │              │    输入验证 & 规范化    │                            │   │
│  │              └────────────┬───────────┘                            │   │
│  └───────────────────────────┼───────────────────────────────────────┘   │
│                              ↓                                            │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      上下文管理层 (Context Layer)                    │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────────┐  │   │
│  │  │  会话管理     │  │  记忆系统     │  │  RAG 检索增强            │  │   │
│  │  └──────┬───────┘  └──────┬───────┘  └───────────┬──────────────┘  │   │
│  │         └─────────────────┼──────────────────────┘                 │   │
│  │                           ↓                                        │   │
│  │              ┌────────────────────────┐                            │   │
│  │              │    上下文组装 & 裁剪    │                            │   │
│  │              └────────────┬───────────┘                            │   │
│  └───────────────────────────┼───────────────────────────────────────┘   │
│                              ↓                                            │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      工具与技能层 (Tools & Skills)                   │   │
│  │  ┌─────────────────────────────────────────────────────────────┐   │   │
│  │  │                    工具注册中心 (Tool Registry)               │   │   │
│  │  │  ┌─────────────┐  ┌─────────────┐  ┌────────────────────┐  │   │   │
│  │  │  │  内置工具    │  │  Skills     │  │  自定义工具         │  │   │   │
│  │  │  └─────────────┘  └─────────────┘  └────────────────────┘  │   │   │
│  │  └─────────────────────────────────────────────────────────────┘   │   │
│  └──────────────────────────────────────┬──────────────────────────────┘   │
│                                         ↓                                  │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      规则与安全层 (Rules & Security)                 │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────────┐  │   │
│  │  │  输入规则     │  │  输出规则     │  │  工具执行规则            │  │   │
│  │  └──────────────┘  └──────────────┘  └──────────────────────────┘  │   │
│  └──────────────────────────────────────┬──────────────────────────────┘   │
│                                         ↓                                  │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      LLM 交互层 (Model Interface)                    │   │
│  │  ┌─────────────────────────────────────────────────────────────┐   │   │
│  │  │                    Prompt 构建器                              │   │   │
│  │  │  System Prompt + Context + Tools Schema + History + Input   │   │   │
│  │  └─────────────────────────────────────────────────────────────┘   │   │
│  │                             ↓                                       │   │
│  │  ┌─────────────────────────────────────────────────────────────┐   │   │
│  │  │                    模型适配器 (Model Adapter)                 │   │   │
│  │  │  ┌─────────────┐  ┌─────────────┐  ┌────────────────────┐  │   │   │
│  │  │  │  OpenAI     │  │  Anthropic  │  │  其他 (Gemini/...) │  │   │   │
│  │  │  └─────────────┘  └─────────────┘  └────────────────────┘  │   │   │
│  │  └─────────────────────────────────────────────────────────────┘   │   │
│  └──────────────────────────────────────┬──────────────────────────────┘   │
│                                         ↓                                  │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      响应处理层 (Response Layer)                     │   │
│  │                    ┌────────────────────┐                           │   │
│  │                    │    响应解析         │                           │   │
│  │                    └────────┬───────────┘                           │   │
│  │              ┌──────────────┴──────────────┐                        │   │
│  │              ↓                             ↓                        │   │
│  │     ┌────────────────┐          ┌────────────────────┐              │   │
│  │     │  文本回复       │          │  工具调用请求       │              │   │
│  │     │ → 规则检查     │          │ → 权限验证          │              │   │
│  │     │ → 返回用户     │          │ → 执行工具 → 循环   │              │   │
│  │     └────────────────┘          └────────────────────┘              │   │
│  └─────────────────────────────────────────────────────────────────────┘   │
│                                                                             │
│  ┌─────────────────────────────────────────────────────────────────────┐   │
│  │                      可观测性层 (Observability)  ← 贯穿所有层        │   │
│  │  ┌──────────────┐  ┌──────────────┐  ┌──────────────────────────┐  │   │
│  │  │  轨迹记录     │  │  指标收集     │  │  评估系统                │  │   │
│  │  └──────────────┘  └──────────────┘  └──────────────────────────┘  │   │
│  └─────────────────────────────────────────────────────────────────────┘   │
└─────────────────────────────────────────────────────────────────────────────┘

3.2 五大核心组件

组件一:会话管理 (Session Manager)

职责:
├── 维护对话历史 (message history)
├── 管理上下文窗口 (context window)
├── 会话状态持久化
└── 多轮对话的状态跟踪

关键设计:
├── 消息裁剪策略(超出 token 限制时如何处理)
├── 系统提示词管理 (system prompt)
├── 会话隔离(多用户场景)
└── 上下文压缩(摘要、向量检索)

消息裁剪策略对比

策略优点缺点适用场景
保留最近 N 条简单高效可能丢失重要上下文短对话
摘要压缩保留语义信息需要额外调用 LLM长对话
重要性评分保留关键信息实现复杂复杂任务
向量检索按需检索相关历史需要向量数据库超长对话
组件二:工具注册中心 (Tool Registry)
职责:
├── 工具定义(名称、描述、参数 schema)
├── 工具注册/注销
├── 工具调用路由
└── 工具执行结果格式化

关键设计:
├── 统一的工具接口
├── 动态工具加载
├── 工具权限控制
└── 跨 SDK 兼容(OpenAI / Anthropic)

工具调用格式(OpenAI 标准)

{
  "type": "function",
  "function": {
    "name": "calculate",
    "description": "计算数学表达式",
    "parameters": {
      "type": "object",
      "properties": {
        "expression": {
          "type": "string",
          "description": "要计算的表达式"
        }
      },
      "required": ["expression"]
    }
  }
}
组件三:规则引擎 (Rules Engine)
职责:
├── 输入规则(内容过滤、注入防御)
├── 输出规则(格式校验、敏感信息过滤)
├── 工具规则(权限检查、沙箱隔离)
└── 降级策略(LLM 出错时怎么办)
组件四:上下文注入 (Context Injector)
职责:
├── 动态注入相关知识
├── RAG 检索增强
├── 环境变量注入
└── 用户偏好/记忆
组件五:可观测性 (Observability)
职责:
├── 记录 Agent 决策轨迹
├── 工具调用日志
├── 性能指标收集(延迟、Token 用量、成本)
└── 自动化测试用例

四、Skill 系统详解

4.1 什么是 Skill?

Skill(技能) 是 Harness 中 可插拔的能力模块,它将特定领域的知识、工具、规则和工作流封装在一起。

Skill = 领域知识 + 工具定义 + 行为规则 + 工作流

4.2 Skill 在 Harness 中的位置

┌─────────────────────────────────────────────────────┐
│                    Agent Harness                      │
│                                                     │
│  ┌───────────────────────────────────────────────┐  │
│  │          工具注册中心 (Tool Registry)           │  │
│  │  ┌───────────┬──────────────┬───────────────┐  │  │
│  │  │  内置工具  │  Skills 技能  │  自定义工具    │  │  │
│  │  │ • Bash    │ • ssh-sync   │ • 业务 API    │  │  │
│  │  │ • Read    │ • tavily     │ • 数据库查询   │  │  │
│  │  │ • Edit    │ • web-crawl  │ • 第三方服务   │  │  │
│  │  └───────────┴──────────────┴───────────────┘  │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│  ┌───────────────────────────────────────────────┐  │
│  │          规则引擎 (Rules Engine)                │  │
│  │  → Skills 自带行为约束和执行规则                │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│  ┌───────────────────────────────────────────────┐  │
│  │          上下文注入 (Context Injector)          │  │
│  │  → Skills 提供领域知识 (SKILL.md)               │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│                        ↕                            │
│                  LLM (模型)                          │
└─────────────────────────────────────────────────────┘

4.3 Skill 的目录结构

skills/
└── ssh-dev-sync/
    ├── SKILL.md              # 核心文件:触发条件 + 工作流 + 领域知识
    ├── scripts/              # 可执行脚本
    │   └── sync.sh
    ├── reference/            # 参考资料
    │   └── api-docs.md
    └── templates/            # 模板文件
        └── config.yaml

4.4 SKILL.md 的结构示例

---
name: ssh-dev-sync
description: 远程服务器开发工作流,支持代码同步、命令执行、日志获取
trigger: |
  当用户想要:
  - 同步代码到远程服务器
  - 在远程服务器执行命令
  - 获取远程服务器日志
---

SSH Dev Sync

工作流

  1. 连接到远程服务器
  2. 同步代码(git push + git pull)
  3. 执行命令
  4. 获取输出
组件二:工具注册中心 (Tool Registry)
职责:
├── 工具定义(名称、描述、参数 schema)
├── 工具注册/注销
├── 工具调用路由
└── 工具执行结果格式化

关键设计:
├── 统一的工具接口
├── 动态工具加载
├── 工具权限控制
└── 跨 SDK 兼容(OpenAI / Anthropic)

工具调用格式(OpenAI 标准)

{
  "type": "function",
  "function": {
    "name": "calculate",
    "description": "计算数学表达式",
    "parameters": {
      "type": "object",
      "properties": {
        "expression": {
          "type": "string",
          "description": "要计算的表达式"
        }
      },
      "required": ["expression"]
    }
  }
}
组件三:规则引擎 (Rules Engine)
职责:
├── 输入规则(内容过滤、注入防御)
├── 输出规则(格式校验、敏感信息过滤)
├── 工具规则(权限检查、沙箱隔离)
└── 降级策略(LLM 出错时怎么办)
组件四:上下文注入 (Context Injector)
职责:
├── 动态注入相关知识
├── RAG 检索增强
├── 环境变量注入
└── 用户偏好/记忆
组件五:可观测性 (Observability)
职责:
├── 记录 Agent 决策轨迹
├── 工具调用日志
├── 性能指标收集(延迟、Token 用量、成本)
└── 自动化测试用例

四、Skill 系统详解

4.1 什么是 Skill?

Skill(技能) 是 Harness 中 可插拔的能力模块,它将特定领域的知识、工具、规则和工作流封装在一起。

Skill = 领域知识 + 工具定义 + 行为规则 + 工作流

4.2 Skill 在 Harness 中的位置

┌─────────────────────────────────────────────────────┐
│                    Agent Harness                      │
│                                                     │
│  ┌───────────────────────────────────────────────┐  │
│  │          工具注册中心 (Tool Registry)           │  │
│  │  ┌───────────┬──────────────┬───────────────┐  │  │
│  │  │  内置工具  │  Skills 技能  │  自定义工具    │  │  │
│  │  │ • Bash    │ • ssh-sync   │ • 业务 API    │  │  │
│  │  │ • Read    │ • tavily     │ • 数据库查询   │  │  │
│  │  │ • Edit    │ • web-crawl  │ • 第三方服务   │  │  │
│  │  └───────────┴──────────────┴───────────────┘  │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│  ┌───────────────────────────────────────────────┐  │
│  │          规则引擎 (Rules Engine)                │  │
│  │  → Skills 自带行为约束和执行规则                │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│  ┌───────────────────────────────────────────────┐  │
│  │          上下文注入 (Context Injector)          │  │
│  │  → Skills 提供领域知识 (SKILL.md)               │  │
│  └───────────────────────────────────────────────┘  │
│                                                     │
│                        ↕                            │
│                  LLM (模型)                          │
└─────────────────────────────────────────────────────┘

4.3 Skill 的目录结构

skills/
└── ssh-dev-sync/
    ├── SKILL.md              # 核心文件:触发条件 + 工作流 + 领域知识
    ├── scripts/              # 可执行脚本
    │   └── sync.sh
    ├── reference/            # 参考资料
    │   └── api-docs.md
    └── templates/            # 模板文件
        └── config.yaml

4.4 SKILL.md 的结构示例

---
name: ssh-dev-sync
description: 远程服务器开发工作流,支持代码同步、命令执行、日志获取
trigger: |
  当用户想要:
  - 同步代码到远程服务器
  - 在远程服务器执行命令
  - 获取远程服务器日志
---

SSH Dev Sync

工作流

  1. 连接到远程服务器
  2. 同步代码(git push + git pull)
  3. 执行命令
  4. 获取输出

注意事项

  • 不要在生产环境执行危险命令
  • 同步前确认远程分支状态

4.5 Skill 的生命周期

1. 发现 (Discover) → 用户表达需求 → 匹配 Skill trigger
2. 加载 (Load)     → 读取 SKILL.md → 注入上下文
3. 注册 (Register) → 注册工具到 Tool Registry,注册规则到 Rules Engine
4. 执行 (Execute)  → 按照工作流执行
5. 卸载 (Unload)   → 清理资源,释放内存

4.6 Skill 的优势

特性没有 Skill有 Skill
知识复用每次重新编写提示词SKILL.md 可复用
按需加载所有工具始终可用匹配触发才加载
领域专精通用能力特定领域深度优化
社区共享闭源实现标准化格式可分享
热插拔硬编码在代码中目录即插即用

五、环境准备

5.1 安装依赖

pip install openai pyyaml

5.2 配置智谱 API(OpenAI 兼容接口)

智谱提供 OpenAI 兼容接口,只需更换 api_keybase_url

from openai import OpenAI

client = OpenAI(
api_key="your-api-key",
base_url="https://open.bigmodel.cn/api/paas/v4/"
)

提示:使用 OpenAI 库的好处是生态统一,后续换其他模型(Ernie、GPT 等)只需改 base_url。如果智谱接口有差异,也可以用 LiteLLM 转一层。


六、实战:从零搭建 Harness

6.1 第一步:基础 LLM 调用

这是最简单的调用方式,没有 Harness 的任何组件:

from openai import OpenAI

client = OpenAI(
api_key="your-api-key",
base_url="https://open.bigmodel.cn/api/paas/v4/"
)

response = client.chat.completions.create(
model="glm-5-turbo",
messages=[
{"role": "user", "content": "你好,请介绍一下自己"}
],
max_tokens=1024,
temperature=0.7
)

print(response.choices[0].message.content)

输入:

用户: "你好,请介绍一下自己"

输出:

你好!我是 GLM-5-Turbo,是由智谱 AI 开发的大语言模型。
我可以帮助你回答问题、创作文字、进行逻辑推理、编程等任务。

说明: 每次调用都是独立的,没有记忆。


6.2 第二步:添加会话管理

让模型能够记住上下文:

from openai import OpenAI

class SimpleSession:
"""最简会话管理器"""

<span>def</span> <span>__init__</span><span>(</span>self<span>,</span> system_prompt<span>=</span><span>None</span><span>)</span><span>:</span>
    self<span>.</span>messages <span>=</span> <span>[</span><span>]</span>
    <span>if</span> system_prompt<span>:</span>
        self<span>.</span>messages<span>.</span>append<span>(</span><span>{<!-- --></span><span>"role"</span><span>:</span> <span>"system"</span><span>,</span> <span>"content"</span><span>:</span> system_prompt<span>}</span><span>)</span>

<span>def</span> <span>add_user_message</span><span>(</span>self<span>,</span> content<span>)</span><span>:</span>
    self<span>.</span>messages<span>.</span>append<span>(</span><span>{<!-- --></span><span>"role"</span><span>:</span> <span>"user"</span><span>,</span> <span>"content"</span><span>:</span> content<span>}</span><span>)</span>

<span>def</span> <span>add_assistant_message</span><span>(</span>self<span>,</span> content<span>)</span><span>:</span>
    self<span>.</span>messages<span>.</span>append<span>(</span><span>{<!-- --></span><span>"role"</span><span>:</span> <span>"assistant"</span><span>,</span> <span>"content"</span><span>:</span> content<span>}</span><span>)</span>

<span>def</span> <span>get_messages</span><span>(</span>self<span>)</span><span>:</span>
    <span>return</span> self<span>.</span>messages

class MinimalHarness:
"""最小可用 Harness"""

<span>def</span> <span>__init__</span><span>(</span>self<span>,</span> api_key<span>,</span> system_prompt<span>=</span><span>None</span><span>)</span><span>:</span>
    self<span>.</span>client <span>=</span> OpenAI<span>(</span>
        api_key<span>=</span>api_key<span>,</span>
        base_url<span>=</span><span>"https://open.bigmodel.cn/api/paas/v4/"</span>
    <span>)</span>
    self<span>.</span>session <span>=</span> SimpleSession<span>(</span>system_prompt<span>)</span>

<span>def</span> <span>chat</span><span>(</span>self<span>,</span> user_input<span>)</span><span>:</span>
    <span># 1. 添加用户输入到会话</span>
    self<span>.</span>session<span>.</span>add_user_message<span>(</span>user_input<span>)</span>
    
    <span># 2. 调用 LLM</span>
    response <span>=</span> self<span>.</span>client<span>.</span>chat<span>.</span>completions<span>.</span>create<span>(</span>
        model<span>=</span><span>"glm-5-turbo"</span><span>,</span>
        messages<span>=</span>self<span>.</span>session<span>.</span>get_messages<span>(</span><span>)</span><span>,</span>
        max_tokens<span>=</span><span>1024</span><span>,</span>
        temperature<span>=</span><span>0.7</span>
    <span>)</span>
    
    <span># 3. 获取回复并保存到会话</span>
    reply <span>=</span> response<span>.</span>choices<span>[</span><span>0</span><span>]</span><span>.</span>message<span>.</span>content
    self<span>.</span>session<span>.</span>add_assistant_message<span>(</span>reply<span>)</span>
    
    <span>return</span> reply

# === 使用示例 ===
harness = MinimalHarness(
api_key="your-api-key",
system_prompt="你是一个友好的助手,擅长解答各种问题。"
)

# 第一轮对话
reply1 = harness.chat("你好,我想学习 Python")
print(f"AI: {reply1}")

# 第二轮对话(模型会记住上下文)
reply2 = harness.chat("有什么好的学习建议吗?")
print(f"AI: {reply2}")

输入:

第一轮 - 用户: "你好,我想学习 Python"
第二轮 - 用户: "有什么好的学习建议吗?"

输出:

AI: 你好!学习 Python 是个很好的选择...
AI: 关于学习 Python,我有以下建议:
    1. 从基础语法开始...
    2. 多做练习...
    3. 阅读优秀的项目代码...

说明: 第二轮对话中,模型知道"学习建议"指的是学习 Python 的建议,因为它记住了第一轮的对话内容。

会话历史实际内容:

[
  {"role": "system", "content": "你是一个友好的助手..."},
  {"role": "user", "content": "你好,我想学习 Python"},
  {"role": "assistant", "content": "你好!学习 Python 是个很好的选择..."},
  {"role": "user", "content": "有什么好的学习建议吗?"},
  {"role": "assistant", "content": "关于学习 Python,我有以下建议..."}
]

6.3 第三步:工具注册中心

让 LLM 能调用外部工具:

import json
from typing import Callable, Dict, Any
from datetime import datetime

class ToolRegistry:
"""工具注册中心"""

<span>def</span> <span>__init__</span><span>(</span>self<span>)</span><span>:</span>
    self<span>.</span>tools<span>:</span> Dict<span>[</span><span>str</span><span>,</span> Dict<span>[</span><span>str</span><span>,</span> Any<span>]</span><span>]</span> <span>=</span> <span>{<!-- --></span><span>}</span>

<span>def</span> <span>register</span><span>(</span>self<span>,</span> name<span>:</span> <span>str</span><span>,</span> description<span>:</span> <span>str</span><span>,</span> 
             parameters<span>:</span> <span>dict</span><span>,</span> func<span>:</span> Callable<span>)</span><span>:</span>
    <span>"""注册一个工具"""</span>
    self<span>.</span>tools<span>[</span>name<span>]</span> <span>=</span> <span>{<!-- --></span>
        <span>"name"</span><span>:</span> name<span>,</span>
        <span>"description"</span><span>:</span> description<span>,</span>
        <span>"parameters"</span><span>:</span> parameters<span>,</span>
        <span>"function"</span><span>:</span> func
    <span>}</span>

<span>def</span> <span>get_all_schemas</span><span>(</span>self<span>)</span> <span>-</span><span>&gt;</span> <span>list</span><span>:</span>
    <span>"""获取所有工具的 Schema(传给 LLM)"""</span>
    <span>return</span> <span>[</span><span>{<!-- --></span>
        <span>"type"</span><span>:</span> <span>"function"</span><span>,</span>
        <span>"function"</span><span>:</span> <span>{<!-- --></span>
            <span>"name"</span><span>:</span> t<span>[</span><span>"name"</span><span>]</span><span>,</span>
            <span>"description"</span><span>:</span> t<span>[</span><span>"description"</span><span>]</span><span>,</span>
            <span>"parameters"</span><span>:</span> t<span>[</span><span>"parameters"</span><span>]</span>
        <span>}</span>
    <span>}</span> <span>for</span> t <span>in</span> self<span>.</span>tools<span>.</span>values<span>(</span><span>)</span><span>]</span>

<span>def</span> <span>execute</span><span>(</span>self<span>,</span> name<span>:</span> <span>str</span><span>,</span> arguments<span>:</span> <span>dict</span><span>)</span> <span>-</span><span>&gt;</span> Any<span>:</span>
    <span>"""执行工具"""</span>
    <span>return</span> self<span>.</span>tools<span>[</span>name<span>]</span><span>[</span><span>"function"</span><span>]</span><span>(</span><span>**</span>arguments<span>)</span>

# === 定义工具 ===
def get_current_time(timezone: str = "UTC") -> str:
return datetime.now().strftime("%Y-%m-%d %H:%M:%S")

def calculate(expression: str) -> str:
try:
result = eval(expression, {"builtins": {}}, {})
return str(result)
except Exception as e:
return f"计算错误: {str(e)}"

# === 注册工具 ===
registry = ToolRegistry()

registry.register(
name="get_current_time",
description="获取当前日期和时间",
parameters={
"type": "object",
"properties": {
"timezone": {"type": "string", "description": "时区"}
}
},
func=get_current_time
)

registry.register(
name="calculate",
description="计算数学表达式",
parameters={
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
},
func=calculate
)

print("已注册工具:", list(registry.tools.keys()))


6.4 第四步:工具调用流程

完整的工具调用 Harness:

import json
from openai import OpenAI

class ToolCallingHarness:
    """支持工具调用的 Harness"""
    
    def __init__(self, api_key, registry, system_prompt=None):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://open.bigmodel.cn/api/paas/v4/"
        )
        self.registry = registry
        self.messages = []
        if system_prompt:
            self.messages.append({"role": "system", "content": system_prompt})
    
    def chat(self, user_input, max_iterations=3):
        self.messages.append({"role": "user", "content": user_input})
        
        for i in range(max_iterations):
            response = self.client.chat.completions.create(
                model="glm-5-turbo",
                messages=self.messages,
                tools=self.registry.get_all_schemas(),
                max_tokens=1024,
                temperature=0.7
            )
            
            message = response.choices[0].message
            
            if message.tool_calls:
                # 有工具调用
                self.messages.append(message)
                
                for tool_call in message.tool_calls:
                    tool_name = tool_call.function.name
                    arguments = json.loads(tool_call.function.arguments)
                    
                    print(f"调用工具: {tool_name}")
                    print(f"   参数: {arguments}")
                    
                    result = self.registry.execute(tool_name, arguments)
                    print(f"   结果: {result}")
                    
                    self.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })
            else:
                # 没有工具调用,返回文本回复
                reply = message.content
                self.messages.append(message)
                return reply
        
        return "达到最大迭代次数"

# === 使用示例 ===
harness = ToolCallingHarness(
    api_key="your-api-key",
    registry=registry,
    system_prompt="你是一个有用的助手。当需要获取时间或计算时,请使用相应的工具。"
)

reply = harness.chat("请帮我计算 123 * 456 + 789")
print(f"AI: {reply}")

执行过程:

调用工具: calculate
   参数: {"expression": "123 * 456 + 789"}
   结果: 56877

输出:

AI: 计算结果:123 * 456 + 789 = 56877

消息历史变化:

[用户] "请帮我计算 123 * 456 + 789"
     ↓
[AI] tool_call: calculate({"expression": "123 * 456 + 789"})
     ↓
[工具] 56877
     ↓
[AI] "计算结果:123 * 456 + 789 = 56877"

6.5 第五步:完整的 Agent Harness

把前面几个组件拼起来:工具注册、规则引擎、可观测性。

import json
from datetime import datetime
from typing import Callable, Dict, Any, List
from openai import OpenAI

# ==========================================
# 组件 1:规则引擎
# ==========================================
class RulesEngine:
    def __init__(self):
        self.input_rules = []
        self.output_rules = []
        self.tool_rules = []
    
    def add_input_rule(self, rule):
        self.input_rules.append(rule)
    
    def add_output_rule(self, rule):
        self.output_rules.append(rule)
    
    def add_tool_rule(self, rule):
        self.tool_rules.append(rule)
    
    def check_input(self, user_input: str) -> tuple:
        for rule in self.input_rules:
            ok, msg = rule(user_input)
            if not ok:
                return False, msg
        return True, ""
    
    def check_output(self, output: str) -> tuple:
        for rule in self.output_rules:
            ok, msg = rule(output)
            if not ok:
                return False, msg
        return True, ""
    
    def check_tool_call(self, tool_name: str, arguments: dict) -> tuple:
        for rule in self.tool_rules:
            ok, msg = rule(tool_name, arguments)
            if not ok:
                return False, msg
        return True, ""

# ==========================================
# 组件 2:可观测性(日志记录)
# ==========================================
class Logger:
    def __init__(self):
        self.logs: List[dict] = []
    
    def log(self, event_type: str, data: dict):
        entry = {
            "timestamp": datetime.now().isoformat(),
            "type": event_type,
            "data": data
        }
        self.logs.append(entry)
        print(f"[{event_type}] {json.dumps(data, ensure_ascii=False)[:100]}...")
# ==========================================
# 组件 3:完整的 Agent Harness
# ==========================================
class AgentHarness:
    def __init__(self, api_key, system_prompt, registry, rules, logger):
        self.client = OpenAI(
            api_key=api_key,
            base_url="https://open.bigmodel.cn/api/paas/v4/"
        )
        self.registry = registry
        self.rules = rules
        self.logger = logger
        self.messages = [{"role": "system", "content": system_prompt}]
    
    def chat(self, user_input: str, max_iterations: int = 5) -> str:
        self.logger.log("user_input", {"content": user_input})
        
        # 规则检查:输入
        ok, msg = self.rules.check_input(user_input)
        if not ok:
            self.logger.log("rule_violation", {"rule": "input", "message": msg})
            return f"输入被拒绝:{msg}"
        
        self.messages.append({"role": "user", "content": user_input})
        
        for i in range(max_iterations):
            self.logger.log("llm_call", {"iteration": i + 1})
            
            response = self.client.chat.completions.create(
                model="glm-5-turbo",
                messages=self.messages,
                tools=self.registry.get_all_schemas() if self.registry.tools else None,
                max_tokens=2048,
                temperature=0.7
            )
            
            message = response.choices[0].message
            
            if message.tool_calls:
                self.messages.append(message)
                
                for tool_call in message.tool_calls:
                    tool_name = tool_call.function.name
                    arguments = json.loads(tool_call.function.arguments)
                    
                    # 规则检查:工具调用
                    ok, msg = self.rules.check_tool_call(tool_name, arguments)
                    if not ok:
                        result = f"工具调用被拒绝:{msg}"
                    else:
                        self.logger.log("tool_call", {"tool": tool_name})
                        try:
                            result = self.registry.execute(tool_name, arguments)
                        except Exception as e:
                            result = f"工具执行错误: {str(e)}"
                    
                    self.messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": str(result)
                    })
            else:
                reply = message.content
                self.messages.append(message)
                
                # 规则检查:输出
                ok, msg = self.rules.check_output(reply)
                if not ok:
                    return "输出被规则拦截"
                
                self.logger.log("assistant_reply", {"content": reply[:50]})
                return reply
        
        return "达到最大迭代次数"

# ==========================================
# 使用示例
# ==========================================
if __name__ == "__main__":
    # 1. 注册工具(使用前面定义的 ToolRegistry)
    registry = ToolRegistry()
    registry.register("calculate", "计算数学表达式", {
        "type": "object",
        "properties": {
            "expression": {"type": "string", "description": "数学表达式"}
        },
        "required": ["expression"]
    }, calculate)
    
    # 2. 配置规则
    rules = RulesEngine()
    rules.add_input_rule(lambda x: (False, "输入不能为空") if not x.strip() else (True, ""))
    
    # 3. 创建日志记录器
    logger = Logger()
    
    # 4. 创建 Harness
    harness = AgentHarness(
        api_key="your-api-key",
        system_prompt="你是一个有用的 AI 助手。",
        registry=registry,
        rules=rules,
        logger=logger
    )
    
    # 5. 测试
    print("\n=== 测试 1:正常对话 ===")
    reply = harness.chat("你好!请帮我计算 25 * 48")
    print(f"回复: {reply}")
    
    print("\n=== 测试 2:规则拦截 ===")
    reply = harness.chat("")
    print(f"回复: {reply}")

运行效果

=== 测试 1:正常对话 ===
[user_input] {"content": "你好!请帮我计算 25 * 48"}...
[llm_call] {"iteration": 1}...
[tool_call] {"tool": "calculate"}...
[assistant_reply] {"content": "你好!25 * 48 的计算结果是 1200。"}...
回复: 你好!25 * 48 的计算结果是 1200。

=== 测试 2:规则拦截 ===
[user_input] {"content": ""}...
[rule_violation] {"rule": "input", "message": "输入不能为空"}...
回复: 输入被拒绝:输入不能为空

七、搭建 Agent 的注意事项

7.1 会话管理

问题 解决方案
Token 超出限制 实现消息裁剪策略,保留最近的 N 条消息或摘要
上下文丢失 定期总结对话历史,将摘要注入系统提示
多会话隔离 为每个用户/会话创建独立的 Session 实例
状态持久化 将会话数据保存到数据库,支持断线恢复

7.2 工具设计

问题 解决方案
工具描述不清 使用清晰、具体的描述,避免歧义
参数 schema 错误 严格遵循 JSON Schema 格式
工具执行超时 设置超时时间,处理超时错误
工具结果过大 截断或摘要过大的结果

7.3 安全

风险 防护措施
Prompt 注入 严格分离系统提示和用户输入
敏感信息泄露 过滤输出中的敏感数据
工具滥用 实现速率限制和权限控制
恶意代码执行 沙箱隔离工具执行环境

7.4 错误处理

# 重试机制示例
import time
def retry_with_backoff(func, max_retries=3, base_delay=1):
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            delay = base_delay * (2 ** attempt)
            print(f"重试 {attempt + 1}/{max_retries},等待 {delay} 秒...")
            time.sleep(delay)

这个重试函数用了指数退避。第一次失败等 1 秒,第二次等 2 秒,第三次等 4 秒。调用外部 API 的时候很实用,网络抖动或者服务端限流都能扛过去。

八、总结

这篇文章覆盖了这些内容:

  1. Agent Harness 的概念 - 为什么需要 Harness,它解决了什么问题
  2. Agent 技术演进 - 从 Prompt Engineering 到 Agent Harness 的发展历程
  3. 核心架构 - 五大组件及其职责
  4. Skill 系统 - 可插拔的能力模块设计
  5. 实战代码 - 从基础调用到完整 Harness 的 6 个渐进式示例

完整代码仓库

所有示例代码已开源到 GitHub:

https://github.com/Farewell-CK/learn_harness

欢迎 Star、Fork、提 Issue。

下一步

  • 尝试添加更多工具(文件操作、网络请求、数据库查询等)
  • 实现 Skill 系统并创建自定义 Skill
  • 添加评估和测试框架
  • 探索多 Agent 协作模式

如果觉得本文有帮助,欢迎点赞、收藏、关注。

有问题可以在评论区留言,或者在 GitHub 提 Issue 讨论。

常见问题(FAQ)

Agent Harness 和 LangChain 这类框架有什么区别?

Framework 定义 Agent 结构和行为,如 LangChain;Harness 是包裹 Agent 的基础设施,负责会话管理、工具注册、规则引擎等。一句话:Framework 是设计图纸,Harness 是物业管理系统。

Agent Loop 具体是怎么工作的?

Agent Loop 即 ReAct 循环:思考→行动→观察→判断。LLM 先分析状态决定行动,调用工具后观察结果,再判断任务是否完成。未完成则回到思考,完成则返回结果。

为什么说 Harness 是 Agent 发展的必然趋势?

Agent 越复杂,调试困难、行为不可控、无法评估等问题越突出。Harness 通过完整轨迹记录、规则引擎、自动化测试等,让 Agent 从“能跑”变成“可靠”,并实现标准化和可复用。

晓婷深圳
本文由 晓婷 审核,最后更新于 2026年9月22日
联系编辑 →
← 返回文章列表
分享到:微博

版权与免责声明:本文仅用于信息分享与交流,不构成任何形式的法律、投资、医疗或其他专业建议,也不构成对任何结果的承诺或保证。

文中提及的商标、品牌、Logo、产品名称及相关图片/素材,其权利归各自合法权利人所有。本站内容可能基于公开资料整理,亦可能使用 AI 辅助生成或润色;我们尽力确保准确与合规,但不保证完整性、时效性与适用性,请读者自行甄别并以官方信息为准。

若本文内容或素材涉嫌侵权、隐私不当或存在错误,请相关权利人/当事人联系本站,我们将及时核实并采取删除、修正或下架等处理措施。也请勿在评论或联系信息中提交身份证号、手机号、住址等个人敏感信息。