Agent Harness 的单元测试策略：构建可靠 AI Agent 的基石

---

一、引言 (Introduction)

1.1 钩子：你是否也被 Agent 的“不可控”折磨过？

你是否有过这样的经历：花费数周时间开发了一个智能客服 Agent，上线前手动测试了几十轮都没问题，结果上线第一天就收到了一堆投诉——用户问“我的订单什么时候发货”，Agent 却答非所问地推荐起了新产品；调用物流查询工具时，明明参数正确却频繁报错；更糟的是，出了问题根本没法复现，因为 Agent 的行为依赖于前几轮的对话历史，每次测试的结果都不一样……

如果你有过类似的经历，那么你一定深刻体会到：AI Agent 的开发，最难的不是“让它跑起来”，而是“让它可靠地跑起来”。而要实现“可靠”，单元测试是绕不开的基石——尤其是对于基于 Agent Harness 这类框架开发的应用，模块化的设计既给了我们灵活性，也给了我们通过单元测试锁定每个组件行为的机会。

1.2 定义问题：Agent Harness 是什么？为什么它的单元测试如此特殊？

在深入探讨测试策略之前，我们先明确两个核心概念：

什么是 Agent Harness？

Agent Harness 是一类用于快速构建、部署和管理 AI Agent的开发框架（本文中我们将其抽象为一个通用的模块化框架，实际开发中可能是 LangChain、AutoGPT 开发工具包，或是企业内部的定制框架）。它的核心价值在于将 Agent 拆解为可复用的组件，典型的组件包括：

• Agent Core：负责意图识别、决策生成（是直接回复还是调用工具）；
• Tool Registry：管理 Agent 可以调用的工具（如查询数据库、调用第三方 API）；
• Memory Store：存储对话历史、用户偏好、中间执行状态（支持短期记忆和长期记忆）；
• Planner：将复杂任务分解为可执行的子任务序列；
• Executor：负责执行子任务、调用工具、处理异常和流程控制；
• Observer：收集日志、指标，用于调试和监控。

一个典型的 Agent Harness 工作流程是：用户输入 → 记忆检索 → 意图识别 → 规划（可选） → 决策 → 执行 → 更新记忆 → 返回结果。

为什么 Agent Harness 的单元测试不能照搬传统方法？

传统软件的单元测试已经有非常成熟的方法论（如 FIRST 原则：Fast、Isolated、Repeatable、Self-validating、Timely），但 AI Agent 有三个本质特性，让传统测试方法“水土不服”：

1. 不确定性：Agent 的决策通常依赖大语言模型（LLM），而 LLM 的输出具有天然的随机性——同样的输入，即使 temperature 设为 0，也可能因为模型内部的微小差异产生不同的输出；
2. 状态依赖：Agent 的行为高度依赖历史对话、用户偏好等“记忆”，测试用例之间容易互相影响；
3. 外部交互：Agent 需要频繁调用工具、API、数据库等外部依赖，这些依赖的可用性会直接影响测试的稳定性和速度。

如果我们用测试传统 CRUD 应用的方法来测试 Agent，要么测试用例极其脆弱（稍微改一下 prompt 就全挂），要么测试慢得离谱（每次都要调用真实的 LLM 和 API），甚至可能因为外部依赖的问题导致测试结果不可信。

1.3 亮明观点：本文将带你构建一套“接地气”的 Agent Harness 单元测试体系

读完这篇文章，你将能够：

• 理解 Agent Harness 各个组件的测试痛点和对应策略；
• 从零搭建一套可复用的 Agent 测试环境（基于 Python + pytest）；
• 掌握 Mock LLM、外部 API、记忆存储的最佳实践；
• 学会处理 Agent 的“不确定性”，写出稳定、可维护的测试用例；
• 了解 Agent 测试的行业最佳实践和未来趋势。

本文将以一个通用的 Agent Harness 框架为例，结合大量实战代码、图表和数学模型，循序渐进地讲解——即使你使用的是具体的框架（如 LangChain），这些策略也完全可以迁移使用。

---

二、基础知识与背景铺垫：先搞懂“测什么”，再想“怎么测”

在开始写测试用例之前，我们需要先夯实基础：明确 Agent Harness 每个组件的职责，理解单元测试的核心原则，以及选对合适的技术栈。

2.1 Agent Harness 核心组件深度解析

要测好一个组件，首先要知道它“负责什么、输入什么、输出什么、依赖什么”。我们先把 Agent Harness 的核心组件拆解清楚：

2.1.1 组件定义与职责

