一、引言

本文以 “智能运维助手” 项目为主线，覆盖开发 → 联调 → 部署 → 运维全生命周期。

前置条件：Python 3.12+、Docker Desktop。pip install -e ".[dev]" 一键安装依赖。

二、项目初始化

技术选型

层面	选择	理由
语言	Python 3.12+	asyncio 原生异步
LLM 接入	OpenAI SDK（兼容协议）	可切 vLLM/Ollama 等本地模型
Web 框架	FastAPI	异步 + 自动 OpenAPI 文档
配置	Pydantic Settings	类型安全的环境变量
容器化	Docker + docker-compose	标准化交付
CI/CD	GitHub Actions	零额外成本
可观测性	structlog + Prometheus	结构化日志 + 指标

目录结构

enterprise-agent/
├── pyproject.toml
├── .env.example
├── Dockerfile
├── docker-compose.yml
├── .github/workflows/ci.yml
├── src/
│   ├── __init__.py
│   ├── main.py
│   ├── config.py
│   ├── agent/
│   │   ├── engine.py
│   │   └── orchestrator.py
│   ├── llm/
│   │   └── client.py
│   ├── tools/
│   │   ├── base.py
│   │   ├── registry.py
│   │   └── builtins/
│   │       └── database.py
│   └── observability/
│       ├── logger.py
│       └── metrics.py
├── tests/
│   ├── conftest.py
│   ├── unit/test_orchestrator.py
│   └── integration/test_agent_flow.py
└── deploy/k8s/deployment.yaml

pyproject.toml

[project]
name = "enterprise-agent"
version = "1.0.0"
requires-python = ">=3.12"
dependencies = [
    "fastapi>=0.115.0",
    "uvicorn[standard]>=0.30.0",
    "openai>=1.50.0",
    "pydantic-settings>=2.5.0",
    "structlog>=24.0.0",
    "prometheus-fastapi-instrumentator>=7.0.0",
    "httpx>=0.27.0",
    "redis>=5.0.0",
    "sqlalchemy[asyncio]>=2.0.0",
    "aiosqlite>=0.20.0",
]

[project.optional-dependencies]
dev = [
    "pytest>=8.0",
    "pytest-asyncio>=0.24.0",
    "pytest-cov>=5.0",
    "respx>=0.21.0",
    "ruff>=0.6.0",
    "mypy>=1.11.0",
]

[tool.pytest.ini_options]
asyncio_mode = "auto"
testpaths = ["tests"]

.env.example

LLM_API_KEY=your-api-key-here
LLM_BASE_URL=https://api.openai.com/v1
LLM_MODEL=gpt-4o-mini
DATABASE_URL=sqlite+aiosqlite:///./agent.db
REDIS_URL=redis://localhost:6379/0

三、开发阶段：核心代码

3.1 config.py —— 配置管理

llm_api_key 默认 ""，生产环境通过 .env 注入，测试环境由 conftest.py 自动设置。

# src/config.py
from __future__ import annotations
from functools import lru_cache
from pydantic_settings import BaseSettings, SettingsConfigDict


class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env", env_file_encoding="utf-8", extra="ignore"
    )
    llm_api_key: str = ""
    llm_base_url: str = "https://api.openai.com/v1"
    llm_model: str = "gpt-4o-mini"
    llm_temperature: float = 0.1
    llm_max_tokens: int = 4096
    llm_timeout_seconds: int = 60
    agent_max_tool_rounds: int = 10
    host: str = "0.0.0.0"
    port: int = 8080
    redis_url: str = "redis://localhost:6379/0"
    database_url: str = "sqlite+aiosqlite:///./agent.db"


@lru_cache
def get_settings() -> Settings:
    return Settings()

3.2 client.py —— LLM 客户端封装

# src/llm/client.py
from __future__ import annotations
from typing import Any, AsyncGenerator, Dict, List, Optional
from openai import AsyncOpenAI
from src.config import get_settings


