Skip to content
🔗 分享本题
查看我的学习进度 →

24 模块 Q6 教学图:如何用 pytest + Mock 测试一个 LLM 应用?

🧠 图解记忆:LLM API 调用有两个问题;点击图片可查看原图。

💡 答案要点

为什么需要 Mock?

LLM API 调用有两个问题:

  • 成本:每次测试都调 API,费用高
  • 不确定性:LLM 输出不稳定,同样的输入可能输出不同
生产测试:
调用 OpenAI API → $$$

用 Mock 测试:
Mock LLM 响应 → 免费 + 稳定

pytest + Mock 实战:

展开 Python 代码示例(86 行)
python
import pytest
from unittest.mock import AsyncMock, MagicMock, patch

# 被测试的 LLM 服务类
class LLMService:
    def __init__(self, api_key: str):
        self.client = OpenAI(api_key=api_key)
    
    def chat(self, prompt: str, system_prompt: str = "你是一个助手") -> str:
        response = self.client.chat.completions.create(
            model="gpt-4o",
            messages=[
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": prompt}
            ]
        )
        return response.choices[0].message.content
    
    async def achat(self, prompt: str) -> str:
        response = await self.client.chat.completions.create(
            model="gpt-4o",
            messages=[{"role": "user", "content": prompt}]
        )
        return response.choices[0].message.content


# Mock 响应
@pytest.fixture
def mock_response():
    """Mock OpenAI API 响应"""
    mock_msg = MagicMock()
    mock_msg.choices[0].message.content = "这是 Mock 的 LLM 响应"
    
    mock_resp = MagicMock()
    mock_resp.choices = [mock_msg]
    return mock_resp


class TestLLMService:
    """LLM 服务单元测试"""
    
    def test_chat_success(self, mock_response):
        """测试正常调用"""
        with patch("openai.OpenAI") as MockOpenAI:
            mock_client = MockOpenAI.return_value
            mock_client.chat.completions.create.return_value = mock_response
            
            service = LLMService(api_key="fake-key")
            result = service.chat("你好")
            
            assert result == "这是 Mock 的 LLM 响应"
            mock_client.chat.completions.create.assert_called_once()
    
    def test_chat_with_system_prompt(self, mock_response):
        """测试 system prompt 是否正确传递"""
        with patch("openai.OpenAI") as MockOpenAI:
            mock_client = MockOpenAI.return_value
            mock_client.chat.completions.create.return_value = mock_response
            
            service = LLMService(api_key="fake-key")
            result = service.chat("今天天气如何?", system_prompt="你是一个天气预报员")
            
            # 验证 system prompt 被传递
            call_args = mock_client.chat.completions.create.call_args
            messages = call_args.kwargs["messages"]
            assert messages[0]["role"] == "system"
            assert messages[0]["content"] == "你是一个天气预报员"


class TestLLMServiceAsync:
    """异步 LLM 服务测试"""
    
    @pytest.mark.asyncio
    async def test_async_chat(self):
        """测试异步调用"""
        mock_response = AsyncMock()
        mock_response.choices[0].message.content = "异步响应"
        
        with patch("openai.AsyncOpenAI") as MockAsyncOpenAI:
            mock_client = MockAsyncOpenAI.return_value
            mock_client.chat.completions.create = AsyncMock(return_value=mock_response)
            
            service = LLMService(api_key="fake-key")
            result = await service.achat("你好")
            
            assert result == "异步响应"

respx Mock HTTP 响应(更真实):

python
import respx
from httpx import Response

@respx.mock
def test_with_respx():
    """用 respx Mock HTTP 响应"""
    # 模拟 OpenAI API 的 HTTP 响应
    respx.post("https://api.openai.com/v1/chat/completions").mock(
        return_value=Response(200, json={
            "choices": [{
                "message": {"content": "respx mock 响应"}
            }]
        })
    )
    
    client = OpenAI(api_key="test")
    response = client.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": "测试"}]
    )
    
    assert response.choices[0].message.content == "respx mock 响应"

用 VCRPy 录制真实响应(回归测试):

python
# 第一次运行:录制真实 API 调用
# @pytest.fixture(scope="module")
# def vcr_config():
#     return {
#         "record_mode": "new",  # 首次录制
#         "filter_headers": [("authorization", "REDACTED")],
#     }

# 后续运行:使用录制文件(不调用真实 API)
import vcr

@vcr.use_cassette("fixtures/vcr/test_chat.yaml")
def test_chat_with_cassette():
    """使用录制的 API 响应"""
    # 不会真正调用 API,会从 fixtures/vcr/test_chat.yaml 读取
    service = LLMService(api_key="real-key")
    result = service.chat("你好")
    assert "你好" in result  # 验证响应内容

集成测试(用 testcontainers 跑真实服务):

python
@pytest.fixture(scope="module")
def redis_container():
    """用 testcontainers 启动真实 Redis"""
    import testcontainers.postgres
    container = testcontainers.postgres.PostgresContainer("postgres:15")
    container.start()
    yield container
    container.stop()


def test_with_real_redis(redis_container):
    """集成测试:使用真实 Redis"""
    from langchain.redis import RedisCache
    
    cache = RedisCache.from_connection_string(
        redis_container.get_connection_url()
    )
    
    # 测试缓存逻辑
    cache.set("key", "value")
    assert cache.get("key") == "value"

测试 AI 输出质量(Regression 测试):

python
def test_llm_quality_regression():
    """防止 LLM 输出质量退化"""
    service = LLMService(api_key="fake-key")
    
    # 准备测试用例(输入 + 期望关键词)
    test_cases = [
        {"input": "北京天气", "keywords": ["天气", "温度", "北京"]},
        {"input": "Python是什么", "keywords": ["Python", "编程", "语言"]},
    ]
    
    for case in test_cases:
        with patch("openai.OpenAI") as MockOpenAI:
            # Mock 一个合规的响应
            mock_response = MagicMock()
            mock_response.choices[0].message.content = f"回答中包含 {case['keywords'][0]}"
            MockOpenAI.return_value.chat.completions.create.return_value = mock_response
            
            result = service.chat(case["input"])
            
            # 验证输出包含必要关键词
            for keyword in case["keywords"]:
                assert keyword in result, f"输出缺少关键词: {keyword}"

面试话术:

"LLM 应用测试的核心是'成本控制和输出稳定性'。我用三层测试:单元测试用 Mock(不调真实 API),集成测试用 VCR 录制(首次录一次,后续回放),回归测试用关键词检查(防止输出质量退化)。Mock 要注意不要过度——测的是你的业务逻辑,不是 OpenAI SDK。另外,asyncio 的测试用 @pytest.mark.asyncio,Mock 要用 AsyncMock。"

📚 参考:pytest 官方文档(Mock 与 fixture)