组件名称	核心职责	输入	输出	依赖
Agent Core	意图识别、决策生成（直接回复/调用工具）、结果解析	用户输入、相关记忆	决策对象（含回复或工具调用信息）	LLM Client、Memory Store
Tool Registry	工具的注册、检索、参数验证	工具名称/查询、参数	工具实例/验证结果	无（或工具元数据存储）
Memory Store	记忆的添加、检索、更新、删除	记忆文本、查询条件、top-k	检索到的记忆列表/操作状态	向量嵌入模型、存储后端（如 ChromaDB）
Planner	复杂任务分解为子任务序列	用户任务、相关记忆	子任务序列（含依赖关系）	LLM Client、Memory Store
Executor	子任务执行、工具调用、流程控制、结果聚合	子任务序列	最终执行结果	Tool Registry、Memory Store
LLM Client	封装 LLM API 调用、处理 API 异常	Prompt、模型参数	LLM 生成的文本	外部 LLM API（如 OpenAI GPT-4）

2.1.2 组件交互流程（Mermaid 架构图）

为了更直观地理解组件之间的关系，我们用 Mermaid 画一个交互架构图：

这个流程图清晰地展示了 Agent 的完整工作流，也为我们指明了单元测试的“隔离边界”：每个组件的测试都应该只关注自身的逻辑，而把依赖的其他组件 Mock 掉——比如测试 Agent Core 时，不需要真实的 Memory Store 和 LLM Client，只需要 Mock 它们的行为即可。

2.2 单元测试的核心原则与 Agent 测试的“适配”

传统单元测试的 FIRST 原则（Fast、Isolated、Repeatable、Self-validating、Timely）依然是 Agent 测试的基石，但我们需要针对 Agent 的特性做一些“适配”：

2.2.1 FIRST 原则的 Agent 适配版

原则	传统定义	Agent 测试的适配
Fast（快速）	测试用例要在毫秒级完成，避免 IO 操作	必须 Mock 所有外部依赖（LLM、API、数据库），绝对不要在单元测试中调用真实的 LLM
Isolated（隔离）	测试用例之间互不影响，每个测试只关注一个组件	每个测试用例使用独立的 Mock 对象，测试前后清理状态（如清空 Memory Store）
Repeatable（可重复）	同样的输入必须产生同样的输出	通过固定 LLM 的输出（Mock）、固定随机种子、使用固定的向量嵌入来消除随机性
Self-validating（自验证）	测试用例必须有明确的断言，不需要人工判断	避免“精确匹配字符串”的断言，改用“语义断言”或“结构断言”（如“回复包含订单号”“输出是合法 JSON”）
Timely（及时）	测试用例要在代码写完之前或之后尽快编写	对于 Agent 核心逻辑，尽量采用 TDD（测试驱动开发）；对于 LLM 相关的逻辑，可以先写代码再写测试

2.2.2 Agent 单元测试与集成测试的边界

很多开发者容易混淆 Agent 的单元测试和集成测试，我们在这里明确一下：

• 单元测试：只测试单个组件的逻辑，所有依赖都 Mock 掉，目标是验证“这个组件本身是否做了它该做的事”；
• 集成测试：测试多个组件之间的交互，可能使用真实的 Memory Store 或工具，但依然 Mock LLM 和外部 API，目标是验证“组件之间能不能配合好”；
• E2E 测试：使用真实的 LLM、工具和存储，测试完整的 Agent 流程，目标是验证“Agent 能不能解决实际问题”。

本文的重点是单元测试——只有把每个组件的单元测试写好了，集成测试和 E2E 测试才能事半功倍。

2.3 技术栈选型：选对工具，事半功倍

Agent Harness 的单元测试不需要太复杂的技术栈，我们选择 Python 生态中最成熟、最常用的工具：

工具类型	工具名称	选择理由
编程语言	Python 3.10+	大多数 Agent 框架都是 Python 写的，生态最丰富
测试框架	pytest	简单灵活，支持 fixtures、参数化测试、标记等高级特性
Mock 框架	unittest.mock + pytest-mock	Python 内置的 Mock 库，配合 pytest-mock 更易用
HTTP Mock	responses	专门用于 Mock HTTP 请求，适合测试工具调用的 API
向量嵌入 Mock	自定义 MagicMock	不需要真实的嵌入模型，返回固定的向量即可
测试覆盖率	pytest-cov	生成详细的覆盖率报告，帮助我们发现未测试的代码
语义断言（可选）	spaCy + sentence-transformers	用于验证回复的语义相似度，适合复杂的断言

在后面的实战环节，我们会详细讲解这些工具的使用方法。

---

三、核心内容：Agent Harness 单元测试实战演练（从 0 到 1）

