从零搭建 Agent Harness：手把手教你构建第一个 AI Agent（附完整代码）

摘要：本文带你从零开始理解 Agent Harness 的概念，深入讲解 Agent 技术演进、核心架构、Skill 系统设计，并使用智谱 GLM-5-Turbo 模型（OpenAI 兼容接口）动手搭建一个完整的 Agent。包含 6 个渐进式示例，从基础调用到完整 Harness，代码可直接运行。

一、什么是 Agent Harness？

1.1 从 LLM 到 Agent

一个普通的 LLM 调用是这样的：

用户输入 → LLM → 文本回复

这只是一个"问答机器"。要让它成为真正的 Agent（智能体），我们需要：

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

这些基础设施就是 Agent Harness。

1.2 Harness 的比喻

Harness = 马具

马（LLM）本身有力量，但需要马具来：
– 控制方向（规则引擎）
– 连接马车（工具系统）
– 记录行程（追踪系统）
– 管理骑手（会话管理）

1.3 为什么需要 Harness？

1.4 Harness vs Framework vs Runtime

这三个概念经常被混淆，理解它们的区别很重要：

一句话总结：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 解决的核心问题是：让 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)
├── 会话隔离（多用户场景）
└── 上下文压缩（摘要、向量检索）

消息裁剪策略对比：

组件二：工具注册中心 (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 的优势

五、环境准备

5.1 安装依赖

pip install openai pyyaml

5.2 配置智谱 API（OpenAI 兼容接口）

智谱提供 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/"
)

提示：使用 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:
"""最简会话管理器"""

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，我有以下建议…"}
]

6.3 第三步：工具注册中心

让 LLM 能调用外部工具：

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_all_schemas(self) –> list:
"""获取所有工具的 Schema（传给 LLM）"""
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)

# === 定义工具 ===
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 会话管理

7.2 工具设计

7.3 安全

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)

八、总结

通过本文，你学习了：

完整代码仓库

所有示例代码已开源到 GitHub：
👉 https://github.com/Farewell-CK/learn_harness

欢迎 Star、Fork、提 Issue！

下一步

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

如果觉得本文有帮助，欢迎点赞、收藏、关注！
有问题可以在评论区留言，或者在 GitHub 提 Issue 讨论。