学习第10天：日志、测试、配置与容器化部署
·
学习第10天：日志、测试、配置与容器化部署
贯穿项目：Enterprise AI Agent Platform — 本章为平台建立工程化基础：日志系统、测试体系、配置管理与 Docker 部署
1. 学习目标
完成本章学习后，你将能够：
- 使用 structlog 构建结构化日志系统，替代 Java 的 SLF4J + Logback
- 使用 pytest 编写单元测试、集成测试、异步测试和参数化测试
- 使用 pydantic-settings 实现类型安全的配置管理
- 编写 Dockerfile 和 docker-compose.yml，完成多容器编排
- 建立 CI/CD 流程（GitHub Actions）
- 理解 Python 测试与 Java 测试的核心差异
| 维度 | Java 经验 | Python 对应 |
|---|---|---|
| 日志框架 | SLF4J + Logback | structlog |
| 测试框架 | JUnit 5 + Mockito | pytest + pytest-asyncio |
| 配置管理 | Spring Boot application.yml | pydantic-settings |
| 容器化 | Docker + docker-compose | 完全相同 |
| CI/CD | Jenkins / GitHub Actions | GitHub Actions（完全相同） |
| 覆盖率 | JaCoCo | pytest-cov / coverage.py |
2. 知识体系图
3. 核心知识
3.1 structlog — Python 的 SLF4J + MDC
structlog 是 Python 生态中最先进的结构化日志库。它不只是打印日志行，而是记录结构化事件。
Java 心理映射：
Java SLF4J:
logger.info("User {} created agent {}", userId, agentId);
Java MDC:
MDC.put("requestId", requestId);
logger.info("Processing request");
Python structlog:
logger = structlog.get_logger()
logger.info("user_created_agent", user_id=userId, agent_id=agentId)
Python structlog.bind():
logger = logger.bind(request_id=request_id)
logger.info("processing_request") # 自动包含 request_id
3.2 pytest — Python 的 JUnit 5
pytest 是 Python 的测试框架之王，比 JUnit 更简洁灵活：
- Fixture 替代
@BeforeEach/@AfterEach - 参数化测试
@pytest.mark.parametrize替代@ParameterizedTest - Mock 可用
unittest.mock或pytest-mock - 协程测试 用
pytest-asyncio
3.3 pydantic-settings — 类型安全的配置
相当于 Spring Boot 的 @ConfigurationProperties + application.yml，但全部运行时验证：
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
redis_url: str = "redis://localhost:6379"
debug: bool = False
model_config = {"env_file": ".env"}
settings = Settings() # 自动从环境变量和 .env 文件加载
4. 详细讲解
4.1 structlog 完整配置
# app/core/logging.py
import structlog
import logging
from structlog.typing import Processor
from app.core.config import settings
def setup_logging() -> None:
"""初始化结构化日志系统"""
# 共享的 processors
shared_processors: list[Processor] = [
structlog.contextvars.merge_contextvars,
structlog.stdlib.add_log_level,
structlog.stdlib.add_logger_name,
structlog.processors.TimeStamper(fmt="iso"),
structlog.stdlib.PositionalArgumentsFormatter(),
structlog.processors.StackInfoRenderer(),
structlog.processors.format_exc_info,
structlog.processors.UnicodeDecoder(),
]
if settings.LOG_FORMAT == "json":
# 生产环境：JSON 输出，方便 ELK/Loki 采集
renderer = structlog.processors.JSONRenderer()
else:
# 开发环境：彩色控制台输出
renderer = structlog.dev.ConsoleRenderer(colors=True)
structlog.configure(
processors=shared_processors + [
structlog.stdlib.ProcessorFormatter.wrap_for_formatter,
],
context_class=dict,
logger_factory=structlog.stdlib.LoggerFactory(),
wrapper_class=structlog.stdlib.BoundLogger,
cache_logger_on_first_use=True,
)
# 配置标准库 logging 的 handler
formatter = structlog.stdlib.ProcessorFormatter(
foreign_pre_chain=shared_processors,
processors=[
structlog.stdlib.ProcessorFormatter.remove_processors_meta,
renderer,
],
)
handler = logging.StreamHandler()
handler.setFormatter(formatter)
root_logger = logging.getLogger()
root_logger.addHandler(handler)
root_logger.setLevel(settings.LOG_LEVEL.upper())
# 降低第三方库日志噪音
logging.getLogger("httpx").setLevel(logging.WARNING)
logging.getLogger("httpcore").setLevel(logging.WARNING)
logging.getLogger("sqlalchemy.engine").setLevel(logging.WARNING)
使用方式：
import structlog
logger = structlog.get_logger(__name__)
# 标准日志
logger.info("agent_created", agent_id=42, name="AssistantBot")
# 带上下文的日志
log = logger.bind(request_id="req-abc-123", user_id=100)
log.info("request_started", path="/api/v1/chat")
log.info("llm_call_completed", model="gpt-4", tokens=1500, latency_ms=2300)
log.warning("rate_limit_approaching", remaining_requests=5)
# 异常日志
try:
result = 1 / 0
except ZeroDivisionError:
logger.exception("calculation_failed", operation="division")
开发环境输出（彩色）：
2026-07-23T14:30:00.123Z [info ] agent_created agent_id=42 name=AssistantBot生产环境输出（JSON）：
{"timestamp": "2026-07-23T14:30:00.123Z", "level": "info", "event": "agent_created", "agent_id": 42, "name": "AssistantBot", "logger": "app.services.agent_service"}
4.2 pytest 核心概念
Fixture — 比 JUnit 的 @BeforeEach 更强大
# tests/conftest.py
import pytest
import pytest_asyncio
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from httpx import AsyncClient, ASGITransport
from app.main import app
# 作用域级别：function（默认）、class、module、package、session
@pytest.fixture(scope="session")
def test_settings():
"""测试环境配置（整个测试会话共享）"""
from app.core.config import Settings
return Settings(
database_url="sqlite+aiosqlite:///./test.db",
debug=True,
redis_url="redis://localhost:6379/1",
)
@pytest_asyncio.fixture(scope="function")
async def db_session():
"""每个测试函数独立的数据库会话"""
engine = create_async_engine("sqlite+aiosqlite:///./test.db")
async with engine.begin() as conn:
from app.models.base import Base
await conn.run_sync(Base.metadata.create_all)
async_session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
async with async_session() as session:
yield session
await session.rollback()
await engine.dispose()
@pytest_asyncio.fixture
async def client():
"""HTTP 测试客户端（不需要启动真实服务器）"""
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
Java 对比：Fixture 比 JUnit 的
@BeforeEach/@AfterEach强大得多——它支持：
- 作用域控制（function/class/module/session）
- 依赖注入（Fixture 可以依赖其他 Fixture）
- 自动 teardown（通过 yield）
- 参数化（Fixture 本身也可以参数化）
测试示例
# tests/test_agent_service.py
import pytest
from app.services.agent_service import AgentService
from app.repository.agent_repository import AgentRepository
@pytest.mark.asyncio
async def test_create_agent(db_session):
"""测试创建 Agent"""
repo = AgentRepository(db_session)
service = AgentService(repo, cache=None)
agent = await service.create_agent({
"name": "TestBot",
"model": "gpt-4",
"system_prompt": "You are a helpful assistant",
})
assert agent.id is not None
assert agent.name == "TestBot"
assert agent.model == "gpt-4"
@pytest.mark.asyncio
async def test_get_agent_not_found(db_session):
"""测试查询不存在的 Agent"""
repo = AgentRepository(db_session)
service = AgentService(repo, cache=None)
agent = await service.get_agent(99999)
assert agent is None
参数化测试
@pytest.mark.parametrize("model,expected_min_tokens", [
("gpt-4", 8000),
("gpt-4-turbo", 128000),
("gpt-3.5-turbo", 4000),
])
def test_model_context_window(model, expected_min_tokens):
"""参数化测试：验证不同模型的上下文窗口"""
from app.core.model_config import MODEL_CONFIGS
assert MODEL_CONFIGS[model]["max_tokens"] >= expected_min_tokens
Mock 的正确使用
from unittest.mock import Mock, AsyncMock, patch
@pytest.mark.asyncio
async def test_chat_with_mock_llm(db_session):
"""使用 Mock 测试 LLM 调用"""
mock_llm_response = AsyncMock()
mock_llm_response.choices = [
Mock(message=Mock(content="Hello! How can I help?"))
]
with patch("openai.AsyncOpenAI") as mock_client:
mock_client.return_value.chat.completions.create = AsyncMock(
return_value=mock_llm_response
)
# 执行测试...
result = await chat_service.process_message("Hi")
assert "Hello" in result
mock_client.return_value.chat.completions.create.assert_awaited_once()
4.3 pydantic-settings 配置管理
# app/core/config.py
from typing import Literal
from pydantic import Field, model_validator
from pydantic_settings import BaseSettings, SettingsConfigDict
class Settings(BaseSettings):
"""全局应用配置
相当于 Spring Boot 的 @ConfigurationProperties
"""
model_config = SettingsConfigDict(
env_file=".env",
env_file_encoding="utf-8",
extra="ignore", # 忽略未知环境变量
)
# ========== 应用配置 ==========
APP_NAME: str = "Enterprise AI Agent Platform"
APP_VERSION: str = "1.0.0"
DEBUG: bool = False
LOG_LEVEL: Literal["DEBUG", "INFO", "WARNING", "ERROR"] = "INFO"
LOG_FORMAT: Literal["console", "json"] = "console"
# ========== 数据库配置 ==========
DATABASE_URL: str = Field(..., description="数据库连接串")
DB_POOL_SIZE: int = 20
DB_MAX_OVERFLOW: int = 10
# ========== Redis 配置 ==========
REDIS_URL: str = "redis://localhost:6379/0"
# ========== LLM 配置 ==========
OPENAI_API_KEY: str = Field(..., description="OpenAI API Key")
OPENAI_BASE_URL: str = "https://api.openai.com/v1"
DEFAULT_MODEL: str = "gpt-4"
MAX_TOKENS: int = 4096
LLM_TIMEOUT: int = 60 # 秒
# ========== CORS 配置 ==========
CORS_ORIGINS: list[str] = ["http://localhost:3000"]
# ========== 速率限制 ==========
RATE_LIMIT_ENABLED: bool = True
RATE_LIMIT_REQUESTS: int = 60
RATE_LIMIT_WINDOW: int = 60
@model_validator(mode="after")
def validate_model_combination(self):
"""验证配置组合的合理性"""
if self.DEBUG and self.LOG_FORMAT == "json":
# 自动修正：开发环境用 console 格式
object.__setattr__(self, "LOG_FORMAT", "console")
return self
# 全局单例
settings = Settings()
.env 文件示例：
# .env
APP_NAME=Enterprise AI Agent Platform
DEBUG=true
LOG_LEVEL=DEBUG
DATABASE_URL=postgresql+asyncpg://postgres:postgres@localhost:5432/agent_platform
REDIS_URL=redis://localhost:6379/0
OPENAI_API_KEY=sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx
DEFAULT_MODEL=gpt-4
4.4 Docker 容器化
项目结构（src layout）
agent-platform/
├── src/
│ └── app/
│ ├── __init__.py
│ ├── main.py
│ ├── api/
│ ├── core/
│ ├── models/
│ ├── repository/
│ └── services/
├── tests/
├── alembic/
├── pyproject.toml
├── Dockerfile
├── docker-compose.yml
├── .env
└── .dockerignore
Dockerfile — 多阶段构建
# Dockerfile
# Stage 1: 构建阶段
FROM python:3.12-slim AS builder
WORKDIR /app
# 安装 uv
COPY --from=ghcr.io/astral-sh/uv:latest /uv /usr/local/bin/uv
# 复制依赖文件
COPY pyproject.toml uv.lock ./
# 安装依赖到虚拟环境
RUN uv sync --frozen --no-dev
# Stage 2: 运行阶段
FROM python:3.12-slim AS runtime
WORKDIR /app
# 创建非 root 用户
RUN groupadd -r appuser && useradd -r -g appuser appuser
# 复制虚拟环境
COPY --from=builder /app/.venv /app/.venv
# 复制应用代码
COPY src/ src/
COPY alembic/ alembic/
COPY alembic.ini .
# 设置环境变量
ENV PATH="/app/.venv/bin:$PATH"
ENV PYTHONUNBUFFERED=1
ENV PYTHONDONTWRITEBYTECODE=1
USER appuser
EXPOSE 8000
# 健康检查
HEALTHCHECK --interval=30s --timeout=5s --start-period=10s --retries=3 \
CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')" || exit 1
CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]
# .dockerignore
__pycache__
*.pyc
*.pyo
.pytest_cache
.coverage
htmlcov
.git
.gitignore
.env.local
.venv
venv
*.egg-info
dist
build
.cursor
.workbuddy
docker-compose.yml
# docker-compose.yml
version: "3.9"
services:
# -------- 主应用 --------
api:
build: .
container_name: agent-platform-api
ports:
- "8000:8000"
env_file:
- .env
environment:
- DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/agent_platform
- REDIS_URL=redis://redis:6379/0
depends_on:
db:
condition: service_healthy
redis:
condition: service_healthy
restart: unless-stopped
volumes:
- ./logs:/app/logs
networks:
- agent-network
# -------- PostgreSQL --------
db:
image: pgvector/pgvector:pg16 # 带 pgvector 扩展，后续 RAG 需要
container_name: agent-platform-db
environment:
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
POSTGRES_DB: agent_platform
ports:
- "5432:5432"
volumes:
- pgdata:/var/lib/postgresql/data
healthcheck:
test: ["CMD-SHELL", "pg_isready -U postgres"]
interval: 5s
timeout: 5s
retries: 5
networks:
- agent-network
# -------- Redis --------
redis:
image: redis:7-alpine
container_name: agent-platform-redis
ports:
- "6379:6379"
volumes:
- redisdata:/data
healthcheck:
test: ["CMD", "redis-cli", "ping"]
interval: 5s
timeout: 3s
retries: 5
networks:
- agent-network
# -------- 后台任务 Worker --------
worker:
build: .
container_name: agent-platform-worker
command: >
arq src.app.worker.WorkerSettings
env_file:
- .env
environment:
- DATABASE_URL=postgresql+asyncpg://postgres:postgres@db:5432/agent_platform
- REDIS_URL=redis://redis:6379/0
depends_on:
db:
condition: service_healthy
redis:
condition: service_healthy
restart: unless-stopped
networks:
- agent-network
volumes:
pgdata:
driver: local
redisdata:
driver: local
networks:
agent-network:
driver: bridge
4.5 GitHub Actions CI/CD
# .github/workflows/ci.yml
name: CI/CD Pipeline
on:
push:
branches: [main]
pull_request:
branches: [main]
jobs:
test:
runs-on: ubuntu-latest
services:
postgres:
image: pgvector/pgvector:pg16
env:
POSTGRES_USER: postgres
POSTGRES_PASSWORD: postgres
POSTGRES_DB: agent_platform_test
ports:
- 5432:5432
options: >-
--health-cmd pg_isready
--health-interval 10s
--health-timeout 5s
--health-retries 5
redis:
image: redis:7-alpine
ports:
- 6379:6379
options: >-
--health-cmd "redis-cli ping"
--health-interval 10s
--health-timeout 5s
--health-retries 5
steps:
- uses: actions/checkout@v4
- name: Install uv
uses: astral-sh/setup-uv@v2
- name: Set up Python
run: uv python install 3.12
- name: Install dependencies
run: uv sync --frozen
- name: Run linter
run: uv run ruff check .
- name: Run type checker
run: uv run mypy src/
- name: Run tests
env:
DATABASE_URL: postgresql+asyncpg://postgres:postgres@localhost:5432/agent_platform_test
REDIS_URL: redis://localhost:6379/0
OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
run: |
uv run alembic upgrade head
uv run pytest -v --cov=src --cov-report=term-missing --cov-report=xml
- name: Upload coverage
uses: codecov/codecov-action@v3
with:
file: ./coverage.xml
5. 代码示例
5.1 完整的日志中间件
# app/middleware/logging_middleware.py
import time
import uuid
import structlog
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
logger = structlog.get_logger(__name__)
class LoggingMiddleware(BaseHTTPMiddleware):
"""HTTP 请求日志中间件"""
async def dispatch(self, request: Request, call_next):
request_id = request.headers.get("X-Request-ID", str(uuid.uuid4()))
start_time = time.monotonic()
# 绑定请求上下文
structlog.contextvars.bind_contextvars(
request_id=request_id,
method=request.method,
path=request.url.path,
client_ip=request.client.host if request.client else None,
)
# 记录请求开始
logger.info("request_started")
try:
response = await call_next(request)
response.headers["X-Request-ID"] = request_id
elapsed_ms = (time.monotonic() - start_time) * 1000
logger.info(
"request_completed",
status_code=response.status_code,
elapsed_ms=round(elapsed_ms, 2),
)
return response
except Exception as exc:
elapsed_ms = (time.monotonic() - start_time) * 1000
logger.exception(
"request_failed",
error_type=type(exc).__name__,
elapsed_ms=round(elapsed_ms, 2),
)
raise
finally:
structlog.contextvars.clear_contextvars()
5.2 贯穿项目：完善测试套件
# tests/conftest.py（完整版）
import pytest
import pytest_asyncio
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import (
create_async_engine, AsyncSession, async_sessionmaker,
)
from httpx import AsyncClient, ASGITransport
from unittest.mock import AsyncMock, patch
from app.main import app
from app.models.base import Base
from app.core.config import Settings
TEST_DATABASE_URL = "sqlite+aiosqlite:///:memory:"
@pytest.fixture(scope="session")
def settings():
return Settings(
DATABASE_URL=TEST_DATABASE_URL,
DEBUG=True,
OPENAI_API_KEY="test-key",
REDIS_URL="redis://localhost:6379/1",
)
@pytest_asyncio.fixture
async def engine():
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
await engine.dispose()
@pytest_asyncio.fixture
async def db_session(engine) -> AsyncGenerator[AsyncSession, None]:
session_factory = async_sessionmaker(
engine, class_=AsyncSession, expire_on_commit=False
)
async with session_factory() as session:
async with session.begin():
yield session
await session.rollback()
@pytest_asyncio.fixture
async def client() -> AsyncGenerator[AsyncClient, None]:
transport = ASGITransport(app=app)
async with AsyncClient(transport=transport, base_url="http://test") as ac:
yield ac
@pytest.fixture
def mock_openai():
"""Mock OpenAI API 调用"""
with patch("openai.AsyncOpenAI") as mock:
mock_client = AsyncMock()
mock_response = AsyncMock()
mock_choice = AsyncMock()
mock_choice.message.content = "Mocked AI response"
mock_response.choices = [mock_choice]
mock_client.chat.completions.create = AsyncMock(return_value=mock_response)
mock.return_value = mock_client
yield mock
# tests/test_agent_api.py
import pytest
@pytest.mark.asyncio
async def test_create_agent_api(client, db_session):
"""测试创建 Agent 的完整 HTTP 流程"""
response = await client.post(
"/api/v1/agents",
json={
"name": "TestAgent",
"description": "A test agent",
"model": "gpt-4",
"system_prompt": "You are a test agent",
"temperature": 0.7,
"max_tokens": 2048,
},
)
assert response.status_code == 201
data = response.json()
assert data["name"] == "TestAgent"
assert data["model"] == "gpt-4"
assert "id" in data
@pytest.mark.asyncio
async def test_list_agents_api(client, db_session):
"""测试 Agent 列表 API"""
# 先创建几个 Agent
for i in range(3):
await client.post("/api/v1/agents", json={
"name": f"Agent_{i}",
"model": "gpt-4",
"system_prompt": "Test",
})
response = await client.get("/api/v1/agents")
assert response.status_code == 200
data = response.json()
assert len(data["items"]) == 3
@pytest.mark.asyncio
async def test_get_agent_not_found(client):
"""测试查询不存在的 Agent"""
response = await client.get("/api/v1/agents/99999")
assert response.status_code == 404
@pytest.mark.parametrize("invalid_data,expected_field", [
({"name": "", "model": "gpt-4"}, "name"),
({"name": "Test", "model": ""}, "model"),
({}, "name"),
])
@pytest.mark.asyncio
async def test_create_agent_validation(
client, invalid_data, expected_field
):
"""参数化测试：验证字段校验"""
response = await client.post("/api/v1/agents", json=invalid_data)
assert response.status_code == 422
errors = response.json()["detail"]
field_names = [e["loc"][-1] for e in errors]
assert expected_field in field_names
5.3 运行测试的完整命令
# 运行所有测试
pytest -v
# 只运行集成测试
pytest -v -m "integration"
# 运行指定文件的测试
pytest tests/test_agent_api.py -v
# 带覆盖率报告
pytest --cov=src --cov-report=html --cov-report=term
# 失败即停
pytest -x
# 只运行上次失败的测试
pytest --lf
# 先运行上次失败的，再运行全部
pytest --ff
# 并行运行（需要 pytest-xdist）
pytest -n auto
pyproject.toml 中的 pytest 配置：
[tool.pytest.ini_options]
minversion = "8.0"
asyncio_mode = "auto"
testpaths = ["tests"]
pythonpath = ["src"]
addopts = [
"-v",
"--strict-markers",
"--tb=short",
]
markers = [
"slow: marks tests as slow (deselect with '-m \"not slow\"')",
"integration: marks tests as integration tests",
"unit: marks tests as unit tests",
]
6. 实战案例
案例1：健康检查端点 + 启动就绪探针
# app/api/health.py
from fastapi import APIRouter
from app.core.database import engine
from app.cache.redis_cache import RedisCache
from dataclasses import dataclass
router = APIRouter(tags=["health"])
@dataclass
class HealthStatus:
status: str # "healthy" | "degraded" | "unhealthy"
version: str
checks: dict
@router.get("/health")
async def health_check() -> dict:
"""存活探针：简单返回 200"""
return {"status": "ok"}
@router.get("/health/ready")
async def readiness_check() -> dict:
"""就绪探针：检查所有依赖是否就绪"""
checks = {}
# 检查数据库
try:
async with engine.connect() as conn:
await conn.execute(text("SELECT 1"))
checks["database"] = "ok"
except Exception as e:
checks["database"] = f"error: {str(e)}"
# 检查 Redis
try:
await RedisCache().redis.ping()
checks["redis"] = "ok"
except Exception as e:
checks["redis"] = f"error: {str(e)}"
all_healthy = all(v == "ok" for v in checks.values())
return HealthStatus(
status="healthy" if all_healthy else "degraded",
version=settings.APP_VERSION,
checks=checks,
)
案例2：数据库连接的优雅关闭
# app/main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI
from app.core.database import engine
from app.cache.redis_cache import redis_cache
@asynccontextmanager
async def lifespan(app: FastAPI):
"""应用生命周期管理"""
# 启动时
logger.info("application_starting", version=settings.APP_VERSION)
await redis_cache.connect()
yield
# 关闭时：优雅关闭所有连接
logger.info("application_shutting_down")
await redis_cache.disconnect()
await engine.dispose()
logger.info("application_stopped")
app = FastAPI(
title=settings.APP_NAME,
version=settings.APP_VERSION,
lifespan=lifespan,
)
7. Java 对比
| 维度 | Java (Spring Boot) | Python (FastAPI) |
|---|---|---|
| 日志框架 | SLF4J + Logback | structlog |
| 日志注入 | @Slf4j (Lombok) |
logger = structlog.get_logger(__name__) |
| MDC | MDC.put("key", val) |
logger.bind(key=val) 或 structlog.contextvars.bind_contextvars() |
| 日志格式 | Logback XML 配置 | Python 代码配置 |
| 测试框架 | JUnit 5 + Mockito | pytest + pytest-asyncio + unittest.mock |
| 测试前后置 | @BeforeEach/@AfterEach |
Fixture (scope function/class/module/session) |
| Mock 方法 | Mockito.when().thenReturn() |
Mock(return_value=...) / patch() |
| 参数化测试 | @ParameterizedTest + @CsvSource |
@pytest.mark.parametrize |
| 测试报告 | surefire-report | pytest-html / allure-pytest |
| 配置 | application.yml + @ConfigurationProperties |
.env + pydantic-settings |
| 多环境 | application-{profile}.yml |
多个 .env.{env} + Settings(env_file=".env.prod") |
| Docker | 完全一致 | 完全一致 |
| CI/CD | GitHub Actions / Jenkins | GitHub Actions |
| 依赖注入 | 构造函数注入 | Depends() 函数式注入 |
关键差异：测试哲学
Java：重量级测试工具链。Mockito + AssertJ + TestContainers + WireMock…
Python：轻量级。pytest + unittest.mock 基本够用，需要时加 pytest-asyncio。
Java：每个测试类对应一个被测类。AgentServiceTest → AgentService
Python：测试按模块组织。test_agent_service.py 可以测试多个相关类，一个文件包含所有相关测试。
8. 企业最佳实践
8.1 日志最佳实践
# ✅ 使用结构化字段，不用字符串拼接
logger.info("agent_created", agent_id=42, by_user="admin")
# ❌ logger.info(f"Agent {42} created by admin")
# ✅ 异常使用 logger.exception（自动包含 traceback）
try:
risky_operation()
except Exception:
logger.exception("risky_operation_failed")
# 会自动记录完整的 traceback
# ✅ 敏感信息脱敏
logger.info("user_login", email=redact_email("user@example.com"))
# 输出：email=u***@example.com
# ✅ 记录耗时操作
start = time.monotonic()
result = await heavy_operation()
logger.info("heavy_operation_done", elapsed_ms=(time.monotonic() - start) * 1000)
8.2 测试金字塔
/\
/E2E\ ← 少量端到端测试（用 docker-compose 环境）
/------\
/ 集成 \ ← 中等数量的集成测试（测试 API 端点 + DB）
/----------\
/ 单元测试 \ ← 大量单元测试（纯逻辑，无 IO）
/--------------\
时间分配建议：单元测试 70%，集成测试 20%，E2E 测试 10%
8.3 Dockerfile 最佳实践
- 多阶段构建：分离 build 和 runtime，减小镜像体积
- 非 root 用户：
USER appuser - 健康检查：
HEALTHCHECK指令 - .dockerignore：排除不必要的文件
- 固定基础镜像：
python:3.12-slim而非python:latest - 层缓存优化：先 COPY 依赖文件，再 COPY 代码
9. 常见错误与解决方案
| 错误 | 原因 | 解决 |
|---|---|---|
RuntimeError: Event loop is closed |
异步测试未使用 @pytest.mark.asyncio |
添加 @pytest.mark.asyncio 并确保 asyncio_mode = "auto" |
| Fixture 作用域冲突 | session scope fixture 依赖 function scope fixture | 降低 scope 或提升依赖的 scope |
structlog 日志不输出 |
未配置 handler | 确保调用了 setup_logging() |
| Docker 容器无法连接 localhost | 容器内 localhost 是容器自身 | 使用服务名（如 db、redis）而非 localhost |
pydantic-settings 不加载 .env |
.env 文件路径不正确 | 确保 .env 在项目根目录，或显式指定 env_file |
| 测试数据库未隔离 | 多个测试共享同一数据库 | 使用 :memory: SQLite 或每个测试一个事务 |
| Mock 未生效 | 在 import 之后才 patch | patch 对象必须在 使用它的模块中 patch，而非定义的模块 |
patch 的经典陷阱
# ❌ 错误：在定义处 patch
# tests/test_service.py
from unittest.mock import patch
@patch("app.services.llm_service.AsyncOpenAI") # ❌
async def test_chat():
...
# ✅ 正确：在使用处 patch——因为 service.py 中 `from openai import AsyncOpenAI`，
# AsyncOpenAI 已经在 service 的命名空间中了
@patch("app.services.llm_service.AsyncOpenAI") # ❌ 这是 service 的 import
# 实际上应该：
@patch("openai.AsyncOpenAI") # 根据实际情况选择
async def test_chat():
...
# 规则：patch 的路径是模块 import 后的引用路径，而非原始定义路径。
# 如果你的代码是 `from openai import AsyncOpenAI`，那在 `your_module.AsyncOpenAI`
# 如果你的代码是 `import openai; openai.AsyncOpenAI`，那在 `openai.AsyncOpenAI`
10. 本章总结
本章完成了 Enterprise AI Agent Platform 的 工程化基础：
- structlog 日志系统：结构化日志输出、request_id 追踪、JSON 格式兼容 ELK/Loki
- pytest 测试体系：Fixture 依赖注入、异步测试、Mock 策略、参数化测试
- pydantic-settings 配置：类型安全的配置管理，支持 .env 文件和多环境
- Docker 容器化：多阶段构建、docker-compose 编排（API + DB + Redis + Worker）
- CI/CD：GitHub Actions 自动测试流水线
贯穿项目进度：现在平台具备完整的日志追踪、测试覆盖和容器化部署能力。下一章将进入 AI Agent 的核心——LLM SDK 集成。
11. 面试题
11.1 基础题
structlog相比标准logging的优势是什么？什么是结构化日志？- pytest 的 Fixture 是如何工作的？scope 有哪几种？
pydantic-settings是如何加载配置的？优先级顺序是什么？- Docker 多阶段构建的好处是什么？
11.2 进阶题
- 在异步测试中，
pytest-asyncio的asyncio_mode = "auto"做了什么？如果不用这个模式会怎样？ unittest.mock.patch的路径规则是什么？为什么有时 patch 不生效？- 如何在 Docker Compose 中实现服务启动顺序控制？
depends_on的局限性是什么？如何用condition: service_healthy解决？ - 解释 “测试金字塔” 以及为什么不应该写太多 E2E 测试。
11.3 系统设计题
- 设计一个支持多环境的配置管理方案（dev / staging / production）。如何安全地管理生产环境的敏感配置（如 API Key）？
- 一个高并发的 AI Agent 平台每天产生数百万条日志。设计日志收集、存储和查询方案。
12. 练习
练习1：完善日志系统（难度：★★）
"""要求：
1. 为 LLMService 的每次 API 调用添加耗时日志
2. 实现一个日志脱敏处理器：自动脱敏 email、手机号、API Key
3. 添加慢请求告警：超过 2 秒的请求输出 WARNING 级别日志
"""
练习2：编写完整的测试套件（难度：★★★）
"""要求：
1. 为 ConversationService 编写完整测试（创建会话、添加消息、统计）
2. 使用 Mock 模拟 OpenAI API 调用
3. 使用参数化测试覆盖不同角色（user/assistant/system）的消息
4. 确保测试覆盖率 > 80%
"""
练习3：实现 Docker Compose 多环境部署（难度：★★★）
"""要求：
1. 编写 docker-compose.dev.yml（热重载）
2. 编写 docker-compose.prod.yml（生产配置，关闭调试）
3. 使用 Traefik 作为反向代理，配置 SSL
4. 添加 Prometheus + Grafana 监控栈
"""
13. 作业
作业1：为贯穿项目添加完整测试
- 为所有 Repository 编写单元测试（覆盖率 > 90%）
- 为所有 Service 编写单元测试（Mock 外部依赖）
- 为核心 API 端点编写集成测试
- 添加 GitHub Actions CI 配置
作业2：容器化贯穿项目
- 确保
docker-compose up一键启动全部服务 - 添加数据持久化（volumes）
- 实现优雅关闭（graceful shutdown）
- 添加健康检查端点
作业3：技术报告
写一份 500 字的对比报告，分析 pytest 相比 JUnit 5 的 3 个核心优势。要求有实际代码示例支撑。
14. 预习
下一章将学习：AI 开发基础与多 LLM SDK 集成
预习要点：
- OpenAI Python SDK 的异步用法
- Anthropic Claude SDK
- Google Gemini / Vertex AI SDK
- DeepSeek SDK
- 多模型抽象层设计（统一接口）
- Token 计数与管理
思考题：4 家 LLM 厂商的 SDK 各有不同的接口设计。在构建企业级 Agent 平台时，如何抽象出一个统一的 LLMProvider 接口，使得切换模型只需改一行配置？
文档版本：v1.0 | 创建日期：2026-07-23 | 适用 Python 版本：3.12+
Agent 垂直技术社区，欢迎活跃、内容共建。
更多推荐
2
0- 0
已为社区贡献5条内容
所有评论(0)