class LLMClient:
    def __init__(self) -> None:
        settings = get_settings()
        self._client = AsyncOpenAI(
            api_key=settings.llm_api_key or "unset",
            base_url=settings.llm_base_url,
            timeout=float(settings.llm_timeout_seconds),
        )
        self.model = settings.llm_model
        self.temperature = settings.llm_temperature
        self.max_tokens = settings.llm_max_tokens

    async def chat(
        self,
        messages: List[Dict[str, Any]],
        tools: Optional[List[Dict[str, Any]]] = None,
        tool_choice: str = "auto",
    ) -> Dict[str, Any]:
        kwargs: Dict[str, Any] = {
            "model": self.model, "messages": messages,
            "temperature": self.temperature, "max_tokens": self.max_tokens,
        }
        if tools:
            kwargs["tools"] = tools
            kwargs["tool_choice"] = tool_choice
        response = await self._client.chat.completions.create(**kwargs)
        choice = response.choices[0]
        return {
            "choices": [{
                "index": 0,
                "message": {
                    "role": choice.message.role,
                    "content": choice.message.content,
                    "tool_calls": [
                        {"id": tc.id, "function": {"name": tc.function.name, "arguments": tc.function.arguments}}
                        for tc in (choice.message.tool_calls or [])
                    ] if choice.message.tool_calls else None,
                },
                "finish_reason": choice.finish_reason,
            }],
            "usage": {
                "prompt_tokens": response.usage.prompt_tokens if response.usage else 0,
                "completion_tokens": response.usage.completion_tokens if response.usage else 0,
                "total_tokens": response.usage.total_tokens if response.usage else 0,
            },
        }

    async def chat_stream(
        self, messages: List[Dict[str, Any]],
        tools: Optional[List[Dict[str, Any]]] = None,
    ) -> AsyncGenerator[Dict[str, Any], None]:
        kwargs: Dict[str, Any] = {
            "model": self.model, "messages": messages,
            "temperature": self.temperature, "max_tokens": self.max_tokens,
            "stream": True,
        }
        if tools:
            kwargs["tools"] = tools
            kwargs["tool_choice"] = "auto"
        stream = await self._client.chat.completions.create(**kwargs)
        async for chunk in stream:
            delta = chunk.choices[0].delta if chunk.choices else None
            if delta is None:
                continue
            yield {"content": delta.content, "role": delta.role, "finish_reason": chunk.choices[0].finish_reason}

3.3 工具插件化 —— base.py + registry.py + database.py

# src/tools/base.py
from __future__ import annotations
from abc import ABC, abstractmethod
from dataclasses import dataclass
from typing import Any, Dict


@dataclass
class ToolResult:
    success: bool
    data: Any = None
    error: str = ""


class BaseTool(ABC):
    name: str = ""
    description: str = ""
    parameters_schema: Dict[str, Any] = {}

    @abstractmethod
    async def execute(self, **kwargs: Any) -> ToolResult: ...

    def to_openai_function(self) -> Dict[str, Any]:
        return {
            "type": "function",
            "function": {"name": self.name, "description": self.description, "parameters": self.parameters_schema},
        }

# src/tools/registry.py
from __future__ import annotations
from typing import Dict, List, Optional
from src.tools.base import BaseTool


class ToolRegistry:
    """单例模式，提供 reset() 用于测试隔离"""
    _instance: Optional["ToolRegistry"] = None

    def __new__(cls) -> "ToolRegistry":
        if cls._instance is None:
            cls._instance = super().__new__(cls)
            cls._instance._tools: Dict[str, BaseTool] = {}
        return cls._instance

    @classmethod
    def reset(cls) -> None:
        cls._instance = None

    def register(self, tool: BaseTool) -> None:
        if tool.name in self._tools:
            raise ValueError(f"工具 '{tool.name}' 已被注册")
        self._tools[tool.name] = tool

    def get(self, name: str) -> Optional[BaseTool]:
        return self._tools.get(name)

    def to_openai_format(self) -> List[Dict[str, Any]]:
        return [t.to_openai_function() for t in self._tools.values()]

