本教程带你从零开始理解 Agent Harness 的概念，并使用智谱 GLM-5-Turbo 模型（OpenAI 兼容接口）动手搭建一个完整的 Agent。
- 第一章：什么是 Agent Harness
- 第二章：Agent 技术演进
- 第三章：核心架构解析
- 第四章：Skill 系统详解
- 第五章：环境准备
- 第六章：最小可用 Harness
- 第七章：工具注册系统
- 第八章：完整的 Agent Harness
- 第九章：搭建 Agent 的注意事项
一个普通的 LLM 调用是这样的：
用户输入 → LLM → 文本回复
这只是一个"问答机器"。要让它成为真正的 Agent（智能体），我们需要：
- 记住对话历史（会话管理）
- 调用外部工具（工具调用）
- 遵守规则约束（安全策略）
- 追踪执行过程（可观测性）
这些基础设施就是 Agent Harness。
Harness = 马具
马（LLM）本身有力量，但需要马具来：
- 控制方向（规则引擎）
- 连接马车（工具系统）
- 记录行程（追踪系统）
- 管理骑手（会话管理）
| 没有 Harness | 有 Harness |
|---|---|
| 每次对话都是独立的 | 记住上下文和历史 |
| 只能输出文本 | 可以调用工具执行操作 |
| 无法控制行为 | 有规则和约束 |
| 出错无法追溯 | 完整轨迹记录 |
| 无法测试评估 | 自动化评估框架 |
这三个概念经常被混淆，理解它们的区别很重要：
┌─────────────────────────────────────────────────────────┐
│ 完整 AI Agent 栈 │
│ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Framework（框架） │ │
│ │ 职责：定义和配置 Agent │ │
│ │ 举例：LangChain, LlamaIndex, CrewAI │ │
│ │ 类比：乐高积木的设计图纸 │ │
│ └───────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Runtime（运行时） │ │
│ │ 职责：运行 Agent 的环境 │ │
│ │ 举例：Vercel AI SDK, OpenAI Agents SDK │ │
│ │ 类比：乐高积木的拼装平台 │ │
│ └───────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ Harness（马具/基础设施） │ │
│ │ 职责：包裹 Agent 的基础设施 │ │
│ │ 举例：会话管理、工具注册、规则引擎、评估系统 │ │
│ │ 类比：乐高积木的展示底座和防护罩 │ │
│ └───────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌───────────────────────────────────────────────────┐ │
│ │ LLM（模型） │ │
│ │ 职责：推理和生成 │ │
│ │ 举例：GLM-5-Turbo, GPT-4, Claude │ │
│ │ 类比：乐高积木本身 │ │
│ └───────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────┘
| 层级 | 职责 | 解决的核心问题 | 举例 |
|---|---|---|---|
| Framework | 定义 Agent 结构和行为 | 如何组织 Agent 的逻辑 | LangChain, CrewAI |
| Runtime | 提供运行环境 | 如何执行 Agent 代码 | Vercel AI SDK |
| Harness | 包裹 Agent 的基础设施 | 如何管理、控制、评估 Agent | 会话管理、工具注册 |
| LLM | 推理和生成 | 如何理解和生成内容 | GLM-5-Turbo, GPT-4 |
阶段 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 基础设施
└── 突破：标准化、可评估、可观测、可复用
Agent 的核心是一个循环过程，称为 ReAct Loop：
┌──────────────────────────────────────────────────────┐
│ Agent 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 "达到最大迭代次数"随着 Agent 越来越复杂，以下问题变得突出：
| 问题 | 没有 Harness | 有 Harness |
|---|---|---|
| 调试困难 | Agent 出错不知原因 | 完整轨迹记录，可回放 |
| 行为不可控 | LLM 可能做危险操作 | 规则引擎拦截 |
| 无法评估 | 不知道 Agent 好不好 | 自动化测试和评分 |
| 难以复用 | 每个项目重新造轮子 | 标准化组件可插拔 |
| 多模型切换 | 代码耦合特定 SDK | 统一接口，热切换 |
Harness 解决的核心问题是：让 Agent 从"能跑"变成"可靠"。
graph TB
subgraph Harness [Agent Harness]
direction TB
Input[输入处理] --> Context[上下文管理]
Context --> Tools[工具注册中心]
Tools --> Rules[规则引擎]
Rules --> LLM[(LLM 模型)]
LLM --> Response[响应处理]
Response --> Decision{类型?}
Decision -->|文本| Output[返回用户]
Decision -->|工具调用| Execute[执行工具]
Execute -->|结果| Context
Obs[可观测性] -.-> Input
Obs -.-> Context
Obs -.-> Tools
Obs -.-> LLM
Obs -.-> Response
end
User[用户] --> Input
Output --> User
职责：
├── 维护对话历史 (message history)
├── 管理上下文窗口 (context window)
├── 会话状态持久化
└── 多轮对话的状态跟踪
关键设计：
├── 消息裁剪策略（超出 token 限制时如何处理）
├── 系统提示词管理 (system prompt)
├── 会话隔离（多用户场景）
└── 上下文压缩（摘要、向量检索）
消息裁剪策略对比：
| 策略 | 优点 | 缺点 | 适用场景 |
|---|---|---|---|
| 保留最近 N 条 | 简单高效 | 可能丢失重要上下文 | 短对话 |
| 摘要压缩 | 保留语义信息 | 需要额外调用 LLM | 长对话 |
| 重要性评分 | 保留关键信息 | 实现复杂 | 复杂任务 |
| 向量检索 | 按需检索相关历史 | 需要向量数据库 | 超长对话 |
职责：
├── 工具定义（名称、描述、参数 schema）
├── 工具注册/注销
├── 工具调用路由
└── 工具执行结果格式化
关键设计：
├── 统一的工具接口
├── 动态工具加载
├── 工具权限控制
└── 跨 SDK 兼容（OpenAI / Anthropic）
工具调用格式（OpenAI 标准）：
{
"type": "function",
"function": {
"name": "calculate",
"description": "计算数学表达式",
"parameters": {
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "要计算的表达式"
}
},
"required": ["expression"]
}
}
}职责：
├── 输入规则（内容过滤、注入防御）
├── 输出规则（格式校验、敏感信息过滤）
├── 工具规则（权限检查、沙箱隔离）
└── 降级策略（LLM 出错时怎么办）
关键设计：
├── 规则优先级
├── 规则热更新
├── 违规处理（拦截、警告、终止）
└── 规则日志
职责：
├── 动态注入相关知识
├── RAG 检索增强
├── 环境变量注入
└── 用户偏好/记忆
关键设计：
├── 上下文相关性评分
├── 上下文窗口优化
├── 缓存策略
└── 多源上下文融合
职责：
├── 记录 Agent 决策轨迹
├── 工具调用日志
├── 性能指标收集（延迟、Token 用量、成本）
└── 自动化测试用例
关键设计：
├── 轨迹序列化
├── 回放能力
├── 评分标准
└── 可视化面板
用户: "帮我搜索最新的 React 19 特性并总结"
│
↓
[输入层] 验证输入格式、权限检查 ✓
│
↓
[上下文层] 加载对话历史 + 用户偏好 + 相关知识
│
↓
[工具层] 可用工具: search_web, read_file, calculate...
│
↓
[规则层] 检查: 允许搜索外部网站 ✓
│
↓
[LLM层] 构建 Prompt → 调用 GLM-5-Turbo
│
↓
[响应层] 模型返回: tool_call("search_web", {query: "React 19 features"})
│
↓
[工具层] 执行 search_web → 返回搜索结果
│
↓
[上下文层] 将搜索结果注入对话历史
│
↓
[LLM层] 再次调用模型（带着搜索结果）
│
↓
[响应层] 模型返回: 文本回复（总结 React 19 特性）
│
↓
[规则层] 检查输出格式、敏感信息 ✓
│
↓
[可观测性] 记录完整轨迹、Token 用量、延迟
│
↓
返回给用户
Skill（技能） 是 Harness 中 可插拔的能力模块，它将特定领域的知识、工具、规则和工作流封装在一起。
Skill = 领域知识 + 工具定义 + 行为规则 + 工作流
┌─────────────────────────────────────────────────────┐
│ Agent Harness │
│ │
│ ┌───────────────────────────────────────────────┐ │
│ │ 工具注册中心 (Tool Registry) │ │
│ │ ┌───────────┬──────────────┬───────────────┐ │ │
│ │ │ 内置工具 │ Skills 技能 │ 自定义工具 │ │ │
│ │ │ │ │ │ │ │
│ │ │ • Bash │ • ssh-sync │ • 业务 API │ │ │
│ │ │ • Read │ • tavily │ • 数据库查询 │ │ │
│ │ │ • Edit │ • web-crawl │ • 第三方服务 │ │ │
│ │ │ • Glob │ • find-skills│ │ │ │
│ │ └───────────┴──────────────┴───────────────┘ │ │
│ └───────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────┐ │
│ │ 规则引擎 (Rules Engine) │ │
│ │ → Skills 自带行为约束和执行规则 │ │
│ └───────────────────────────────────────────────┘ │
│ │
│ ┌───────────────────────────────────────────────┐ │
│ │ 上下文注入 (Context Injector) │ │
│ │ → Skills 提供领域知识 (SKILL.md) │ │
│ └───────────────────────────────────────────────┘ │
│ │
│ ↕ │
│ LLM (模型) │
└─────────────────────────────────────────────────────┘
一个标准的 Skill 目录结构：
skills/
└── ssh-dev-sync/
├── SKILL.md # 核心文件：触发条件 + 工作流 + 领域知识
├── scripts/ # 可执行脚本
│ └── sync.sh
├── reference/ # 参考资料
│ └── api-docs.md
└── templates/ # 模板文件
└── config.yaml
---
name: ssh-dev-sync
description: 远程服务器开发工作流，支持代码同步、命令执行、日志获取
trigger: |
当用户想要：
- 同步代码到远程服务器
- 在远程服务器执行命令
- 获取远程服务器日志
---
# SSH Dev Sync
## 工作流
1. 连接到远程服务器
2. 同步代码（git push + git pull）
3. 执行命令
4. 获取输出
## 配置
```yaml
host: your-server-ip
user: username
port: 22- 不要在生产环境执行危险命令
- 同步前确认远程分支状态
### 4.5 Skill 的生命周期
┌─────────────────────────────────────────────────────┐ │ Skill 生命周期 │ │ │ │ 1. 发现 (Discover) │ │ └── 用户表达需求 → 匹配 Skill trigger │ │ │ │ 2. 加载 (Load) │ │ └── 读取 SKILL.md → 注入上下文 │ │ │ │ 3. 注册 (Register) │ │ └── 注册工具到 Tool Registry │ │ 注册规则到 Rules Engine │ │ │ │ 4. 执行 (Execute) │ │ └── 按照工作流执行 │ │ │ │ 5. 卸载 (Unload) │ │ └── 清理资源，释放内存 │ └─────────────────────────────────────────────────────┘
### 4.6 Skill 的代码实现
```python
import os
import yaml
from typing import Dict, Any, List
class Skill:
"""Skill 类：封装领域知识、工具、规则和工作流"""
def __init__(self, skill_path: str):
self.skill_path = skill_path
self.name = ""
self.description = ""
self.trigger = ""
self.tools = []
self.rules = []
self.context = ""
# 加载 SKILL.md
self._load_skill()
def _load_skill(self):
"""解析 SKILL.md 文件"""
skill_file = os.path.join(self.skill_path, "SKILL.md")
with open(skill_file, 'r', encoding='utf-8') as f:
content = f.read()
# 解析 frontmatter（--- 之间的元数据）
parts = content.split("---")
if len(parts) >= 3:
metadata = yaml.safe_load(parts[1])
self.name = metadata.get("name", "")
self.description = metadata.get("description", "")
self.trigger = metadata.get("trigger", "")
self.context = parts[2].strip()
def register_tools(self, registry):
"""将 Skill 的工具注册到 Tool Registry"""
# 从 scripts/ 目录加载可执行脚本
scripts_dir = os.path.join(self.skill_path, "scripts")
if os.path.exists(scripts_dir):
for script in os.listdir(scripts_dir):
# 注册脚本为工具
pass
def register_rules(self, rules_engine):
"""将 Skill 的规则注册到 Rules Engine"""
for rule in self.rules:
rules_engine.add_tool_rule(rule)
def get_context(self) -> str:
"""获取 Skill 提供的领域知识"""
return self.context
def matches_trigger(self, user_input: str) -> bool:
"""检查用户输入是否匹配此 Skill 的触发条件"""
# 简单的关键词匹配，实际可用更复杂的 NLP
trigger_keywords = self.trigger.lower().split()
user_lower = user_input.lower()
return any(kw in user_lower for kw in trigger_keywords if len(kw) > 3)
class SkillManager:
"""Skill 管理器：负责 Skill 的发现、加载和生命周期管理"""
def __init__(self, skills_dir: str):
self.skills_dir = skills_dir
self.skills: Dict[str, Skill] = {}
self.active_skills: List[Skill] = []
# 扫描并加载所有 Skill
self._discover_skills()
def _discover_skills(self):
"""扫描 skills 目录，发现所有 Skill"""
if not os.path.exists(self.skills_dir):
return
for name in os.listdir(self.skills_dir):
skill_path = os.path.join(self.skills_dir, name)
if os.path.isdir(skill_path):
skill_file = os.path.join(skill_path, "SKILL.md")
if os.path.exists(skill_file):
skill = Skill(skill_path)
self.skills[skill.name] = skill
def activate_matching_skills(self, user_input: str):
"""激活匹配用户输入的 Skill"""
for skill in self.skills.values():
if skill.matches_trigger(user_input):
self.activate_skill(skill.name)
def activate_skill(self, name: str):
"""激活指定 Skill"""
if name not in self.skills:
return
skill = self.skills[name]
if skill not in self.active_skills:
self.active_skills.append(skill)
def get_active_context(self) -> str:
"""获取所有激活 Skill 的上下文"""
contexts = []
for skill in self.active_skills:
ctx = skill.get_context()
if ctx:
contexts.append(f"## {skill.name}\n{ctx}")
return "\n\n".join(contexts)
class AgentHarnessWithSkills:
"""集成 Skill 系统的 Agent Harness"""
def __init__(self, api_key, system_prompt, registry, rules, logger, skills_dir):
self.client = OpenAI(api_key=api_key, base_url="...")
self.registry = registry
self.rules = rules
self.logger = logger
self.skill_manager = SkillManager(skills_dir)
self.messages = [{"role": "system", "content": system_prompt}]
def chat(self, user_input, max_iterations=5):
# 1. 激活匹配的 Skill
self.skill_manager.activate_matching_skills(user_input)
# 2. 注入 Skill 上下文
skill_context = self.skill_manager.get_active_context()
if skill_context:
self.messages.append({
"role": "system",
"content": f"## 激活的技能知识\n{skill_context}"
})
# 3. 注册 Skill 的工具和规则
for skill in self.skill_manager.active_skills:
skill.register_tools(self.registry)
skill.register_rules(self.rules)
# 4. 正常的 Harness 流程
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=2048
)
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:
result = self.registry.execute(tool_name, arguments)
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 "达到最大迭代次数"| 特性 | 没有 Skill | 有 Skill |
|---|---|---|
| 知识复用 | 每次重新编写提示词 | SKILL.md 可复用 |
| 按需加载 | 所有工具始终可用 | 匹配触发才加载 |
| 领域专精 | 通用能力 | 特定领域深度优化 |
| 社区共享 | 闭源实现 | 标准化格式可分享 |
| 热插拔 | 硬编码在代码中 | 目录即插即用 |
pip install openai pyyaml智谱提供 OpenAI 兼容接口，只需更换 api_key 和 base_url：
from openai import OpenAI
client = OpenAI(
api_key="your-api-key",
base_url="https://open.bigmodel.cn/api/paas/v4/"
)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": "你好"}]
)
print(response.choices[0].message.content)文件: examples/01_basic_chat.py
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 开发的大语言模型。
我可以帮助你回答问题、创作文字、进行逻辑推理、编程等任务。
有什么我可以帮助你的吗？
说明: 这是最基础的 LLM 调用，没有 Harness 的任何组件。每次调用都是独立的，没有记忆。
文件: examples/02_session_management.py
from openai import OpenAI
class SimpleSession:
"""最简会话管理器"""
def __init__(self, system_prompt=None):
self.messages = []
if system_prompt:
self.messages.append({"role": "system", "content": system_prompt})
def add_user_message(self, content):
self.messages.append({"role": "user", "content": content})
def add_assistant_message(self, content):
self.messages.append({"role": "assistant", "content": content})
def get_messages(self):
return self.messages
class MinimalHarness:
"""最小可用 Harness"""
def __init__(self, api_key, system_prompt=None):
self.client = OpenAI(
api_key=api_key,
base_url="https://open.bigmodel.cn/api/paas/v4/"
)
self.session = SimpleSession(system_prompt)
def chat(self, user_input):
# 1. 添加用户输入到会话
self.session.add_user_message(user_input)
# 2. 调用 LLM
response = self.client.chat.completions.create(
model="glm-5-turbo",
messages=self.session.get_messages(),
max_tokens=1024,
temperature=0.7
)
# 3. 获取回复并保存到会话
reply = response.choices[0].message.content
self.session.add_assistant_message(reply)
return 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，我有以下建议..."}
]文件: examples/03_system_prompt.py
harness = MinimalHarness(
api_key="your-api-key",
system_prompt="""你是一个专业的编程导师，专门帮助初学者学习编程。
请遵循以下原则：
1. 用简单易懂的语言解释概念
2. 提供具体的代码示例
3. 鼓励用户动手实践
4. 指出常见的错误和陷阱"""
)
reply = harness.chat("什么是变量？")
print(f"AI: {reply}")输入:
用户: "什么是变量？"
输出:
AI: 变量是编程中最基本的概念之一。让我用简单的方式解释：
想象你有一个盒子，你可以在里面放不同的东西。变量就是这个"盒子"。
# Python 示例
name = "小明" # 创建一个叫 name 的盒子，里面装着 "小明"
age = 25 # 创建一个叫 age 的盒子，里面装着 25
print(name) # 输出：小明
print(age) # 输出：25
常见陷阱：
- 变量名不能以数字开头
- 不要使用 Python 关键字作为变量名
说明: 系统提示词（System Prompt）是 Harness 中 上下文注入 的重要部分。它定义了 AI 的行为模式和回复风格。好的系统提示词可以显著提升输出质量。
工具调用让 LLM 不仅能"说话"，还能"做事"：
用户: "现在几点了？"
LLM: 我不知道当前时间...
↓ 工具调用
调用 get_current_time()
↓
返回: "2024-01-15 14:30:00"
↓
LLM: 现在是下午 2 点 30 分。
文件: examples/04_tool_registry.py
import json
from typing import Callable, Dict, Any
from datetime import datetime
class ToolRegistry:
"""工具注册中心"""
def __init__(self):
self.tools: Dict[str, Dict[str, Any]] = {}
def register(self, name: str, description: str,
parameters: dict, func: Callable):
"""注册一个工具"""
self.tools[name] = {
"name": name,
"description": description,
"parameters": parameters,
"function": func
}
def get_tool_schema(self, name: str) -> dict:
"""获取工具的 JSON Schema（用于传给 LLM）"""
tool = self.tools[name]
return {
"type": "function",
"function": {
"name": tool["name"],
"description": tool["description"],
"parameters": tool["parameters"]
}
}
def get_all_schemas(self) -> list:
"""获取所有工具的 Schema"""
return [self.get_tool_schema(name) for name in self.tools]
def execute(self, name: str, arguments: dict) -> Any:
"""执行工具"""
if name not in self.tools:
raise ValueError(f"工具 {name} 未注册")
return self.tools[name]["function"](**arguments)
# === 定义工具 ===
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)}"
def search_web(query: str) -> str:
"""搜索网络信息（模拟）"""
return f"搜索结果：关于 '{query}' 的相关信息..."
# === 注册工具 ===
registry = ToolRegistry()
registry.register(
name="get_current_time",
description="获取当前日期和时间",
parameters={
"type": "object",
"properties": {
"timezone": {
"type": "string",
"description": "时区，如 UTC, CST, PST",
"default": "UTC"
}
},
"required": []
},
func=get_current_time
)
registry.register(
name="calculate",
description="计算数学表达式",
parameters={
"type": "object",
"properties": {
"expression": {
"type": "string",
"description": "要计算的表达式，如 2+3*4"
}
},
"required": ["expression"]
},
func=calculate
)
registry.register(
name="search_web",
description="搜索网络获取信息",
parameters={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词"
}
},
"required": ["query"]
},
func=search_web
)
print("已注册工具:", list(registry.tools.keys()))
# 输出: 已注册工具: ['get_current_time', 'calculate', 'search_web']说明: 工具注册中心负责：
- 定义工具: 名称、描述、参数 schema
- 注册工具: 存储工具定义和执行函数
- 生成 Schema: 将工具定义转为 LLM 可理解的格式
- 执行工具: 根据 LLM 的调用请求执行对应函数
文件: examples/05_tool_calling.py
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):
# 调用 LLM
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:
# 保存 AI 的工具调用请求
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="你是一个有用的助手。当需要获取时间、计算或搜索时，请使用相应的工具。"
)
# 示例 1：需要调用计算工具
print("\n=== 示例 1：数学计算 ===")
reply = harness.chat("请帮我计算 123 * 456 + 789")
print(f"AI: {reply}")
# 示例 2：需要调用时间工具
print("\n=== 示例 2：获取时间 ===")
reply = harness.chat("现在几点了？")
print(f"AI: {reply}")
# 示例 3：多轮对话（带记忆）
print("\n=== 示例 3：多轮对话 ===")
reply = harness.chat("刚才计算的结果是多少来着？")
print(f"AI: {reply}")输入:
用户: "请帮我计算 123 * 456 + 789"
执行过程:
🔧 调用工具: calculate
参数: {"expression": "123 * 456 + 789"}
结果: 56877
输出:
AI: 计算结果：123 * 456 + 789 = 56877
说明: 这个例子展示了完整的工具调用流程：
- 用户提问
- LLM 判断需要调用工具
- Harness 执行工具并获取结果
- 将结果返回给 LLM
- LLM 生成最终回复
消息历史变化:
[用户] "请帮我计算 123 * 456 + 789"
↓
[AI] tool_call: calculate({"expression": "123 * 456 + 789"})
↓
[工具] 56877
↓
[AI] "计算结果：123 * 456 + 789 = 56877"
文件: examples/06_full_harness.py
import json
from datetime import datetime
from typing import Callable, Dict, Any, List
from openai import OpenAI
# ==========================================
# 组件 1：工具注册中心
# ==========================================
class ToolRegistry:
def __init__(self):
self.tools: Dict[str, Dict[str, Any]] = {}
def register(self, name: str, description: str,
parameters: dict, func: Callable):
self.tools[name] = {
"name": name,
"description": description,
"parameters": parameters,
"function": func
}
def get_all_schemas(self) -> list:
return [{
"type": "function",
"function": {
"name": t["name"],
"description": t["description"],
"parameters": t["parameters"]
}
} for t in self.tools.values()]
def execute(self, name: str, arguments: dict) -> Any:
return self.tools[name]["function"](**arguments)
# ==========================================
# 组件 2：规则引擎
# ==========================================
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, ""
# ==========================================
# 组件 3：可观测性（日志记录）
# ==========================================
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]}...")
def get_trajectory(self) -> list:
return self.logs
# ==========================================
# 组件 4：完整的 Agent Harness
# ==========================================
class AgentHarness:
def __init__(self, api_key: str, system_prompt: str,
registry: ToolRegistry,
rules: RulesEngine,
logger: 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, "messages_count": len(self.messages)})
# 调用 LLM
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:
self.logger.log("rule_violation", {"rule": "tool", "message": msg})
result = f"工具调用被拒绝：{msg}"
else:
# 执行工具
self.logger.log("tool_call", {"tool": tool_name, "args": arguments})
try:
result = self.registry.execute(tool_name, arguments)
self.logger.log("tool_result", {"tool": tool_name, "result": str(result)[:100]})
except Exception as e:
result = f"工具执行错误: {str(e)}"
self.logger.log("tool_error", {"tool": tool_name, "error": 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:
self.logger.log("rule_violation", {"rule": "output", "message": msg})
return "输出被规则拦截"
self.logger.log("assistant_reply", {"content": reply[:100]})
return reply
return "达到最大迭代次数"
# ==========================================
# 使用示例
# ==========================================
if __name__ == "__main__":
# 1. 定义工具
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)}"
def read_file(path: str) -> str:
try:
with open(path, 'r', encoding='utf-8') as f:
return f.read()[:500]
except Exception as e:
return f"读取错误: {str(e)}"
# 2. 注册工具
registry = ToolRegistry()
registry.register("get_current_time", "获取当前时间", {
"type": "object",
"properties": {
"timezone": {"type": "string", "description": "时区"}
}
}, get_current_time)
registry.register("calculate", "计算数学表达式", {
"type": "object",
"properties": {
"expression": {"type": "string", "description": "数学表达式"}
},
"required": ["expression"]
}, calculate)
registry.register("read_file", "读取文件内容", {
"type": "object",
"properties": {
"path": {"type": "string", "description": "文件路径"}
},
"required": ["path"]
}, read_file)
# 3. 配置规则
rules = RulesEngine()
# 输入规则：禁止空输入
rules.add_input_rule(lambda x: (False, "输入不能为空") if not x.strip() else (True, ""))
# 输入规则：禁止危险命令
rules.add_input_rule(lambda x: (False, "包含危险命令") if "删除所有文件" in x else (True, ""))
# 工具规则：禁止读取敏感文件
def check_read_file(tool_name, args):
if tool_name == "read_file":
path = args.get("path", "")
if any(s in path for s in [".env", "password", "secret", "credential"]):
return False, "禁止读取敏感文件"
return True, ""
rules.add_tool_rule(check_read_file)
# 输出规则：禁止包含敏感信息
rules.add_output_rule(lambda x: (False, "包含敏感信息") if "password" in x.lower() else (True, ""))
# 4. 创建日志记录器
logger = Logger()
# 5. 创建 Harness
harness = AgentHarness(
api_key="your-api-key",
system_prompt="你是一个有用的 AI 助手，可以帮助用户完成各种任务。",
registry=registry,
rules=rules,
logger=logger
)
# 6. 测试
print("\n=== 测试 1：正常对话 ===")
reply = harness.chat("你好！请帮我计算 25 * 48")
print(f"回复: {reply}")
print("\n=== 测试 2：规则拦截 ===")
reply = harness.chat("")
print(f"回复: {reply}")
print("\n=== 测试 3：查看执行轨迹 ===")
print(f"总日志数: {len(logger.logs)}")
for log in logger.logs:
print(f" [{log['type']}] {log['timestamp']}")=== 测试 1：正常对话 ===
[user_input] {"content": "你好！请帮我计算 25 * 48"}...
[llm_call] {"iteration": 1, "messages_count": 2}...
[tool_call] {"tool": "calculate", "args": {"expression": "25 * 48"}}...
[tool_result] {"tool": "calculate", "result": "1200"}...
[assistant_reply] {"content": "你好！25 * 48 的计算结果是 1200。"}...
回复: 你好！25 * 48 的计算结果是 1200。
=== 测试 2：规则拦截 ===
[user_input] {"content": ""}...
[rule_violation] {"rule": "input", "message": "输入不能为空"}...
回复: 输入被拒绝：输入不能为空
=== 测试 3：查看执行轨迹 ===
总日志数: 7
[user_input] 2024-01-15T14:30:00
[llm_call] 2024-01-15T14:30:01
[tool_call] 2024-01-15T14:30:02
[tool_result] 2024-01-15T14:30:02
[assistant_reply] 2024-01-15T14:30:03
[user_input] 2024-01-15T14:30:05
[rule_violation] 2024-01-15T14:30:05
| 问题 | 解决方案 |
|---|---|
| Token 超出限制 | 实现消息裁剪策略，保留最近的 N 条消息或摘要 |
| 上下文丢失 | 定期总结对话历史，将摘要注入系统提示 |
| 多会话隔离 | 为每个用户/会话创建独立的 Session 实例 |
| 状态持久化 | 将会话数据保存到数据库，支持断线恢复 |
# Token 管理示例
class TokenAwareSession:
MAX_TOKENS = 8000
def trim_messages(self):
while self.estimate_tokens() > self.MAX_TOKENS:
for i, msg in enumerate(self.messages):
if msg["role"] != "system":
self.messages.pop(i)
break| 问题 | 解决方案 |
|---|---|
| 工具描述不清 | 使用清晰、具体的描述，避免歧义 |
| 参数 schema 错误 | 严格遵循 JSON Schema 格式，测试边界情况 |
| 工具执行超时 | 设置超时时间，处理超时错误 |
| 工具权限问题 | 实现权限检查，防止未授权操作 |
| 工具结果过大 | 截断或摘要过大的结果 |
| 问题 | 解决方案 |
|---|---|
| 规则冲突 | 定义规则优先级，避免互相矛盾 |
| 规则过多影响性能 | 使用规则引擎优化，或缓存规则检查结果 |
| 规则更新 | 支持热更新，无需重启服务 |
| 误拦截 | 提供规则白名单和申诉机制 |
| 错误类型 | 处理策略 |
|---|---|
| API 调用失败 | 重试机制（指数退避） |
| 工具执行错误 | 返回错误信息给 LLM，让它尝试修复 |
| LLM 输出格式错误 | 解析失败时重试或返回默认值 |
| 网络超时 | 设置合理的超时时间，实现优雅降级 |
# 重试机制示例
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)| 风险 | 防护措施 |
|---|---|
| Prompt 注入 | 严格分离系统提示和用户输入 |
| 敏感信息泄露 | 过滤输出中的敏感数据 |
| 工具滥用 | 实现速率限制和权限控制 |
| 恶意代码执行 | 沙箱隔离工具执行环境 |
| 优化点 | 方法 |
|---|---|
| 减少 Token 消耗 | 精简系统提示，压缩上下文 |
| 缓存常见结果 | 对相同输入缓存 LLM 回复 |
| 并行工具调用 | 多个独立工具同时执行 |
| 流式输出 | 使用 stream=True 提升用户体验 |
| 测试类型 | 方法 |
|---|---|
| 单元测试 | 测试每个组件的独立功能 |
| 集成测试 | 测试组件间的协作 |
| 端到端测试 | 模拟完整用户场景 |
| 回归测试 | 确保修改不破坏现有功能 |
通过本教程，你学习了：
- Agent Harness 的概念 - 为什么需要 Harness，它解决了什么问题
- Agent 技术演进 - 从 Prompt Engineering 到 Agent Harness 的发展历程
- 核心架构 - 五大组件及其职责
- Skill 系统 - 可插拔的能力模块设计
- 逐步构建 - 从基础调用到完整 Harness
- 最佳实践 - 搭建 Agent 时的注意事项
- 尝试添加更多工具（文件操作、网络请求、数据库查询等）
- 实现 Skill 系统并创建自定义 Skill
- 添加评估和测试框架
- 探索多 Agent 协作模式
learn_harness/
├── README.md # 本教程
├── examples/
│ ├── 01_basic_chat.py # 基础 LLM 调用
│ ├── 02_session_management.py # 会话管理
│ ├── 03_system_prompt.py # 系统提示词
│ ├── 04_tool_registry.py # 工具注册中心
│ ├── 05_tool_calling.py # 工具调用
│ └── 06_full_harness.py # 完整 Harness
├── skills/
│ └── example-skill/
│ ├── SKILL.md
│ ├── scripts/
│ └── reference/
└── tests/
└── test_harness.py # 测试用例