这一章是本文的核心，我们将从测试环境搭建开始，逐个讲解 Agent Harness 核心组件的测试策略，并配合完整的代码示例。

3.1 测试环境搭建：搭好“架子”才能写出好测试

3.1.1 项目结构设计

首先，我们需要设计一个合理的项目结构，把源代码和测试代码分开：

agent-harness-demo/
├── src/                          # 源代码目录
│   └── agent_harness/            # Agent Harness 核心包
│       ├── __init__.py
│       ├── agent_core.py         # Agent Core 组件
│       ├── tool_registry.py      # Tool Registry 组件
│       ├── tools/                # 工具定义目录
│       │   ├── __init__.py
│       │   ├── weather_tool.py   # 天气查询工具
│       │   └── order_tool.py     # 订单查询工具
│       ├── memory_store.py       # Memory Store 组件
│       ├── planner.py            # Planner 组件
│       ├── executor.py           # Executor 组件
│       └── llm_client.py         # LLM Client 组件
├── tests/                         # 测试代码目录
│   ├── conftest.py               # pytest 全局 fixtures
│   ├── unit/                      # 单元测试目录
│   │   ├── test_agent_core.py
│   │   ├── test_tool_registry.py
│   │   ├── test_memory_store.py
│   │   ├── test_planner.py
│   │   └── test_executor.py
│   └── integration/               # 集成测试目录（本文暂不涉及）
├── pytest.ini                     # pytest 配置文件
├── requirements.txt               # 生产依赖
└── requirements-dev.txt           # 开发/测试依赖

这样的结构清晰明了，符合 Python 项目的最佳实践。

3.1.2 依赖安装

首先，我们创建 requirements.txt（生产依赖）：

# 模拟 LLM 客户端（实际项目中可能是 openai、anthropic 等）
openai>=1.0.0
# 向量数据库（用于 Memory Store）
chromadb>=0.4.0
# 向量嵌入模型
sentence-transformers>=2.2.0
# 工具调用的 HTTP 客户端
requests>=2.31.0

然后创建 requirements-dev.txt（开发/测试依赖）：

pytest>=7.4.0
pytest-mock>=3.11.0
pytest-cov>=4.1.0
responses>=0.23.0
spacy>=3.6.0

接下来，创建虚拟环境并安装依赖：

# 创建虚拟环境
python -m venv venv
# 激活虚拟环境（Windows）
venv\Scripts\activate
# 激活虚拟环境（Mac/Linux）
source venv/bin/activate
# 安装依赖
pip install -r requirements.txt -r requirements-dev.txt

3.1.3 pytest 配置

创建 pytest.ini 文件，配置 pytest 的行为：

[pytest]
# 测试文件的匹配规则
testpaths = tests
# 测试函数的匹配规则
python_files = test_*.py
python_functions = test_*
# 标记（用于分类测试用例）
markers =
    slow: 标记运行较慢的测试用例
    integration: 标记集成测试用例
# 覆盖率配置
addopts =
    --cov=src/agent_harness
    --cov-report=term-missing
    --cov-report=html:htmlcov
    --strict-markers

这个配置的作用是：

• 指定测试文件在 tests/ 目录下；
• 定义了 slow 和 integration 两个标记（方便我们单独运行或跳过某些测试）；
• 自动生成覆盖率报告（终端显示未覆盖的行，HTML 报告存放在 htmlcov/ 目录）。

3.1.4 全局 Fixtures 定义

Fixtures 是 pytest 的核心特性之一，它可以帮助我们复用测试 setup 和 teardown 代码。我们在 tests/conftest.py 中定义一些全局的 fixtures：

# tests/conftest.py
import pytest
from unittest.mock import MagicMock, patch
from src.agent_harness.llm_client import LLMClient
from src.agent_harness.memory_store import MemoryStore


@pytest.fixture
def mock_llm_client():
    """Mock LLM Client，返回预设的响应"""
    mock = MagicMock(spec=LLMClient)
    # 默认返回空响应，测试用例中可以覆盖
    mock.generate.return_value = '{"response": "默认回复"}'
    return mock


@pytest.fixture
def mock_memory_store():
    """Mock Memory Store，返回预设的记忆"""
    mock = MagicMock(spec=MemoryStore)
    # 默认返回空记忆列表，测试用例中可以覆盖
    mock.search.return_value = []
    return mock


@pytest.fixture
def fixed_embedding():
    """返回固定的向量嵌入，用于测试 Memory Store 的相似度检索"""
    # 简单的 3 维向量，实际项目中可能是 768 维或更高
    return lambda text: [hash(text) % 10, hash(text + "a") % 10, hash(text + "b") % 10]

