大模型时代 RAG Agent 代码测试：无需真实 LLM 或向量数据库的单元测试指南

在大模型驱动的软件开发中，测试至关重要。传统的软件测试侧重于代码的“单元”，即单元测试。然而，对于生成式 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 应用。最后，正如原文所说，测试的目的不是为了证明代码是完美的，而是为了确保当代码出现问题时，能够以响亮的方式发出警报。因此，应该鼓励其他人为你的代码编写测试，以确保测试的客观性和全面性。拥抱测试，构建更可靠的 大模型 应用！

大模型时代 RAG Agent 代码测试：无需真实 LLM 或向量数据库的单元测试指南