# src/tools/builtins/database.py
from src.tools.base import BaseTool, ToolResult
from sqlalchemy import text
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.orm import sessionmaker
from src.config import get_settings


class DatabaseQueryTool(BaseTool):
    name = "database_query"
    description = "执行只读 SQL 查询，仅支持 SELECT 语句。"
    parameters_schema = {
        "type": "object",
        "properties": {
            "sql": {"type": "string", "description": "只读 SQL 查询语句"},
            "limit": {"type": "integer", "description": "限制返回行数", "default": 50},
        },
        "required": ["sql"],
    }

    def __init__(self) -> None:
        settings = get_settings()
        self._engine = create_async_engine(settings.database_url, echo=False)
        self._session_factory = sessionmaker(self._engine, class_=AsyncSession, expire_on_commit=False)

    async def execute(self, sql: str, limit: int = 50) -> ToolResult:
        sanitized = sql.strip().upper()
        if not sanitized.startswith("SELECT"):
            return ToolResult(success=False, error="仅允许 SELECT 查询")
        if "LIMIT" not in sanitized:
            sql = f"{sql.rstrip(';')} LIMIT {limit}"
        try:
            async with self._session_factory() as session:
                result = await session.execute(text(sql))
                rows = [dict(row._mapping) for row in result.fetchall()]
                return ToolResult(success=True, data={"rows": rows, "count": len(rows)})
        except Exception as e:
            return ToolResult(success=False, error=str(e))

3.4 orchestrator.py —— ReAct 推理循环

核心要点：list(messages) 浅拷贝避免修改入参；自动注入 system prompt；MAX_TOOL_ROUNDS 防死循环。

# src/agent/orchestrator.py
from __future__ import annotations
import json, time
from typing import Any, AsyncGenerator, Dict, List
from src.llm.client import LLMClient
from src.tools.registry import ToolRegistry
from src.observability.logger import get_logger

logger = get_logger(__name__)

SYSTEM_PROMPT = """你是一个智能运维助手。使用工具时遵循规则：
1. 每次只调用必要的工具
2. 工具返回结果后，综合分析再回复用户
3. 简单问题直接回答，不需要调用工具"""


class AgentOrchestrator:
    MAX_TOOL_ROUNDS = 10

    def __init__(self, llm: LLMClient, tools: ToolRegistry) -> None:
        self.llm = llm
        self.tools = tools

    async def run(self, messages: List[Dict[str, Any]]) -> AsyncGenerator[Dict[str, Any], None]:
        working_msgs: List[Dict[str, Any]] = list(messages)
        round_count = 0
        start_time = time.monotonic()

        if not working_msgs or working_msgs[0].get("role") != "system":
            working_msgs.insert(0, {"role": "system", "content": SYSTEM_PROMPT})

        while round_count < self.MAX_TOOL_ROUNDS:
            round_count += 1
            logger.info("agent_round_start", round=round_count)
            yield {"type": "status", "content": f"第 {round_count} 轮推理..."}

            try:
                response = await self.llm.chat(messages=working_msgs, tools=self.tools.to_openai_format())
            except Exception as e:
                logger.error("llm_call_failed", error=str(e))
                yield {"type": "error", "content": f"LLM 调用失败: {e}"}
                return

            choice = response["choices"][0]
            msg = choice["message"]

            if msg.get("tool_calls"):
                tool_calls = msg["tool_calls"]
                logger.info("tool_calls_detected", count=len(tool_calls))
                yield {"type": "tool_call", "calls": [
                    {"name": tc["function"]["name"], "args": tc["function"]["arguments"]}
                    for tc in tool_calls
                ]}

                working_msgs.append({"role": "assistant", "content": msg.get("content"), "tool_calls": tool_calls})

                for tc in tool_calls:
                    tool_name = tc["function"]["name"]
                    try:
                        args = json.loads(tc["function"]["arguments"])
                    except json.JSONDecodeError:
                        args = {}

                    tool = self.tools.get(tool_name)
                    if not tool:
                        tool_result: Dict[str, Any] = {"error": f"未知工具: {tool_name}"}
                    else:
                        try:
                            result = await tool.execute(**args)
                            tool_result = {"success": result.success, "data": result.data, "error": result.error}
                        except Exception as e:
                            tool_result = {"error": str(e)}

                    logger.info("tool_executed", tool=tool_name, success=tool_result.get("success"))
                    yield {"type": "tool_result", "tool": tool_name, "result": tool_result}
                    working_msgs.append({
                        "role": "tool", "tool_call_id": tc["id"],
                        "content": json.dumps(tool_result, ensure_ascii=False),
                    })
            else:
                content = msg.get("content", "")
                working_msgs.append({"role": "assistant", "content": content})
                elapsed = time.monotonic() - start_time
                logger.info("agent_round_complete", rounds=round_count, elapsed_ms=elapsed * 1000)
                yield {"type": "final", "content": content}
                return

        yield {"type": "error", "content": f"超过最大推理轮次 ({self.MAX_TOOL_ROUNDS})"}