这些 fixtures 会在后面的测试用例中被反复使用，大大减少重复代码。

3.2 Agent Core 的单元测试策略：锁定决策逻辑

Agent Core 是 Agent 的“大脑”，负责意图识别和决策生成——它是最核心、也是最容易出问题的组件之一。

3.2.1 测试的核心点

我们测试 Agent Core 时，主要关注以下几点：

1. 意图识别的正确性：给定用户输入，是否能正确识别意图（如“查询订单”还是“咨询产品”）；
2. 决策逻辑的正确性：根据意图和上下文，是否能做出正确的决策（是直接回复，还是调用工具？调用哪个工具？传递什么参数？）；
3. LLM 响应解析的正确性：LLM 返回的文本是否能正确解析为决策对象（JSON 格式）；
4. 异常处理的正确性：当输入无效、记忆为空、LLM 返回错误格式时，是否能友好处理；
5. 记忆更新的正确性：对话历史是否能正确写入 Memory Store。

3.2.2 先写一个简单的 Agent Core

为了方便测试，我们先实现一个简化版的 Agent Core：

# src/agent_harness/agent_core.py
import json
from typing import Dict, Any
from .llm_client import LLMClient
from .memory_store import MemoryStore


class AgentCore:
    def __init__(self, llm_client: LLMClient, memory_store: MemoryStore):
        self.llm_client = llm_client
        self.memory_store = memory_store

    def process(self, user_input: str) -> Dict[str, Any]:
        """
        处理用户输入，返回决策结果
        :param user_input: 用户输入的文本
        :return: 决策结果，格式为 {"response": "..."} 或 {"tool_name": "...", "parameters": {...}}
        """
        if not user_input or not user_input.strip():
            return {"response": "请输入有效的问题。"}

        # 1. 检索相关记忆
        related_memories = self.memory_store.search(user_input, top_k=3)

        # 2. 构建 Prompt
        prompt = self._build_prompt(user_input, related_memories)

        # 3. 调用 LLM
        try:
            llm_response = self.llm_client.generate(prompt)
        except Exception as e:
            return {"response": f"抱歉，我暂时无法处理你的请求（错误：{str(e)}）。"}

        # 4. 解析 LLM 响应
        decision = self._parse_llm_response(llm_response)

        # 5. 更新记忆
        self.memory_store.add(f"用户：{user_input}")
        self.memory_store.add(f"Agent：{decision.get('response', '') or f'调用工具：{decision.get('tool_name')}'}")

        return decision

    def _build_prompt(self, user_input: str, related_memories: list) -> str:
        """构建发送给 LLM 的 Prompt"""
        memory_str = "\n".join([f"- {mem}" for mem in related_memories])
        return f"""你是一个智能客服 Agent。以下是相关的对话历史：
{memory_str if memory_str else '无'}

用户当前的问题是：{user_input}

请根据对话历史和用户问题，生成你的回复或决定调用的工具：
- 如果不需要调用工具，请返回 JSON 格式：{{"response": "你的回复"}}
- 如果需要调用工具，请返回 JSON 格式：{{"tool_name": "工具名称", "parameters": {{"参数名": "参数值"}}}}
- 只能返回 JSON，不要包含其他文本！
"""

    def _parse_llm_response(self, llm_response: str) -> Dict[str, Any]:
        """解析 LLM 返回的文本为决策对象"""
        # 清理 LLM 响应中的无关文本（比如 ```json ... ```）
        cleaned_response = llm_response.strip()
        if cleaned_response.startswith("```json"):
            cleaned_response = cleaned_response[7:]
        if cleaned_response.endswith("```"):
            cleaned_response = cleaned_response[:-3]
        cleaned_response = cleaned_response.strip()

        try:
            decision = json.loads(cleaned_response)
            # 验证决策格式
            if "response" in decision:
                return decision
            elif "tool_name" in decision and "parameters" in decision:
                return decision
            else:
                return {"response": "抱歉，我暂时无法理解你的问题。"}
        except json.JSONDecodeError:
            return {"response": "抱歉，我暂时无法处理你的请求。"}

这个 Agent Core 虽然简化，但已经包含了核心逻辑：记忆检索、Prompt 构建、LLM 调用、响应解析、记忆更新。

3.2.3 编写测试用例

现在，我们为这个 Agent Core 编写测试用例，存放在 tests/unit/test_agent_core.py 中：

用例 1：测试空输入的异常处理

