在 大模型 技术蓬勃发展的今天,构建基于检索增强生成 (RAG) 的应用变得越来越普遍。然而,传统的软件测试方法在面对 RAG 应用时常常显得力不从心。由于 LLM 的随机性以及调用真实 LLM 和向量数据库的高昂成本,我们急需一种新的 代码测试 策略。本文将深入探讨如何在不依赖真实 LLM 或向量数据库的情况下,对 RAG 应用进行有效的 代码测试,通过 Mock 技术构建 “纸上谈兵” 的环境,确保代码的可靠性和可维护性,同时降低开发成本。
RAG 应用测试的挑战与传统方法的局限性
RAG (Retrieval-Augmented Generation,检索增强生成) 是一种结合了信息检索和文本生成的技术,它允许 LLM 在生成文本时从外部知识库中检索相关信息,从而提高生成内容的准确性和相关性。然而,RAG 应用的测试面临着以下几个关键挑战:
-
LLM 的随机性: LLM 的输出具有随机性,即使对于相同的输入,每次生成的文本也可能略有不同。这使得传统的 “断言相等” 的测试方法失效,无法准确判断代码是否按预期工作。例如,对于同一个问题,LLM 可能会生成多个不同的答案,这些答案在语义上可能都是正确的,但字面上却存在差异。
-
高昂的测试成本: 调用真实的 LLM 和向量数据库需要消耗大量的计算资源和金钱成本。尤其是在持续集成/持续部署 (CI/CD) 流程中,频繁地调用这些服务会显著增加开发成本。根据一些估算,调用一次大型 LLM 的成本可能高达几美分甚至几美元,这对于大规模的测试来说是一笔不小的开销。
-
复杂的依赖关系: RAG 应用通常依赖于多个组件,包括 LLM、向量数据库、数据加载器、节点解析器等。这些组件之间的复杂依赖关系使得搭建一个干净、可重复的测试环境变得非常困难。例如,如果测试环境需要连接到远程 LLM 服务或向量数据库,测试可能会因为网络问题或服务故障而失败,而这与代码本身的逻辑无关。
-
缺乏明确的测试标准: 传统的软件测试通常有明确的测试标准和预期结果。然而,对于 RAG 应用来说,评估生成内容的质量是一个主观且复杂的过程。例如,如何判断一个 LLM 生成的答案是否准确、完整、相关?这需要领域专家进行评估,难以自动化。
Mock 技术:RAG 应用测试的利器
Mock (模拟) 是一种在测试中用模拟对象替代真实依赖项的技术。通过 Mock,我们可以创建一个可控、隔离的测试环境,避免对真实服务或资源的依赖。在 RAG 应用的测试中,Mock 技术可以用来模拟 LLM 的行为、向量数据库的查询结果、以及其他外部依赖项。
Mock 的核心思想是创建一个“假”的对象,这个对象具有与真实对象相同的接口和行为,但其内部实现是可控的。在测试过程中,我们可以配置 Mock 对象返回预定义的结果,从而模拟各种场景,并验证代码是否按预期工作。
Mock 技术在 RAG 应用测试中的优势:
-
降低测试成本: 通过 Mock 掉真实的 LLM 和向量数据库,可以显著降低测试成本。Mock 对象可以在本地运行,无需连接到远程服务,从而避免了计算资源和网络开销。
-
提高测试效率: Mock 对象可以快速返回预定义的结果,无需等待真实服务的响应。这可以大大缩短测试时间,提高开发效率。
-
隔离测试环境: Mock 对象可以隔离测试环境,避免对真实服务或资源的依赖。这可以确保测试的可重复性和可靠性,避免因为外部因素而导致测试失败。
-
模拟各种场景: Mock 对象可以模拟各种场景,包括正常情况、异常情况、边界情况等。这可以帮助我们发现代码中的潜在问题,提高代码的健壮性。
基于 PyTest 和 Mock 的 RAG 代码测试实践
文章作者使用了 Python 的 unittest
和 pytest
框架进行 代码测试。unittest
擅长结构化的、基于类的单元测试,而 pytest
则更加轻量级和灵活,支持各种类型的测试。下面是一个使用 pytest
和 unittest.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 等。下面是一个 Mock 掉 QdrantVectorStore 和 VectorStoreIndex 的例子:
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 掉了 QueryEngineTool
和 FunctionAgent
,并配置 FunctionAgent
的 run
方法返回一个预定义的响应。这使得我们可以验证 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 应用!
测试并不是为了证明代码是完美的,而是为了确保代码在出错时能够发出响亮的警报。因此,让别人来为你的代码编写测试用例总是更好的选择,否则,你可能会倾向于编写只会通过的测试,而不是那些能够发现问题的测试。