一、引言

本文以 “智能运维助手” 项目为主线，按"基础设施 → 核心引擎 → 测试联调 → 容器化部署 → 运维保障"的递进逻辑，覆盖开发 → 联调 → 部署 → 运维全生命周期，所有代码可直接运行。

前置条件：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

三、基础设施层

在编写 Agent 核心逻辑之前，需要先搭建四个底层组件：配置管理、LLM 客户端、工具系统、可观测性。这些组件被所有上层模块依赖，必须先就位。

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 客户端封装

将 AsyncOpenAI 的对象响应统一转为字典格式，后续编排器和引擎都依赖这个稳定接口。

# 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

BaseTool 定义统一接口，ToolRegistry 用单例模式管理所有工具实例。

# 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 可观测性 —— logger.py + metrics.py

编排器在推理循环中依赖 get_logger，因此日志和指标也属于基础设施，需在核心引擎之前定义。

# 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")

---

基础设施就绪，下面进入 Agent 核心逻辑的实现。三大组件按依赖关系排列：编排器（ReAct 推理引擎）→ 引擎（封装编排器）→ FastAPI（包装引擎为 HTTP 服务）。

四、核心引擎实现

4.1 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})"}

4.2 engine.py —— Agent 引擎封装

将编排器包装为 AgentEngine，提供统一的 process() 入口并附带 session_id。

# 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}

4.3 main.py —— FastAPI SSE 服务入口

引擎就绪后，用 FastAPI 的 lifespan 初始化 Agent 并注册工具，通过 SSE 流式接口对外暴露。

# 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"}

---

核心代码编写完毕，进入测试验证阶段。在真实 LLM 不可用的情况下，用 respx 拦截 HTTP 请求来模拟 LLM 行为。

五、联调测试

5.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()

5.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"]

5.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"

5.3 常见问题排查

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

---

测试通过后，进入部署环节。核心思路：Docker 镜像化 → 本地编排 → CI/CD 自动化 → K8s 生产部署。

六、容器化部署

6.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"]

6.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

6.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

6.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 } }

---

部署上线后，需要持续关注安全、可靠性和成本。以下是不依赖代码的运维要点。

七、生产检查清单

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

---

八、运行操作讲解

下面从零开始，逐步搭建项目、跑测试、启动服务、调试 API、最后 Docker 部署。

8.1 环境准备

python --version      # 应输出 Python 3.12.x 或更高
docker --version      # 应输出 Docker version 24.x 或更高

8.2 创建项目并写入代码

mkdir enterprise-agent
cd enterprise-agent

# 创建目录结构
mkdir src\agent; mkdir src\llm; mkdir src\tools\builtins; mkdir src\observability
mkdir tests\unit; mkdir tests\integration; mkdir deploy\k8s; mkdir .github\workflows

# 创建所有空 __init__.py
New-Item -Path src\__init__.py -ItemType File -Force
New-Item -Path src\agent\__init__.py -ItemType File -Force
New-Item -Path src\llm\__init__.py -ItemType File -Force
New-Item -Path src\tools\__init__.py -ItemType File -Force
New-Item -Path src\tools\builtins\__init__.py -ItemType File -Force
New-Item -Path src\observability\__init__.py -ItemType File -Force
New-Item -Path tests\__init__.py -ItemType File -Force

按以下顺序创建文件并粘贴代码：

顺序	文件	对应章节
1	pyproject.toml	二
2	.env.example	二
3	src/config.py	3.1
4	src/llm/client.py	3.2
5	src/tools/base.py	3.3
6	src/tools/registry.py	3.3
7	src/tools/builtins/database.py	3.3
8	src/observability/logger.py	3.4
9	src/observability/metrics.py	3.4
10	src/agent/orchestrator.py	4.1
11	src/agent/engine.py	4.2
12	src/main.py	4.3
13	tests/conftest.py	5.0
14	tests/unit/test_orchestrator.py	5.1
15	tests/integration/test_agent_flow.py	5.2
16	Dockerfile	6.1
17	docker-compose.yml	6.2
18	.github/workflows/ci.yml	6.3
19	deploy/k8s/deployment.yaml	6.4

8.3 安装依赖

python -m venv .venv
.venv\Scripts\Activate.ps1       # Windows PowerShell
# .venv\Scripts\activate.bat     # Windows CMD
# source .venv/bin/activate      # macOS / Linux
pip install -e ".[dev]"

验证：

pip list | findstr "fastapi openai pytest structlog"

8.4 运行测试（无需真实 LLM）

pytest tests/ -v

输出：

tests/unit/test_orchestrator.py::test_orchestrator_direct_answer PASSED
tests/unit/test_orchestrator.py::test_orchestrator_with_tool_call PASSED
tests/integration/test_agent_flow.py::test_chat_stream_api PASSED

8.5 启动开发服务器

cp .env.example .env            # 编辑填入真实 LLM_API_KEY
uvicorn src.main:app --reload --port 8080

访问 http://localhost:8080/docs，展开 /api/v1/chat/stream → Try it out：

{"message": "帮我查一下数据库中有哪些表", "session_id": "demo-001"}

SSE 流式响应示例：

data: {"type":"status","content":"第 1 轮推理...","session_id":"demo-001"}
data: {"type":"tool_call","calls":[{"name":"database_query","args":"..."}],"session_id":"demo-001"}
data: {"type":"tool_result","tool":"database_query","result":{"success":true,...},"session_id":"demo-001"}
data: {"type":"final","content":"数据库中共有 5 张表：...","session_id":"demo-001"}
data: [DONE]

8.6 Docker 部署

docker compose up --build
curl http://localhost:8080/health     # {"status":"healthy"}
docker compose down                    # 停服

8.7 常见踩坑

问题	解决
ModuleNotFoundError: No module named 'src'	在项目根目录执行 pip install -e ".[dev]"
Settings() 构造报错	.env 不存在 → cp .env.example .env
ImportError: cannot import name 'Instrumentator'	确认 prometheus-fastapi-instrumentator 已安装
Docker 构建时 aiosqlite 编译失败	确保 Dockerfile builder 阶段没有遗漏 aiosqlite
测试报 respx 相关错误	respx>=0.21.0，且 mock URL 与 LLM_BASE_URL 一致

九、总结

本文按"基础设施 → 核心引擎 → 联调测试 → 容器化部署 → 运维保障"的递进逻辑，覆盖了企业 AI Agent 的完整工程链路：

配置管理 → LLM客户端 → 工具系统 → 可观测性    （第三章：基础设施层）
    ↓
ReAct编排器 → Agent引擎 → FastAPI SSE服务      （第四章：核心引擎实现）
    ↓
conftest自动注入 → respx Mock → 单元/集成测试  （第五章：联调测试）
    ↓
Docker镜像 → docker-compose → CI/CD → K8s+HPA  （第六章：容器化部署）
    ↓
安全检查清单                                    （第七章：生产运维）