def test_process_empty_input(agent_core):
    """测试用户输入为空时的处理"""
    # 执行
    result = agent_core.process("")
    # 断言
    assert "请输入有效的问题" in result["response"]
    # 验证 LLM 没有被调用
    agent_core.llm_client.generate.assert_not_called()

用例 2：测试直接回复的情况

def test_process_direct_response(agent_core, mock_llm_client):
    """测试 Agent 直接回复用户的情况"""
    # 准备：Mock LLM 返回直接回复
    expected_response = "你好！我是智能客服，有什么可以帮你的？"
    mock_llm_client.generate.return_value = json.dumps({"response": expected_response})
    user_input = "你好"

    # 执行
    result = agent_core.process(user_input)

    # 断言
    assert result["response"] == expected_response
    # 验证 LLM 被调用了一次
    mock_llm_client.generate.assert_called_once()
    # 验证 Prompt 中包含用户输入
    call_args = mock_llm_client.generate.call_args[0][0]
    assert user_input in call_args
    # 验证记忆被更新
    agent_core.memory_store.add.assert_any_call(f"用户：{user_input}")
    agent_core.memory_store.add.assert_any_call(f"Agent：{expected_response}")

用例 3：测试工具调用的情况

def test_process_tool_call(agent_core, mock_llm_client, mock_memory_store):
    """测试 Agent 调用工具的情况"""
    # 准备：Mock LLM 返回工具调用决策
    tool_name = "get_order"
    parameters = {"order_id": "ORD123456"}
    mock_llm_client.generate.return_value = json.dumps({
        "tool_name": tool_name,
        "parameters": parameters
    })
    # Mock Memory Store 返回相关记忆
    mock_memory_store.search.return_value = ["用户：我之前订了一个订单", "Agent：好的，请告诉我订单号"]
    user_input = "我的订单号是 ORD123456，帮我查一下"

    # 执行
    result = agent_core.process(user_input)

    # 断言
    assert result["tool_name"] == tool_name
    assert result["parameters"] == parameters
    # 验证记忆被检索
    mock_memory_store.search.assert_called_once_with(user_input, top_k=3)
    # 验证记忆被更新
    agent_core.memory_store.add.assert_any_call(f"用户：{user_input}")

用例 4：测试 LLM 返回无效 JSON 的情况

def test_process_invalid_llm_response(agent_core, mock_llm_client):
    """测试 LLM 返回无效 JSON 时的处理"""
    # 准备：Mock LLM 返回非 JSON 文本
    mock_llm_client.generate.return_value = "这不是一个合法的 JSON 格式"
    user_input = "你好"

    # 执行
    result = agent_core.process(user_input)

    # 断言
    assert "抱歉，我暂时无法处理你的请求" in result["response"]

用例 5：测试 LLM 返回带 Markdown 标记的 JSON

def test_process_llm_response_with_markdown(agent_core, mock_llm_client):
    """测试 LLM 返回带 ```json 标记的 JSON 时的处理"""
    # 准备：Mock LLM 返回带 Markdown 标记的 JSON
    expected_response = "今天天气晴朗，温度 25 度"
    mock_llm_client.generate.return_value = f"""```json
{{"response": "{expected_response}"}}
```"""
    user_input = "今天天气怎么样"

    # 执行
    result = agent_core.process(user_input)

    # 断言
    assert result["response"] == expected_response

用例 6：测试 LLM 调用失败的情况

def test_process_llm_call_failure(agent_core, mock_llm_client):
    """测试 LLM 调用失败时的异常处理"""
    # 准备：Mock LLM 调用抛出异常
    mock_llm_client.generate.side_effect = Exception("API 调用超时")
    user_input = "你好"

    # 执行
    result = agent_core.process(user_input)

    # 断言
    assert "API 调用超时" in result["response"]

3.2.4 运行测试用例

现在，我们可以运行测试用例，看看是否都通过：

pytest tests/unit/test_agent_core.py -v

如果一切正常，你会看到所有测试用例都通过的提示。同时，pytest 会自动生成覆盖率报告，你可以打开 htmlcov/index.html 查看哪些代码被覆盖了，哪些没有。

3.2.5 数学模型：Agent Core 的决策概率

虽然 Agent Core 的决策依赖 LLM，但我们可以用条件概率来描述它的决策逻辑：
$$P(\text{Decision}|\text{Input},\text{Memory})=\text{LLM}(\text{Prompt}(\text{Input},\text{Memory}))$$
在单元测试中，我们通过 Mock LLM 让这个概率分布“退化”为确定值——即给定 Input 和 Memory，LLM 一定会返回我们预设的 Decision。这样，我们就可以用传统的断言来验证决策的正确性了。