3.5 engine.py + main.py —— Agent 引擎与 FastAPI 入口

# src/agent/engine.py
from __future__ import annotations
from typing import Any, AsyncGenerator, Dict, List
from src.llm.client import LLMClient
from src.tools.registry import ToolRegistry
from src.tools.base import BaseTool
from src.agent.orchestrator import AgentOrchestrator


class AgentEngine:
    def __init__(self) -> None:
        self.llm = LLMClient()
        self.tools = ToolRegistry()
        self.orchestrator = AgentOrchestrator(self.llm, self.tools)

    def register_tool(self, tool: BaseTool) -> None:
        self.tools.register(tool)

    async def process(
        self, session_id: str, user_message: str,
        history: List[Dict[str, Any]] | None = None,
    ) -> AsyncGenerator[Dict[str, Any], None]:
        messages: List[Dict[str, Any]] = list(history) if history else []
        messages.append({"role": "user", "content": user_message})
        async for event in self.orchestrator.run(messages):
            yield {**event, "session_id": session_id}

# src/main.py
from __future__ import annotations
import json
from contextlib import asynccontextmanager
from typing import AsyncGenerator
from fastapi import FastAPI, Request
from fastapi.responses import StreamingResponse
from src.agent.engine import AgentEngine
from src.tools.builtins.database import DatabaseQueryTool
from src.observability.logger import get_logger, setup_logging
from src.observability.metrics import setup_metrics

logger = get_logger(__name__)
agent_engine: AgentEngine | None = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    global agent_engine
    setup_logging()
    setup_metrics(app)
    agent_engine = AgentEngine()
    agent_engine.register_tool(DatabaseQueryTool())
    logger.info("agent_engine_started")
    yield
    logger.info("agent_engine_shutdown")
    agent_engine = None


app = FastAPI(title="Enterprise AI Agent", version="1.0.0", lifespan=lifespan)


@app.post("/api/v1/chat/stream")
async def chat_stream(request: Request):
    body = await request.json()
    user_message = body.get("message", "")
    session_id = body.get("session_id", "default")
    history = body.get("history", [])

    async def event_stream() -> AsyncGenerator[str, None]:
        assert agent_engine is not None
        async for event in agent_engine.process(session_id, user_message, history):
            yield f"data: {json.dumps(event, ensure_ascii=False)}\n\n"
        yield "data: [DONE]\n\n"

    return StreamingResponse(
        event_stream(), media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )


@app.get("/health")
async def health():
    return {"status": "healthy"}

四、联调阶段：测试与排查

4.0 conftest.py —— 全局测试配置

自动注入环境变量 + 每次测试前后重置 ToolRegistry 单例。

# tests/conftest.py
import pytest
from src.tools.registry import ToolRegistry


@pytest.fixture(autouse=True)
def set_test_env(monkeypatch):
    monkeypatch.setenv("LLM_API_KEY", "test-api-key-for-mock")
    monkeypatch.setenv("LLM_BASE_URL", "https://api.openai.com/v1")
    monkeypatch.setattr(
        "src.config.Settings.model_config",
        {"env_file": ".env.test.nonexistent", "env_file_encoding": "utf-8", "extra": "ignore"},
    )


