在大模型驱动的软件开发中，测试至关重要。传统的软件测试侧重于代码的“单元”，即单元测试。然而，对于生成式 AI 应用，特别是涉及RAG Agent (检索增强生成代理) 的应用，传统的测试方法面临挑战。本文将深入探讨如何针对 RAG Agent 进行代码测试，即使没有真实的LLM（大型语言模型）或向量数据库也能进行有效的单元测试，确保代码质量和可靠性。
引言：大模型时代的测试挑战
传统的软件测试通常是确定性的，例如，断言函数 func(a, b) 的输出是否等于 c。然而，LLM 的生成过程是随机的，即使对于相同的输入，每次生成的文本也可能略有不同。这使得传统的测试方法，例如断言精确的输出，变得不可行。更复杂的是，调用真实的 LLM 和向量数据库都需要成本。如果这些调用包含在 CI/CD (持续集成/持续交付) 流程中，可能会因为远程服务不可用而导致测试失败，但这并不意味着代码本身存在问题。因此，我们需要一种在本地、隔离的环境中进行测试的方法，即无需调用真实服务的 单元测试。
RAG Agent 核心组件与测试策略
RAG Agent 的核心在于检索增强生成。它通常由以下几个关键组件构成：
- 文档加载器 (Document Loader): 负责从各种来源（例如，网页、文档、数据库）加载数据。例如，在上述文章中，使用了
DoclingReader来加载 arXiv 上的论文，这使得RAG agent能够处理具有复杂布局的文档。 - 节点解析器 (Node Parser): 将加载的文档分割成更小的、更易于管理的节点，例如段落或句子。这有助于提高检索的精确性。
- 嵌入模型 (Embedding Model): 将文本节点转换成向量表示，以便进行相似度搜索。文章中使用
OllamaEmbedding模型将文本转换成向量。 - 向量数据库 (Vector Database): 存储文本节点的向量表示，并提供高效的相似度搜索功能。Qdrant 是一个流行的开源向量数据库，文章中使用了 Qdrant 来存储和检索文档向量。
- LLM (大型语言模型): 接收检索到的相关文本节点，并生成最终的答案或回复。文章中使用
Ollama("qwen2.5:14b")作为 LLM。 - 查询引擎 (Query Engine): 协调整个检索和生成过程，将用户查询发送到向量数据库，获取相关文本，然后将文本传递给 LLM 以生成最终结果。
- Agent (代理): 决定何时使用哪个工具（例如，查询引擎）来回答用户的问题。
FunctionAgent允许我们将查询引擎包装成一个工具，并让 Agent 智能地选择何时使用它。
针对这些组件，我们可以采用不同的单元测试策略，核心思想是使用 mocking 和 patching 技术来模拟外部依赖项的行为。
Mocking 和 Patching：构建虚拟测试环境
Mocking 和 patching 是 Python 中用于单元测试的强大工具。它们允许我们创建“假的”或“替代的”对象，这些对象行为类似于真实的对象，但在一个受控的、隔离的环境中。
- Mocking (模拟): 创建一个模拟对象，该对象可以替代真实的对象。我们可以预先定义模拟对象的行为，例如，指定模拟对象在被调用时返回什么值。
- Patching (补丁): 替换代码中的某个部分，例如，一个函数或一个类，用一个模拟对象来替代它。这样，当我们运行测试时，代码将使用模拟对象而不是真实的对象。
使用 mocking 和 patching 的优点在于：
- 隔离性: 我们可以隔离被测试的代码，使其不依赖于外部服务或组件。
- 可控性: 我们可以完全控制模拟对象的行为，模拟各种情况，例如，服务不可用、返回错误数据等。
- 效率: 无需调用真实的服务，可以显著提高测试速度。
- 降低成本: 避免调用真实 LLM 和向量数据库，节省测试成本。
案例分析：使用 PyTest 进行 RAG Agent 单元测试
以下是使用 PyTest 框架针对 RAG Agent 进行单元测试的几个示例，这些示例均来自原文，并进行了更详细的解释和分析。
1. 测试 DoclingReader 的文档加载功能
import pytest
from unittest.mock import patch, MagicMock
class TestDoclingAgentNotebook:
"""
Unit tests for the docling agent notebook logic, with all external dependencies mocked.
"""
@patch('llama_index.readers.docling.DoclingReader')
def test_reader_load_data(self, mock_docling_reader: MagicMock) -> None:
"""
Test that DoclingReader.load_data returns a document with the expected text.
All external dependencies are mocked.
"""
mock_reader = mock_docling_reader.return_value
mock_doc = MagicMock()
mock_doc.text_resource.text = 'Sample text.'
mock_reader.load_data.return_value = [mock_doc]
docs = mock_reader.load_data('dummy_url')
assert docs[0].text_resource.text == 'Sample text.'
在这个测试中，我们使用 @patch('llama_index.readers.docling.DoclingReader') 装饰器将 DoclingReader 类替换为一个模拟对象 mock_docling_reader。然后，我们定义了模拟对象的行为：
mock_reader = mock_docling_reader.return_value创建一个模拟的DoclingReader实例。mock_doc = MagicMock()创建一个模拟的文档对象。mock_doc.text_resource.text = 'Sample text.'设置模拟文档对象的文本内容为 “Sample text.”。mock_reader.load_data.return_value = [mock_doc]设置load_data方法的返回值为一个包含模拟文档对象的列表。
最后，我们调用 load_data 方法，并断言返回的文档对象的文本内容是否与预期一致。
2. 测试向量数据库的初始化
import pytest
from unittest.mock import patch, MagicMock
class TestDoclingAgentNotebook:
# ... (previous test)
@patch('qdrant_client.AsyncQdrantClient')
@patch('qdrant_client.QdrantClient')
@patch('llama_index.vector_stores.qdrant.QdrantVectorStore')
def test_vector_store_index_setup_with_qdrant(
self,
mock_qdrant_vector_store: MagicMock,
mock_qdrant_client: MagicMock,
mock_async_qdrant_client: MagicMock
) -> None:
"""
Test that QdrantVectorStore is initialized with the correct arguments and returns the mock instance.
All Qdrant-related dependencies are mocked.
"""
client = mock_qdrant_client.return_value
aclient = mock_async_qdrant_client.return_value
vector_store = mock_qdrant_vector_store('docling', client=client, aclient=aclient, enable_hybrid=True, fastembed_sparse_model='Qdrant/bm42-all-minilm-l6-v2-attentions')
assert vector_store == mock_qdrant_vector_store.return_value
这个测试模拟了 Qdrant 向量数据库的初始化过程。我们使用 @patch 装饰器将 QdrantClient, AsyncQdrantClient 和 QdrantVectorStore 类替换为模拟对象。然后，我们断言 QdrantVectorStore 类是否使用正确的参数进行初始化，并且返回的是模拟的实例。
3. 测试查询引擎
import pytest
from unittest.mock import patch, MagicMock
class TestDoclingAgentNotebook:
# ... (previous tests)
@patch('llama_index.core.VectorStoreIndex')
def test_query_engine(self, mock_vector_store_index: MagicMock) -> None:
"""
Test that the query engine returns the expected mocked response string.
"""
index = mock_vector_store_index.return_value
index.as_query_engine.return_value = MagicMock()
query_engine = index.as_query_engine(sparse_top_k=10, similarity_top_k=6, llm=MagicMock())
query_engine.query.return_value = MagicMock(source_nodes=[], __str__=lambda self: 'Query response')
response = query_engine.query('Test query')
assert str(response) == 'Query response'
这个测试模拟了查询引擎的行为。我们使用 @patch 装饰器将 VectorStoreIndex 类替换为模拟对象。然后，我们定义了模拟对象的行为：
index.as_query_engine.return_value = MagicMock()设置as_query_engine方法的返回值为一个模拟的查询引擎对象。query_engine.query.return_value = MagicMock(source_nodes=[], __str__=lambda self: 'Query response')设置query方法的返回值为一个模拟的响应对象，该对象的字符串表示为 “Query response”。
最后，我们调用 query 方法，并断言返回的响应对象的字符串表示是否与预期一致。
4. 测试 Agent 的执行
import pytest
from unittest.mock import patch, MagicMock, AsyncMock
@pytest.mark.asyncio
@patch('llama_index.core.tools.QueryEngineTool')
@patch('llama_index.core.agent.workflow.FunctionAgent')
async def test_agent_with_query_engine_tool_async(mock_function_agent: MagicMock, mock_query_engine_tool: MagicMock) -> None:
"""
Async test for FunctionAgent.run with a mocked QueryEngineTool.
Ensures the agent returns the expected response when awaited.
"""
query_engine_tool = mock_query_engine_tool.return_value
agent = mock_function_agent(tools=[query_engine_tool], llm=MagicMock())
mock_function_agent.return_value.run = AsyncMock(return_value='Agent response')
agent_response = await agent.run('What is docling?')
assert agent_response == 'Agent response'
这个测试模拟了 Agent 的执行过程。我们使用 @patch 装饰器将 QueryEngineTool 和 FunctionAgent 类替换为模拟对象。然后，我们定义了模拟对象的行为：
mock_function_agent.return_value.run = AsyncMock(return_value='Agent response')设置run方法的返回值为一个异步模拟对象，该对象的返回值为 “Agent response”。
最后，我们调用 run 方法，并断言返回的响应是否与预期一致。
5. 测试 Agent 的流式响应
这是原文中最复杂的一个测试，因为它需要模拟 Agent 的流式响应行为。为了理解这个测试，我们需要深入了解 LlamaIndex 中流式处理的工作原理。简单来说，Agent 的 run 方法返回一个 handler 对象，我们可以通过 async for 循环来迭代该对象，获取中间事件 (AgentStream)。
import pytest
from unittest.mock import patch, MagicMock
@pytest.mark.asyncio
@patch("llama_index.core.tools.QueryEngineTool")
@patch("llama_index.core.agent.workflow.AgentStream")
@patch("llama_index.core.agent.workflow.FunctionAgent")
async def test_agent_streaming_mocked_everything(
mock_function_agent: MagicMock,
mock_agent_stream_cls: MagicMock,
mock_query_engine_tool_cls: MagicMock,
) -> None:
# Step A: Create two **distinct mock events**, each with a unique delta
ev1 = MagicMock()
ev1.delta = "Delta1"
ev2 = MagicMock()
ev2.delta = "Delta2"
# Step B: Minimal replacement for the handler returned by agent.run()
class FakeHandler:
def __init__(self, events, final_value: str):
self._events = events
self._final_value = final_value
async def stream_events(self):
"""
Simulates streaming a sequence of intermediate agent events.
"""
for e in self._events:
yield e
def __await__(self):
"""
Allows `await handler` to work.
Real LlamaIndex handlers (like `StreamingAgentHandler`) support both:
- `async for` for intermediate events
- `await` for the final result
"""
async def _coro():
return self._final_value
return _coro().__await__()
fake_handler = FakeHandler([ev1, ev2], final_value="FINAL_RESULT")
# Step C: Patch FunctionAgent(...) to return a dummy object whose .run(...) returns the handler
dummy_agent_instance = MagicMock(name="DummyAgent")
dummy_agent_instance.run.return_value = fake_handler
mock_function_agent.return_value = dummy_agent_instance
# Patch QueryEngineTool as well, though we never call its logic in this test
mock_query_engine_tool_cls.return_value = MagicMock(name="DummyTool")
# Step D: Import symbols after patching (to pick up the mocks!)
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata
dummy_tool = QueryEngineTool(
query_engine=MagicMock(), # unused
metadata=ToolMetadata(
name="Docling_Knowledge_Base",
description="Use this tool to answer questions about Docling",
),
)
# Step E: Create the agent (returns our patched dummy agent instance)
agent = FunctionAgent(tools=[dummy_tool], llm=MagicMock())
# Step F: Run the agent with a dummy prompt string
handler = agent.run(
"How is docling an improvement over existing document readers?"
)
# Step G: Collect all streamed deltas from the handler
collected = []
async for event in handler.stream_events():
collected.append(event.delta)
# Step H: Await the handler itself for the final result
final = await handler
# Step I: Assertions – critical contract checks
assert collected == ["Delta1", "Delta2"]
assert final == "FINAL_RESULT"
# Step J: Sanity – ensure mocks were called as expected
mock_function_agent.assert_called_once()
dummy_agent_instance.run.assert_called_once_with(
"How is docling an improvement over existing document readers?"
)
在这个测试中，我们首先创建了两个模拟的事件 ev1 和 ev2，每个事件都有一个唯一的 delta 值。然后，我们定义了一个 FakeHandler 类，该类模拟了真实 handler 的行为。FakeHandler 类的 stream_events 方法使用 yield 关键字来模拟流式响应，而 __await__ 方法则允许我们使用 await handler 来获取最终结果。
接下来，我们使用 @patch 装饰器将 FunctionAgent 类替换为模拟对象，并将模拟对象的 run 方法的返回值设置为 fake_handler 实例。最后，我们调用 run 方法，并使用 async for 循环来迭代 handler 对象，收集所有的 delta 值。我们还使用 await handler 来获取最终结果，并断言收集到的 delta 值和最终结果是否与预期一致。
结论与总结：拥抱测试，构建可靠的 RAG Agent
本文详细介绍了如何使用 mocking 和 patching 技术对 RAG Agent 进行单元测试，即使没有真实的 LLM 或向量数据库也能实现有效的测试。虽然这些测试可能无法完全模拟真实环境中的行为，但它们可以帮助我们验证代码的逻辑是否正确，以及组件之间的集成是否正确。此外，编写单元测试还可以提高代码的可测试性，这对于安全的重构和 CI/CD 的可靠性至关重要。
在大模型时代，测试变得更加重要。由于 LLM 的生成过程是随机的，传统的测试方法不再适用。我们需要采用新的测试策略，例如使用 mocking 和 patching 技术来模拟外部依赖项的行为。只有通过充分的测试，我们才能构建出可靠、高效的 RAG Agent 应用。最后，正如原文所说，测试的目的不是为了证明代码是完美的，而是为了确保当代码出现问题时，能够以响亮的方式发出警报。因此，应该鼓励其他人为你的代码编写测试，以确保测试的客观性和全面性。拥抱测试，构建更可靠的 大模型 应用！