3.3 Tool Registry 与 Tool 调用的单元测试策略：让工具调用万无一失

Tool 是 Agent 与外部世界交互的“手脚”，如果工具调用出了问题，Agent 再聪明也没用。我们需要分别测试 Tool Registry 和单个 Tool 的逻辑。

3.3.1 先实现 Tool Registry 和一个示例 Tool

首先，我们实现一个简化版的 Tool Registry：

# src/agent_harness/tool_registry.py
from typing import Dict, Any, Optional, Callable
from pydantic import BaseModel, ValidationError


class ToolParameter(BaseModel):
    """工具参数的基类，用于参数验证"""
    pass


class Tool:
    """工具的基类"""
    name: str
    description: str
    parameters: type[ToolParameter]

    def run(self, params: Dict[str, Any]) -> Any:
        """执行工具，子类必须实现"""
        raise NotImplementedError


class ToolRegistry:
    """工具注册中心"""
    def __init__(self):
        self._tools: Dict[str, Tool] = {}

    def register(self, tool: Tool) -> None:
        """注册工具"""
        if tool.name in self._tools:
            raise ValueError(f"工具 {tool.name} 已存在")
        self._tools[tool.name] = tool

    def get(self, tool_name: str) -> Optional[Tool]:
        """获取工具"""
        return self._tools.get(tool_name)

    def validate_parameters(self, tool_name: str, params: Dict[str, Any]) -> bool:
        """验证工具参数"""
        tool = self.get(tool_name)
        if not tool:
            raise ValueError(f"工具 {tool_name} 不存在")
        try:
            tool.parameters(**params)
            return True
        except ValidationError:
            return False

然后，我们实现一个天气查询工具作为示例：

# src/agent_harness/tools/weather_tool.py
from typing import Dict, Any
import requests
from pydantic import Field
from src.agent_harness.tool_registry import Tool, ToolParameter


class WeatherParameter(ToolParameter):
    """天气查询工具的参数"""
    city: str = Field(..., description="城市名称，如北京、上海")
    date: str = Field(default="今天", description="日期，如今天、明天、2024-01-01")


class WeatherTool(Tool):
    name = "get_weather"
    description = "查询指定城市指定日期的天气"
    parameters = WeatherParameter

    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.example.com/weather"

    def run(self, params: Dict[str, Any]) -> Dict[str, Any]:
        """执行天气查询"""
        # 验证参数（ToolRegistry 已经验证过，但这里再验证一次更安全）
        validated_params = self.parameters(**params)
        # 调用天气 API
        try:
            response = requests.get(
                self.base_url,
                params={
                    "city": validated_params.city,
                    "date": validated_params.date,
                    "api_key": self.api_key
                },
                timeout=10
            )
            response.raise_for_status()
            return response.json()
        except requests.RequestException as e:
            raise RuntimeError(f"天气查询失败：{str(e)}") from e

3.3.2 Tool Registry 的测试策略

Tool Registry 的测试核心点：

1. 工具的注册和获取是否正确；
2. 重复注册工具是否会抛出异常；
3. 参数验证是否生效（正确的参数返回 True，错误的参数返回 False）。

我们编写测试用例 tests/unit/test_tool_registry.py：

import pytest
from src.agent_harness.tool_registry import ToolRegistry, Tool, ToolParameter
from pydantic import Field


# 定义一个测试用的 Tool
class TestParameter(ToolParameter):
    name: str = Field(..., description="姓名")
    age: int = Field(..., gt=0, description="年龄，必须大于0")


class TestTool(Tool):
    name = "test_tool"
    description = "测试工具"
    parameters = TestParameter

    def run(self, params):
        return f"Hello {params['name']}, you are {params['age']} years old"


def test_register_and_get_tool():
    """测试工具的注册和获取"""
    registry = ToolRegistry()
    tool = TestTool()
    # 注册工具
    registry.register(tool)
    # 获取工具
    retrieved_tool = registry.get("test_tool")
    assert retrieved_tool is not None
    assert retrieved_tool.name == "test_tool"


def test_register_duplicate_tool():
    """测试重复注册工具是否抛出异常"""
    registry = ToolRegistry()
    tool = TestTool()
    registry.register(tool)
    # 重复注册应该抛出 ValueError
    with pytest.raises(ValueError, match="工具 test_tool 已存在"):
        registry.register(tool)


def test_validate_parameters_success():
    """测试参数验证成功的情况"""
    registry = ToolRegistry()
    tool = TestTool()
    registry.register(tool)
    # 正确的参数
    valid_params = {"name": "Alice", "age": 25}
    assert registry.validate_parameters("test_tool", valid_params) is True