@pytest.fixture(autouse=True)
def reset_tool_registry():
    ToolRegistry.reset()
    yield
    ToolRegistry.reset()

4.1 单元测试 —— respx Mock LLM

# tests/unit/test_orchestrator.py
import json
import pytest
import respx
from httpx import Response
from src.agent.orchestrator import AgentOrchestrator
from src.llm.client import LLMClient
from src.tools.registry import ToolRegistry
from src.tools.base import BaseTool, ToolResult


class FakeSearchTool(BaseTool):
    name = "search"
    description = "搜索知识库"
    parameters_schema = {
        "type": "object",
        "properties": {"query": {"type": "string"}},
        "required": ["query"],
    }
    async def execute(self, query: str) -> ToolResult:
        return ToolResult(success=True, data={"results": [f"关于 {query} 的结果"]})


@pytest.fixture
def tools() -> ToolRegistry:
    registry = ToolRegistry()
    registry.register(FakeSearchTool())
    return registry


@pytest.mark.asyncio
async def test_orchestrator_direct_answer(tools: ToolRegistry):
    """简单问候，不调用工具"""
    with respx.mock(base_url="https://api.openai.com/v1") as mock:
        mock.post("/chat/completions").mock(return_value=Response(200, json={
            "choices": [{"index": 0, "message": {"role": "assistant", "content": "你好，有什么可以帮助你的？"}, "finish_reason": "stop"}],
            "usage": {"prompt_tokens": 10, "completion_tokens": 5, "total_tokens": 15},
        }))
        llm = LLMClient()
        orchestrator = AgentOrchestrator(llm, tools)
        events = [e async for e in orchestrator.run([{"role": "user", "content": "你好"}])]
        assert events[-1]["type"] == "final"
        assert "你好" in events[-1]["content"]


@pytest.mark.asyncio
async def test_orchestrator_with_tool_call(tools: ToolRegistry):
    """LLM 调用工具后继续推理"""
    with respx.mock(base_url="https://api.openai.com/v1") as mock:
        mock.post("/chat/completions").mock(side_effect=[
            Response(200, json={
                "choices": [{"index": 0, "message": {"role": "assistant", "content": None, "tool_calls": [
                    {"id": "call_1", "function": {"name": "search", "arguments": json.dumps({"query": "服务器宕机"})}}
                ]}, "finish_reason": "tool_calls"}],
                "usage": {"prompt_tokens": 20, "completion_tokens": 10, "total_tokens": 30},
            }),
            Response(200, json={
                "choices": [{"index": 0, "message": {"role": "assistant", "content": "根据查询结果，建议重启服务器。"}, "finish_reason": "stop"}],
                "usage": {"prompt_tokens": 30, "completion_tokens": 8, "total_tokens": 38},
            }),
        ])
        llm = LLMClient()
        orchestrator = AgentOrchestrator(llm, tools)
        events = [e async for e in orchestrator.run([{"role": "user", "content": "服务器宕机了怎么办？"}])]
        event_types = [e["type"] for e in events]
        assert "tool_call" in event_types and "tool_result" in event_types and "final" in event_types
        assert "重启" in events[-1]["content"]

4.2 集成测试 —— 完整 API 链路

# tests/integration/test_agent_flow.py
import json
import pytest
import respx
from httpx import Response, ASGITransport, AsyncClient
from src.main import app


@pytest.mark.asyncio
async def test_chat_stream_api():
    with respx.mock(base_url="https://api.openai.com/v1") as mock:
        mock.post("/chat/completions").mock(return_value=Response(200, json={
            "choices": [{"index": 0, "message": {"role": "assistant", "content": "系统状态正常。"}, "finish_reason": "stop"}],
            "usage": {"prompt_tokens": 30, "completion_tokens": 15, "total_tokens": 45},
        }))

        transport = ASGITransport(app=app)
        async with AsyncClient(transport=transport, base_url="http://test") as client:
            response = await client.post("/api/v1/chat/stream", json={
                "message": "帮我查一下系统当前状态", "session_id": "test-session-1",
            })
            assert response.status_code == 200
            assert "text/event-stream" in response.headers["content-type"]
            events = [json.loads(line[6:]) for line in response.text.split("\n")
                      if line.startswith("data: ") and line != "data: [DONE]"]
            assert len(events) > 0 and events[-1]["type"] == "final"

