大模型 技术蓬勃发展的今天,构建基于检索增强生成 (RAG) 的应用变得越来越普遍。然而,传统的软件测试方法在面对 RAG 应用时常常显得力不从心。由于 LLM 的随机性以及调用真实 LLM 和向量数据库的高昂成本,我们急需一种新的 代码测试 策略。本文将深入探讨如何在不依赖真实 LLM 或向量数据库的情况下,对 RAG 应用进行有效的 代码测试,通过 Mock 技术构建 “纸上谈兵” 的环境,确保代码的可靠性和可维护性,同时降低开发成本。

RAG 应用测试的挑战与传统方法的局限性

RAG (Retrieval-Augmented Generation,检索增强生成) 是一种结合了信息检索和文本生成的技术,它允许 LLM 在生成文本时从外部知识库中检索相关信息,从而提高生成内容的准确性和相关性。然而,RAG 应用的测试面临着以下几个关键挑战:

  1. LLM 的随机性: LLM 的输出具有随机性,即使对于相同的输入,每次生成的文本也可能略有不同。这使得传统的 “断言相等” 的测试方法失效,无法准确判断代码是否按预期工作。例如,对于同一个问题,LLM 可能会生成多个不同的答案,这些答案在语义上可能都是正确的,但字面上却存在差异。

  2. 高昂的测试成本: 调用真实的 LLM 和向量数据库需要消耗大量的计算资源和金钱成本。尤其是在持续集成/持续部署 (CI/CD) 流程中,频繁地调用这些服务会显著增加开发成本。根据一些估算,调用一次大型 LLM 的成本可能高达几美分甚至几美元,这对于大规模的测试来说是一笔不小的开销。

  3. 复杂的依赖关系: RAG 应用通常依赖于多个组件,包括 LLM、向量数据库、数据加载器、节点解析器等。这些组件之间的复杂依赖关系使得搭建一个干净、可重复的测试环境变得非常困难。例如,如果测试环境需要连接到远程 LLM 服务或向量数据库,测试可能会因为网络问题或服务故障而失败,而这与代码本身的逻辑无关。

  4. 缺乏明确的测试标准: 传统的软件测试通常有明确的测试标准和预期结果。然而,对于 RAG 应用来说,评估生成内容的质量是一个主观且复杂的过程。例如,如何判断一个 LLM 生成的答案是否准确、完整、相关?这需要领域专家进行评估,难以自动化。

Mock 技术:RAG 应用测试的利器

Mock (模拟) 是一种在测试中用模拟对象替代真实依赖项的技术。通过 Mock,我们可以创建一个可控、隔离的测试环境,避免对真实服务或资源的依赖。在 RAG 应用的测试中,Mock 技术可以用来模拟 LLM 的行为、向量数据库的查询结果、以及其他外部依赖项。

Mock 的核心思想是创建一个“假”的对象,这个对象具有与真实对象相同的接口和行为,但其内部实现是可控的。在测试过程中,我们可以配置 Mock 对象返回预定义的结果,从而模拟各种场景,并验证代码是否按预期工作。

Mock 技术在 RAG 应用测试中的优势:

  1. 降低测试成本: 通过 Mock 掉真实的 LLM 和向量数据库,可以显著降低测试成本。Mock 对象可以在本地运行,无需连接到远程服务,从而避免了计算资源和网络开销。

  2. 提高测试效率: Mock 对象可以快速返回预定义的结果,无需等待真实服务的响应。这可以大大缩短测试时间,提高开发效率。

  3. 隔离测试环境: Mock 对象可以隔离测试环境,避免对真实服务或资源的依赖。这可以确保测试的可重复性和可靠性,避免因为外部因素而导致测试失败。

  4. 模拟各种场景: Mock 对象可以模拟各种场景,包括正常情况、异常情况、边界情况等。这可以帮助我们发现代码中的潜在问题,提高代码的健壮性。

基于 PyTest 和 Mock 的 RAG 代码测试实践

文章作者使用了 Python 的 unittestpytest 框架进行 代码测试unittest 擅长结构化的、基于类的单元测试,而 pytest 则更加轻量级和灵活,支持各种类型的测试。下面是一个使用 pytestunittest.mock 模拟 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 对象。在测试方法中,我们可以配置 Mock 对象返回预定义的文档,并验证 load_data 方法是否按预期工作。

类似的,我们可以 Mock 掉其他依赖项,例如向量数据库、LLM 等。下面是一个 MockQdrantVectorStoreVectorStoreIndex 的例子:

import pytest
from unittest.mock import patch, MagicMock
from typing import Any

class TestDoclingAgentNotebook:
    # ... (previous tests)

    @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

    @patch('llama_index.core.VectorStoreIndex')
    @patch('llama_index.vector_stores.qdrant.QdrantVectorStore')
    def test_load_vector_store_from_qdrant(
        self,
        mock_qdrant_vector_store: MagicMock,
        mock_vector_store_index: MagicMock
    ) -> None:
        """
        Test that VectorStoreIndex.from_vector_store returns the mock index when called with a mocked QdrantVectorStore.
        """
        vector_store = mock_qdrant_vector_store.return_value
        embed_model: Any = MagicMock()
        mock_index = mock_vector_store_index.from_vector_store.return_value
        index = mock_vector_store_index.from_vector_store(vector_store, embed_model=embed_model)
        assert index == mock_index

    @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'

通过这些测试,我们可以验证 RAG 应用的各个组件是否按预期工作,而无需连接到真实的向量数据库或 LLM

Agent 测试:模拟复杂交互与流式响应

Agent 是一个能够自主决策并执行任务的智能体。在 RAG 应用中,Agent 可以根据用户的查询,选择合适的工具 (例如查询引擎) 来获取信息,并生成最终的答案。测试 Agent 的关键在于模拟其决策过程和交互行为。

文章作者展示了如何测试 FunctionAgent,以及如何模拟 Agent 的流式响应。由于 Python 的 unittest 框架不支持异步测试函数,因此异步测试需要放在测试类之外,并使用 @pytest.mark.asyncio 装饰器。

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'

在这个测试中,我们 Mock 掉了 QueryEngineToolFunctionAgent,并配置 FunctionAgentrun 方法返回一个预定义的响应。这使得我们可以验证 Agent 是否正确地选择了工具,并生成了预期的答案。

对于 Agent 的流式响应,测试更加复杂。我们需要模拟 Agent 生成一系列事件,并将这些事件以流的方式返回。文章作者通过自定义 FakeHandler 类来实现这一目标。

import pytest
from unittest.mock import patch, MagicMock, AsyncMock
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.core.tools import QueryEngineTool, ToolMetadata

@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.
            """
            async def _coro():
                """This function implements awaitability for the handler"""
                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?"
    )

通过这个测试,我们可以验证 Agent 是否正确地生成了流式响应,并且最终返回了预期的结果。

DevOps 与代码质量的保障

DevOps 环境中,代码测试 是一个至关重要的环节。通过 Mock 技术,我们可以实现高效、可靠的 代码测试,确保代码的质量和可维护性。此外,代码测试 还可以帮助我们满足 DevOps 的一些硬性要求,例如测试覆盖率。

文章作者提到,在使用了 SonarQube 之后,必须保证 80% 的测试覆盖率才能通过 DevOps 的质量门禁。通过 Mock 技术,我们可以轻松地提高测试覆盖率,满足 DevOps 的要求。

此外,代码测试 还可以帮助我们发现代码中的一些潜在问题,例如认知复杂性过高的方法。SonarQube 会给出类似 “Refactor this method to reduce its Cognitive Complexity from 26 to the 15 allowed.” 的警告,提示我们重构代码,提高代码的可读性和可维护性。

结语:拥抱 Mock,构建可靠的 RAG 应用

虽然 Mock 技术在 RAG 应用的 代码测试 中扮演着重要的角色,但它并不能完全替代真实的集成测试和端到端测试。Mock 测试主要关注代码的单元逻辑和组件之间的交互,而集成测试和端到端测试则关注整个系统的行为和性能。因此,我们需要将 Mock 测试与其他类型的测试结合起来,构建一个全面的测试体系,才能真正确保 RAG 应用的质量和可靠性。

总而言之,在 大模型 时代,RAG 应用的 代码测试 面临着新的挑战。通过 Mock 技术,我们可以构建高效、可靠的 代码测试 体系,降低测试成本,提高开发效率,并确保代码的质量和可维护性。让我们拥抱 Mock,构建更加可靠、智能的 RAG 应用!

测试并不是为了证明代码是完美的,而是为了确保代码在出错时能够发出响亮的警报。因此,让别人来为你的代码编写测试用例总是更好的选择,否则,你可能会倾向于编写只会通过的测试,而不是那些能够发现问题的测试。

发表回复

您的邮箱地址不会被公开。 必填项已用 * 标注