def test_validate_parameters_missing_field():
    """测试参数缺少必填字段的情况"""
    registry = ToolRegistry()
    tool = TestTool()
    registry.register(tool)
    # 缺少 age 字段
    invalid_params = {"name": "Alice"}
    assert registry.validate_parameters("test_tool", invalid_params) is False


def test_validate_parameters_invalid_type():
    """测试参数类型错误的情况"""
    registry = ToolRegistry()
    tool = TestTool()
    registry.register(tool)
    # age 是字符串而不是整数
    invalid_params = {"name": "Alice", "age": "25"}
    assert registry.validate_parameters("test_tool", invalid_params) is False


def test_validate_parameters_invalid_value():
    """测试参数值不符合约束的情况"""
    registry = ToolRegistry()
    tool = TestTool()
    registry.register(tool)
    # age 小于等于 0
    invalid_params = {"name": "Alice", "age": 0}
    assert registry.validate_parameters("test_tool", invalid_params) is False

3.3.3 单个 Tool 的测试策略

单个 Tool 的测试核心点：

1. 给定正确的参数，是否能返回正确的结果；
2. 给定错误的参数，是否能抛出正确的异常；
3. 外部 API 调用失败时，是否能正确处理异常；
4. 参数验证是否生效。

这里我们需要 Mock 外部的天气 API，避免真实的网络请求。我们使用 responses 库来 Mock HTTP 请求，编写测试用例 tests/unit/tools/test_weather_tool.py：

import pytest
import responses
from src.agent_harness.tools.weather_tool import WeatherTool


@pytest.fixture
def weather_tool():
    """创建 WeatherTool 实例，使用测试用的 API Key"""
    return WeatherTool(api_key="test_api_key")


@responses.activate
def test_weather_tool_run_success(weather_tool):
    """测试天气查询成功的情况"""
    # 准备：Mock 天气 API 返回成功响应
    mock_response = {
        "city": "北京",
        "date": "今天",
        "weather": "晴朗",
        "temperature": 25,
        "humidity": 60
    }
    responses.add(
        responses.GET,
        "https://api.example.com/weather",
        json=mock_response,
        status=200
    )
    # 正确的参数
    params = {"city": "北京", "date": "今天"}

    # 执行
    result = weather_tool.run(params)

    # 断言
    assert result == mock_response
    # 验证 API 被调用了一次，且参数正确
    assert len(responses.calls) == 1
    call = responses.calls[0]
    assert call.request.params["city"] == "北京"
    assert call.request.params["date"] == "今天"
    assert call.request.params["api_key"] == "test_api_key"


@responses.activate
def test_weather_tool_run_api_error(weather_tool):
    """测试天气 API 返回错误的情况"""
    # 准备：Mock 天气 API 返回 500 错误
    responses.add(
        responses.GET,
        "https://api.example.com/weather",
        status=500,
        body="Internal Server Error"
    )
    params = {"city": "北京", "date": "今天"}

    # 执行 & 断言
    with pytest.raises(RuntimeError, match="天气查询失败"):
        weather_tool.run(params)


def test_weather_tool_run_invalid_params(weather_tool):
    """测试参数错误的情况"""
    # 缺少 city 参数
    invalid_params = {"date": "今天"}
    with pytest.raises(Exception):  # Pydantic 会抛出 ValidationError
        weather_tool.run(invalid_params)

3.3.4 算法流程图：Tool 调用的完整流程

为了更清晰地展示 Tool 调用的流程，我们用 Mermaid 画一个流程图：

3.4 Memory Store 的单元测试策略：确保记忆不会“丢失”或“错乱”

Memory Store 是 Agent 的“记忆库”，如果记忆检索错误或存储失败，Agent 就会“失忆”，做出错误的决策。我们重点测试记忆的存储、检索、更新、删除功能，尤其是相似度检索——这是 Agent 记忆中最常用、也最容易出问题的部分。

3.4.1 先实现一个基于向量的 Memory Store

我们实现一个简化版的基于 ChromaDB 的 Memory Store：

# src/agent_harness/memory_store.py
from typing import List, Optional
import chromadb
from chromadb.utils import embedding_functions