4.3 常见问题排查

问题	原因	排查方法
LLM 不调用工具	System Prompt 未注入 / 工具描述模糊	打印 messages[0] 确认
工具调用后无推理	tool_call_id 不匹配	打印 tc["id"] 核对
推理死循环	LLM 反复调用同一工具	MAX_TOOL_ROUNDS 防呆
Settings() 失败	缺少 LLM_API_KEY	复制 .env.example 或测试走 conftest.py

五、部署阶段

5.1 Dockerfile

多阶段构建 + venv，无需 poetry.lock。

FROM python:3.12-slim AS builder
WORKDIR /app
COPY pyproject.toml ./
RUN python -m venv /opt/venv \
    && /opt/venv/bin/pip install --no-cache-dir --upgrade pip \
    && /opt/venv/bin/pip install --no-cache-dir \
       "fastapi>=0.115.0" "uvicorn[standard]>=0.30.0" "openai>=1.50.0" \
       "pydantic-settings>=2.5.0" "structlog>=24.0.0" \
       "prometheus-fastapi-instrumentator>=7.0.0" \
       "redis>=5.0.0" "aiosqlite>=0.20.0" "sqlalchemy[asyncio]>=2.0.0"

FROM python:3.12-slim AS runtime
RUN apt-get update && apt-get install -y --no-install-recommends curl && rm -rf /var/lib/apt/lists/*
RUN groupadd -r agentuser && useradd -r -g agentuser agentuser
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
WORKDIR /app
COPY src/ ./src/
RUN chown -R agentuser:agentuser /app
USER agentuser
EXPOSE 8080
HEALTHCHECK --interval=30s --timeout=3s --retries=3 CMD curl -f http://localhost:8080/health || exit 1
CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8080"]

5.2 docker-compose.yml

services:
  agent:
    build: .
    ports: ["8080:8080"]
    env_file: [.env]
    environment:
      - REDIS_URL=redis://redis:6379/0
      - DATABASE_URL=sqlite+aiosqlite:///./agent.db
    depends_on:
      redis:
        condition: service_healthy
    restart: unless-stopped
  redis:
    image: redis:7-alpine
    ports: ["6379:6379"]
    healthcheck:
      test: ["CMD", "redis-cli", "ping"]
      interval: 10s; timeout: 3s; retries: 3

5.3 GitHub Actions CI/CD

# .github/workflows/ci.yml
name: CI/CD Pipeline
on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]
env:
  REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}

jobs:
  lint-and-test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - uses: actions/setup-python@v5
        with: { python-version: "3.12" }
      - run: pip install -e ".[dev]"
      - run: ruff check src/ tests/
      - run: mypy src/
      - run: pytest tests/unit/ -v --cov=src --cov-report=term-missing
        env: { LLM_API_KEY: "test-ci-key" }

  build-and-push:
    needs: lint-and-test
    if: github.event_name == 'push' && github.ref == 'refs/heads/main'
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write
    steps:
      - uses: actions/checkout@v4
      - uses: docker/login-action@v3
        with:
          registry: ${{ env.REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}
      - uses: docker/build-push-action@v5
        with:
          context: .
          push: true
          tags: |
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:${{ github.sha }}
            ${{ env.REGISTRY }}/${{ env.IMAGE_NAME }}:latest
          cache-from: type=gha
          cache-to: type=gha,mode=max

5.4 Kubernetes Deployment

# deploy/k8s/deployment.yaml
apiVersion: apps/v1
kind: Deployment
metadata:
  name: enterprise-agent
spec:
  replicas: 3
  selector:
    matchLabels: { app: enterprise-agent }
  template:
    metadata:
      labels: { app: enterprise-agent }
    spec:
      containers:
        - name: agent
          image: ghcr.io/your-org/enterprise-agent:latest
          ports: [{ containerPort: 8080 }]
          env:
            - name: LLM_API_KEY
              valueFrom:
                secretKeyRef: { name: agent-secrets, key: llm-api-key }
            - name: REDIS_URL
              value: "redis://redis-service:6379/0"
          resources:
            requests: { cpu: "250m", memory: "256Mi" }
            limits: { cpu: "1000m", memory: "512Mi" }
          livenessProbe: { httpGet: { path: /health, port: 8080 }, initialDelaySeconds: 10, periodSeconds: 15 }
          readinessProbe: { httpGet: { path: /health, port: 8080 }, initialDelaySeconds: 5, periodSeconds: 10 }
---
apiVersion: v1
kind: Service
metadata:
  name: enterprise-agent-svc
spec:
  selector: { app: enterprise-agent }
  ports: [{ port: 80, targetPort: 8080 }]
  type: ClusterIP
---
apiVersion: autoscaling/v2
kind: HorizontalPodAutoscaler
metadata:
  name: enterprise-agent-hpa
spec:
  scaleTargetRef: { apiVersion: apps/v1, kind: Deployment, name: enterprise-agent }
  minReplicas: 2
  maxReplicas: 10
  metrics:
    - type: Resource
      resource: { name: cpu, target: { type: Utilization, averageUtilization: 70 } }

六、生产运维

6.1 结构化日志 + Prometheus

# src/observability/logger.py
from __future__ import annotations
import structlog, logging


def setup_logging() -> None:
    structlog.configure(
        processors=[
            structlog.stdlib.filter_by_level,
            structlog.stdlib.add_log_level,
            structlog.processors.TimeStamper(fmt="iso"),
            structlog.processors.format_exc_info,
            structlog.processors.JSONRenderer(),
        ],
        context_class=dict,
        logger_factory=structlog.stdlib.LoggerFactory(),
        wrapper_class=structlog.stdlib.BoundLogger,
        cache_logger_on_first_use=True,
    )
    logging.getLogger("httpx").setLevel(logging.WARNING)
    logging.getLogger("openai").setLevel(logging.WARNING)


def get_logger(name: str) -> structlog.stdlib.BoundLogger:
    return structlog.get_logger(name)

# src/observability/metrics.py
from prometheus_fastapi_instrumentator import Instrumentator
from prometheus_client import Counter, Histogram, Gauge
from fastapi import FastAPI

tool_call_counter = Counter("agent_tool_calls_total", "工具调用总次数", ["tool_name", "status"])
llm_call_duration = Histogram("agent_llm_call_duration_seconds", "LLM 调用耗时", buckets=[0.1, 0.5, 1.0, 2.0, 5.0, 10.0, 30.0])
active_sessions = Gauge("agent_active_sessions", "当前活跃会话数")


def setup_metrics(app: FastAPI) -> None:
    Instrumentator().instrument(app).expose(app, endpoint="/metrics")

6.2 生产检查清单

• 安全：API Key 通过 K8s Secret 注入，不硬编码；DatabaseQueryTool 强制仅允许 SELECT，敏感操作需二次确认；Agent 输出做敏感词过滤。
• 可靠性：LLM 和工具调用实现指数退避重试 + 熔断降级；捕获 SIGTERM 优雅关闭。
• 性能：HTTP 连接池 + 数据库连接池合理配置。
• 成本：每次 LLM 调用记录 usage.total_tokens，按日聚合。
• 合规：完整审计日志记录用户输入、推理链、工具调用参数。

七、运行指南

cd enterprise-agent

# 安装
python -m venv .venv
.venv\Scripts\activate          # Windows
pip install -e ".[dev]"

# 配置
cp .env.example .env            # 编辑填入真实 LLM_API_KEY

# 测试（无需真实 LLM）
pytest tests/ -v

# 启动
uvicorn src.main:app --reload --port 8080

启动后访问 http://localhost:8080/docs 查看 Swagger 调试界面。