Python AI Agent 上线前必读：测试、可观测性与零成本方案

你的 AI Agent 在你自己的笔记本上运行正常。开发过程中你跑了上百次，给出的答案都合理。现在有人想把它部署到生产环境。

这就是所有 AI Agent 教程从来没人写的那部分：当凌晨 2 点模型返回了意想不到的结果时，当你的可观测性仪表盘每月要花 400 美元时，当你的测试套件全部通过但 Agent 仍然做了错误的事情时。

这是我刚开始做 Python AI Agent 部署时最希望有一份的检查清单。

生产环境中容易出问题的三个地方（以及如何在上线前测试它们）

AI Agent 的失败模式在开发阶段是不会出现的：

模型改变了输出格式——你的代码期望 JSON，却得到了 markdown

工具调用成功了但返回了垃圾——函数执行了，结果却不是 Agent 期望的

多步推理漂移——第 1 步没问题，第 4 步积累了足够的上下文噪声导致给出错误答案

单元测试捕获不了这些。标准的 mock 也捕获不了这些。AI Agent 所需要的测试模式与 Web 应用所需要的完全不同。

测试 AI Agent：真正有效的方法

核心洞察：你测试的不是代码，而是不确定性下的行为。

这改变了你要做断言的内容：

# 错误：测试精确响应
def test_summarizer():
    result = agent.run("Summarize this document")
    assert result == "The document discusses..."  # 脆弱——模型一更新就挂

# 正确：测试行为属性
def test_summarizer():
    result = agent.run("Summarize this document")
    assert len(result) < len(ORIGINAL_DOCUMENT)  # 压缩发生了
    assert all(key_term in result for key_term in REQUIRED_TERMS)  # 覆盖率检查
    assert not contains_pii(result)  # 安全性属性

模式 1：属性断言优于精确字符串匹配

对 LLM 输出使用 pytest 属性断言，而不是快照测试：

import pytest

def contains_valid_json(text: str) -> bool:
    import json
    try:
        json.loads(text)
        return True
    except json.JSONDecodeError:
        return False

def test_structured_output_agent():
    """Agent 必须返回可解析的 JSON，且包含必需的键。"""
    result = structured_agent.run("Extract entities from: 'Alice called Bob on Tuesday'")

    assert contains_valid_json(result), "Output must be valid JSON"
    parsed = json.loads(result)
    assert "people" in parsed, "Must identify people"
    assert "Alice" in parsed["people"] and "Bob" in parsed["people"]
    assert "date_references" in parsed, "Must identify temporal references"

模式 2：Mock LLM 调用，而不是 Agent

大多数开发者犯的错误：在错误的层次做 mock。

# 错误：mock 得太深，以至于什么都没测到
with patch("my_agent.llm_client") as mock_llm:
    mock_llm.return_value = "mocked response"
    result = agent.run("Do something")
    assert result == "mocked response"  # 你什么都没测到

# 正确：mock API 调用，用受控输入测试 Agent 的行为
from unittest.mock import patch
import json

CONTROLLED_LLM_RESPONSE = json.dumps({
    "action": "search",
    "query": "Python async patterns",
    "confidence": 0.92
})

def test_agent_routes_to_search_tool():
    """Agent 收到信息请求时应调用搜索工具。"""
    with patch("anthropic.Anthropic.messages.create") as mock_create:
        mock_create.return_value = make_mock_response(CONTROLLED_LLM_RESPONSE)

        result = agent.run("What are the best async patterns for Python?")

        # 验证路由，而不是内容
        assert mock_create.called
        assert agent.last_tool_used == "search"
        assert "Python async" in agent.last_search_query

make_mock_response 这个辅助函数（以及它的 fixture）正是值得打包成 starter kit 的东西——我在三个不同项目里重写了三遍之后，把它们都放进了 starter kit。

模式 3：测试工具调用序列，而不仅仅是最终输出

多步 Agent 需要序列级别的测试：

def test_research_agent_tool_sequence():
    """研究 Agent 必须在总结前先搜索——不能有 hallucinate 出来的总结。"""
    tool_calls = []

    def capture_tool(tool_name, **kwargs):
        tool_calls.append(tool_name)
        return MOCK_TOOL_RESPONSES[tool_name]

    with patch_tool_dispatcher(capture_tool):
        result = research_agent.run("What happened at the MCP Dev Summit?")

    # 序列断言：搜索必须在总结之前
    assert "web_search" in tool_calls, "Agent must search before answering"
    assert tool_calls.index("web_search") < tool_calls.index("summarize"), \
        "Search must happen before summarization"
    assert len(tool_calls) <= 5, "Runaway tool calls indicate reasoning failure"

不签 SaaS 合同也能做可观测性

大多数生产可观测性建议都以「设置 LangSmith」或「接入 Datadog」结尾。两者都要花钱，而且增加外部依赖。以下是你可以用 0 美元跑起来、覆盖生产环境 90% 需求的东西。

你实际上需要观测什么

在任何工具设置之前，先确定你需要回答哪些问题：

不依赖 SaaS 的结构化日志

import logging
import json
import time
from functools import wraps
from typing import Any, Callable

# 单一日志器，JSON 格式便于 grep
logging.basicConfig(
    format='%(message)s',
    level=logging.INFO,
    handlers=[
        logging.StreamHandler(),
        logging.FileHandler('/var/log/agent/agent.jsonl')  # 每个服务一个文件
    ]
)
logger = logging.getLogger("agent")

def log_event(event_type: str, **kwargs):
    """结构化日志事件，用于 Agent 可观测性。"""
    logger.info(json.dumps({
        "ts": time.time(),
        "event": event_type,
        **kwargs
    }))