class MemoryStore:
    def __init__(self, embedding_fn=None, collection_name: str = "agent_memory"):
        # 使用内存版的 ChromaDB，避免持久化（适合测试）
        self.client = chromadb.Client()
        # 如果没有传入 embedding_fn，使用默认的 Sentence-BERT
        self.embedding_fn = embedding_fn or embedding_functions.SentenceTransformerEmbeddingFunction(
            model_name="all-MiniLM-L6-v2"
        )
        # 创建或获取集合
        self.collection = self.client.get_or_create_collection(
            name=collection_name,
            embedding_function=self.embedding_fn
        )
        # 记忆的 ID 计数器
        self._id_counter = 0

    def add(self, text: str, metadata: Optional[dict] = None) -> str:
        """添加记忆"""
        memory_id = f"memory_{self._id_counter}"
        self._id_counter += 1
        self.collection.add(
            documents=[text],
            metadatas=[metadata or {}],
            ids=[memory_id]
        )
        return memory_id

    def search(self, query: str, top_k: int = 3) -> List[str]:
        """检索相关记忆"""
        if top_k <= 0:
            return []
        results = self.collection.query(
            query_texts=[query],
            n_results=min(top_k, self.collection.count())
        )
        # 返回检索到的记忆文本
        return results["documents"][0] if results["documents"] else []

    def update(self, memory_id: str, text: str, metadata: Optional[dict] = None) -> None:
        """更新记忆"""
        self.collection.update(
            ids=[memory_id],
            documents=[text],
            metadatas=[metadata or {}]
        )

    def delete(self, memory_id: str) -> None:
        """删除记忆"""
        self.collection.delete(ids=[memory_id])

    def clear(self) -> None:
        """清空所有记忆（用于测试）"""
        self.client.delete_collection(self.collection.name)
        self.collection = self.client.create_collection(
            name=self.collection.name,
            embedding_function=self.embedding_fn
        )
        self._id_counter = 0

3.4.2 Memory Store 的测试核心点

1. 记忆的添加和读取：添加的记忆是否能正确检索到；
2. 相似度检索的正确性：给定查询，是否能返回最相关的 top-k 个记忆；
3. 记忆的更新和删除：更新和删除操作是否生效；
4. 边界情况：top_k 为 0、记忆为空、查询无相关记忆等情况的处理。

这里的关键是相似度检索的测试——我们需要使用固定的向量嵌入，避免真实的嵌入模型带来的随机性。我们可以使用之前在 conftest.py 中定义的 fixed_embedding fixture。

3.4.3 编写测试用例

我们编写测试用例 tests/unit/test_memory_store.py：

import pytest
from src.agent_harness.memory_store import MemoryStore


def test_add_and_search_memory(fixed_embedding):
    """测试添加记忆和简单检索"""
    # 创建 Memory Store，使用固定的 embedding 函数
    memory_store = MemoryStore(embedding_fn=fixed_embedding)
    # 添加记忆
    memory_text = "用户：我叫张三"
    memory_store.add(memory_text)
    # 检索记忆
    results = memory_store.search("我叫什么", top_k=1)
    # 断言
    assert len(results) == 1
    assert results[0] == memory_text


def test_similarity_search_order(fixed_embedding):
    """测试相似度检索的排序是否正确"""
    memory_store = MemoryStore(embedding_fn=fixed_embedding)
    # 添加三个记忆，其中记忆1和查询最相关，记忆3最不相关
    memory1 = "用户：我想查询订单 ORD123456 的状态"
    memory2 = "用户：我想订明天从北京到上海的机票"
    memory3 = "用户：今天天气怎么样"
    memory_store.add(memory1)
    memory_store.add(memory2)
    memory_store.add(memory3)
    # 检索
    query = "我的订单 ORD123456 怎么样了"
    results = memory_store.search(query, top_k=3)
    # 断言：memory1 应该排在第一位
    assert results[0] == memory1


def test_update_memory(fixed_embedding):
    """测试更新记忆"""
    memory_store = MemoryStore(embedding_fn=fixed_embedding)
    # 添加记忆
    memory_id = memory_store.add("用户：我叫张三")
    # 更新记忆
    new_text = "用户：我叫李四"
    memory_store.update(memory_id, new_text)
    # 检索
    results = memory_store.search("我叫什么", top_k=1)
    # 断言
    assert results[0] == new_text


def test_delete_memory(fixed_embedding):
    """测试删除记忆"""
    memory_store = MemoryStore(embedding_fn=fixed_embedding)
    # 添加两个记忆
    memory1 = "用户：我叫张三"
    memory2 = "用户：我今年25岁"
    memory_store.add(memory1)
    memory_id2 = memory_store.add(memory2)
    # 删除第二个记忆
    memory_store.delete(memory_id2)
    # 检索
    results = memory_store.search("我", top_k=2)
    # 断言：只有 memory1 被检索到
    assert len(results) == 1
    assert results[0] == memory1


def test_search_top_k_zero(fixed_embedding):
    """测试 top_k 为 0 的情况