def trace_tool_call(func: Callable) -> Callable:
    """装饰器：记录每一次工具调用，包括耗时和成功/失败。"""
    @wraps(func)
    def wrapper(*args, **kwargs):
        start = time.perf_counter()
        tool_name = func.__name__

        try:
            result = func(*args, **kwargs)
            duration_ms = (time.perf_counter() - start) * 1000

            log_event("tool_call",
                tool=tool_name,
                status="success",
                duration_ms=round(duration_ms, 2),
                input_preview=str(args[0])[:100] if args else None
            )
            return result

        except Exception as e:
            duration_ms = (time.perf_counter() - start) * 1000
            log_event("tool_call",
                tool=tool_name,
                status="error",
                error=str(e),
                duration_ms=round(duration_ms, 2)
            )
            raise

    return wrapper

# 用法
@trace_tool_call
def web_search(query: str) -> str:
    # 你的搜索实现
    ...

这样你就有了一份可查询的 JSONL 文件，记录了每次工具调用、耗时和失败原因。grep "error" agent.jsonl | jq . 在凌晨 2 点排查生产问题时意外地好用。

def log_llm_call(response, prompt_context: str = ""):
    """从 Anthropic 响应中提取并记录 token 使用量。"""
    usage = response.usage

    log_event("llm_call",
        input_tokens=usage.input_tokens,
        output_tokens=usage.output_tokens,
        total_tokens=usage.input_tokens + usage.output_tokens,
        # Anthropic Sonnet 定价：$3/$15 每 1M 输入/输出
        estimated_cost_usd=round(
            (usage.input_tokens * 0.000003) + (usage.output_tokens * 0.000015),
            6
        ),
        context_preview=prompt_context[:50] if prompt_context else None
    )

每天跑一次 cat agent.jsonl | grep llm_call | jq '.estimated_cost_usd' | awk '{sum+=$1} END {print sum}' 就能知道每日开销，不需要仪表盘。

有了结构化日志，你可以写简单的健康检查脚本：

# 简单健康检查脚本——每 5 分钟作为 cron 运行
import json
import sys
from pathlib import Path

LOG_FILE = Path("/var/log/agent/agent.jsonl")
RECENT_MINUTES = 5

def check_agent_health():
    cutoff = time.time() - (RECENT_MINUTES * 60)
    recent_events = [
        json.loads(line)
        for line in LOG_FILE.read_text().splitlines()
        if json.loads(line).get("ts", 0) > cutoff
    ]

    errors = [e for e in recent_events if e.get("status") == "error"]
    if errors:
        error_rate = len(errors) / max(len(recent_events), 1)
        if error_rate > 0.1:  # 10% 错误率
            print(f"ALERT: {error_rate:.0%} error rate in last {RECENT_MINUTES}m")
            sys.exit(1)

    print(f"OK: {len(recent_events)} events, {len(errors)} errors in last {RECENT_MINUTES}m")

check_agent_health()

部署检查清单

把 AI Agent 部署到生产环境之前：

• [ ] 对所有 LLM 输出使用属性断言（而不是精确字符串匹配）
• [ ] 对多步 Agent 进行工具调用序列测试
• [ ] 对抗性输入测试：空输入、超长输入、非英语输入、注入攻击
• [ ] 重试行为测试：LLM 返回畸形 JSON 时会发生什么？
• [ ] 成本上限测试：Agent 是否会无限循环？
• [ ] 每次 LLM 调用都做结构化日志（token 数、延迟、上下文预览）
• [ ] 每次工具调用都做结构化日志（耗时、成功/失败）
• [ ] 失败时记录输入/输出对（用于事后调试）
• [ ] 健康检查脚本或端点，带错误率检查
• [ ] 每个会话的 max token 预算（防止成本失控）
• [ ] 工具调用超时（防止挂在外接服务上）
• [ ] 优雅降级：工具失败时 Agent 返回什么？
• [ ] 回滚方案：如果模型更新破坏了行为，如何快速回退？

在多个项目中构建了这些模式之后，我把可复用的部分打包了出来：

Pytest for AI Agents Starter Kit — mock fixture、属性断言辅助函数和工具调用序列测试器，作为可插入的 pytest 模块。包括 conftest.py 模式、面向 Anthropic 和 OpenAI 的 make_mock_response()，以及一组可复用的 LLM 输出模式属性检查器（Gumroad 上 $49，一次买断）。

Python Agent Observability Toolkit — 结构化日志设置、token 成本追踪器和健康检查脚本，都是生产就绪的 Python 文件。把 agent_logger.py 拖进你的项目，给你的工具加上 @trace_tool_call 装饰器，10 分钟内就能完成 instrumentation（Gumroad 上 $49，一次买断）。

两者都是一次买断，无持续 SaaS 订阅，无供应商锁定，不向任何地方发送遥测数据。

没人谈到的部分

把 AI Agent 部署到生产环境最难的部分不是 LLM 调用本身。而是周围的基础设施：能真正捕获回归的测试、能告诉你哪里坏了的可观测性，以及能防止意外的成本控制。

大多数 AI Agent 教程跳过这部分，因为它不如模型调用那么激动人心。但这正是 demo 和产品的分水岭。

上面的模式足够让你达到生产就绪的基线。测试工具包处理 fixture，这样你就不用在每个项目里重写。观测性工具包处理结构化日志，这样你就不用从零开始搭。

先构建 Agent。然后在它周围构建安全网。

如果你一直在关注「不付订阅税的测试」系列（pytest fixture → Hypothesis → 异步测试 → AI Agent 的 pytest），这篇文章就是终章。从单元测试到生产可观测性的完整测试栈，不需要任何订阅。

有没有我遗漏的生产 AI Agent 模式？欢迎写在评论区。