AI Agent 生产环境部署指南｜附源码

随着大模型能力的快速提升，AI Agent 已经不再只是 Demo 或实验室概念。越来越多企业开始将 Agent 应用于客服、数据分析、运维自动化、知识库问答、销售辅助、代码生成、流程审批等真实业务场景。

但是，从“能跑起来”到“稳定运行在生产环境”，中间有非常大的差距。一个生产级 AI Agent 系统不仅要考虑模型调用，还要关注权限、工具调用安全、上下文管理、日志审计、限流熔断、异常恢复、成本控制、监控告警以及灰度发布等问题。

本文将系统介绍 AI Agent 生产环境部署方案，并提供一套可运行的基础源码示例，帮助你从工程角度搭建一个可扩展、可观测、可维护的 Agent 服务。

一、什么是 AI Agent？

AI Agent 可以理解为具备以下能力的智能应用系统：

1. 理解任务：接收用户输入，分析用户意图；
2. 规划步骤：将复杂任务拆解成多个可执行步骤；
3. 调用工具：根据任务需要调用 API、数据库、搜索引擎、代码执行器等外部工具；
4. 观察结果：读取工具返回结果并判断下一步动作；
5. 生成答案：综合上下文、工具结果和业务规则输出最终结果。

一个典型 Agent 的执行流程如下：

用户输入
   ↓
意图识别 / 任务规划
   ↓
选择工具
   ↓
执行工具调用
   ↓
观察结果
   ↓
继续推理或输出最终答案

与普通 ChatBot 相比，Agent 的核心差异在于：
ChatBot 主要负责对话生成，而 Agent 能够主动决策并执行操作。

二、生产环境部署 AI Agent 的核心挑战

很多团队在本地开发 Agent 时体验很好，但一旦上线就会遇到各种问题。

1. 大模型输出不可控

模型可能出现：

• 胡编乱造；
• 不按指定 JSON 格式输出；
• 错误调用工具；
• 陷入循环推理；
• 输出敏感或违规内容。

因此生产环境必须增加：

• Prompt 约束；
• 输出结构化校验；
• 重试机制；
• 安全过滤；
• 最大推理轮次限制。

2. 工具调用存在安全风险

Agent 一旦可以调用工具，就意味着它可能执行真实操作，例如：

• 查询数据库；
• 发送邮件；
• 操作订单；
• 调用企业内部系统；
• 执行代码或脚本。

如果没有权限控制，可能导致严重事故。

生产环境中，工具调用必须遵循：

• 最小权限原则；
• 工具白名单机制；
• 参数校验；
• 敏感操作二次确认；
• 全量审计日志。

3. 成本不可控

每次调用大模型都会产生 Token 成本。复杂 Agent 可能一次用户请求触发多轮推理、多次工具调用、多次模型请求。

因此需要：

• 限制最大上下文长度；
• 使用缓存；
• 设置用户级 / 租户级限额；
• 对不同任务选择不同模型；
• 监控 Token 消耗。

4. 延迟较高

Agent 的响应链路通常比普通接口更长：

用户请求 → 模型推理 → 工具调用 → 模型总结 → 返回结果

如果工具调用较多，整体耗时会明显增加。

优化方式包括：

• 异步任务处理；
• 流式响应；
• 并行工具调用；
• 缓存中间结果；
• 将复杂任务拆分为后台任务。

5. 可观测性不足

上线后最常见的问题是：

用户说 Agent 答错了，但开发人员不知道它为什么答错。

所以必须记录：

• 用户输入；
• Prompt；
• 模型返回；
• 工具调用参数；
• 工具返回结果；
• Token 消耗；
• 执行耗时；
• 错误日志；
• Trace ID。

三、AI Agent 生产级架构设计

一个推荐的生产级 Agent 架构如下：

┌──────────────────────────┐
│        前端 / 客户端       │
└─────────────┬────────────┘
              │
┌─────────────▼────────────┐
│        API Gateway        │
│ 鉴权 / 限流 / 灰度 / 日志  │
└─────────────┬────────────┘
              │
┌─────────────▼────────────┐
│       Agent Service       │
│  任务规划 / 工具调度 / 推理 │
└───────┬───────────┬──────┘
        │           │
        │           │
┌───────▼──────┐ ┌──▼──────────┐
│  LLM Service │ │ Tool Service │
│ 模型调用封装  │ │ 工具注册调用  │
└───────┬──────┘ └──┬──────────┘
        │           │
┌───────▼──────┐ ┌──▼──────────┐
│ OpenAI/其他模型│ │ DB/API/Search │
└──────────────┘ └─────────────┘

┌──────────────────────────┐
│  Observability Platform   │
│ 日志 / Trace / Metrics / 告警 │
└──────────────────────────┘

核心模块说明：

四、技术选型建议

1. 后端框架

推荐使用：

• Python + FastAPI；
• Node.js + NestJS；
• Java + Spring Boot；
• Go + Gin。

如果团队主要做 AI 应用，Python 生态更成熟，FastAPI 是一个非常适合快速构建 Agent 服务的框架。

2. 模型选择

生产环境不建议只依赖单一模型，可以根据任务复杂度进行分层：

这样既能保证效果，也能控制成本。

3. 数据库与缓存

推荐组合：

• PostgreSQL：存储用户、会话、任务、审计日志；
• Redis：缓存上下文、限流、任务状态；
• Milvus / Qdrant / pgvector：向量数据库；
• Elasticsearch：全文检索和日志搜索。

4. 部署方式

生产部署推荐：

• Docker 容器化；
• Kubernetes 编排；
• Nginx / API Gateway 反向代理；
• Prometheus + Grafana 监控；
• Loki / ELK 日志平台；
• CI/CD 自动发布。

五、项目目录结构

下面提供一个基于 FastAPI + OpenAI Compatible API + Redis + Docker 的 Agent 服务示例。

目录结构如下：

ai-agent-prod-demo/
├── app/
│   ├── main.py
│   ├── config.py
│   ├── agent.py
│   ├── llm.py
│   ├── tools.py
│   ├── schemas.py
│   ├── logger.py
│   └── middleware.py
├── requirements.txt
├── Dockerfile
├── docker-compose.yml
└── README.md

六、核心源码实现

1. 配置文件：app/config.py

import os
from pydantic_settings import BaseSettings


class Settings(BaseSettings):
    APP_NAME: str = "AI Agent Production Demo"
    APP_ENV: str = os.getenv("APP_ENV", "dev")

    OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
    OPENAI_BASE_URL: str = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    OPENAI_MODEL: str = os.getenv("OPENAI_MODEL", "gpt-4o-mini")

    REDIS_HOST: str = os.getenv("REDIS_HOST", "redis")
    REDIS_PORT: int = int(os.getenv("REDIS_PORT", "6379"))

    MAX_AGENT_STEPS: int = int(os.getenv("MAX_AGENT_STEPS", "5"))
    REQUEST_TIMEOUT: int = int(os.getenv("REQUEST_TIMEOUT", "30"))

    class Config:
        env_file = ".env"


settings = Settings()

2. 请求模型：app/schemas.py

from pydantic import BaseModel, Field
from typing import Optional, Dict, Any


class AgentRequest(BaseModel):
    user_id: str = Field(..., description="用户 ID")
    session_id: Optional[str] = Field(None, description="会话 ID")
    query: str = Field(..., description="用户问题")
    metadata: Optional[Dict[str, Any]] = Field(default_factory=dict)


class AgentResponse(BaseModel):
    answer: str
    trace_id: str
    steps: list = []

3. 日志模块：app/logger.py

import logging
import sys


def get_logger(name: str):
    logger = logging.getLogger(name)

    if logger.handlers:
        return logger

    logger.setLevel(logging.INFO)

    handler = logging.StreamHandler(sys.stdout)
    formatter = logging.Formatter(
        fmt="%(asctime)s | %(levelname)s | %(name)s | %(message)s"
    )
    handler.setFormatter(formatter)

    logger.addHandler(handler)
    return logger

4. LLM 调用封装：app/llm.py

from openai import OpenAI
from app.config import settings
from app.logger import get_logger

logger = get_logger(__name__)


class LLMClient:
    def __init__(self):
        self.client = OpenAI(
            api_key=settings.OPENAI_API_KEY,
            base_url=settings.OPENAI_BASE_URL
        )

    def chat(self, messages, temperature=0.2):
        logger.info("Calling LLM model=%s", settings.OPENAI_MODEL)

        response = self.client.chat.completions.create(
            model=settings.OPENAI_MODEL,
            messages=messages,
            temperature=temperature,
        )

        return response.choices[0].message.content

5. 工具注册与调用：app/tools.py

from typing import Dict, Callable, Any
from app.logger import get_logger

logger = get_logger(__name__)


class ToolRegistry:
    def __init__(self):
        self.tools: Dict[str, Callable] = {}

    def register(self, name: str, func: Callable):
        self.tools[name] = func

    def call(self, name: str, params: Dict[str, Any]):
        if name not in self.tools:
            raise ValueError(f"Tool not found: {name}")

        logger.info("Calling tool=%s params=%s", name, params)
        return self.tools[name](**params)

    def list_tools(self):
        return list(self.tools.keys())


def calculator(expression: str):
    """
    简单计算器工具。
    生产环境不建议直接 eval，这里仅用于演示。
    实际应使用安全表达式解析库。
    """
    allowed_chars = "0123456789+-*/(). "
    if not all(c in allowed_chars for c in expression):
        raise ValueError("Invalid expression")

    return str(eval(expression))


def search_order(order_id: str):
    """
    模拟订单查询工具。
    实际生产环境中应访问数据库或业务 API。
    """
    mock_orders = {
        "1001": {"status": "paid", "amount": 199, "product": "AI Course"},
        "1002": {"status": "shipped", "amount": 299, "product": "Cloud Service"},
    }

    return mock_orders.get(order_id, {"status": "not_found"})


tool_registry = ToolRegistry()
tool_registry.register("calculator", calculator)
tool_registry.register("search_order", search_order)

6. Agent 主逻辑：app/agent.py

import json
import uuid
from app.llm import LLMClient
from app.tools import tool_registry
from app.config import settings
from app.logger import get_logger

logger = get_logger(__name__)


SYSTEM_PROMPT = """
你是一个生产环境中的 AI Agent。

你可以根据用户问题选择是否调用工具。

当前可用工具：
1. calculator：用于数学计算，参数：{"expression": "1+2*3"}
2. search_order：用于查询订单，参数：{"order_id": "1001"}

你必须严格使用以下 JSON 格式输出：

如果需要调用工具：
{
  "type": "tool_call",
  "tool_name": "calculator",
  "params": {
    "expression": "1+2"
  }
}

如果可以直接回答：
{
  "type": "final",
  "answer": "你的回答"
}

注意：
- 不要输出 JSON 之外的任何文本；
- 不要调用不存在的工具；
- 工具参数必须符合要求；
- 如果信息不足，请向用户说明需要补充的信息。
"""


class AgentExecutor:
    def __init__(self):
        self.llm = LLMClient()

    def run(self, user_id: str, query: str):
        trace_id = str(uuid.uuid4())
        steps = []

        messages = [
            {"role": "system", "content": SYSTEM_PROMPT},
            {"role": "user", "content": query},
        ]

        for step in range(settings.MAX_AGENT_STEPS):
            logger.info("trace_id=%s agent_step=%s", trace_id, step + 1)

            raw_output = self.llm.chat(messages)

            try:
                decision = json.loads(raw_output)
            except Exception:
                logger.warning("Invalid JSON from LLM: %s", raw_output)
                messages.append({
                    "role": "user",
                    "content": "你的上一次输出不是合法 JSON，请严格按要求重新输出。"
                })
                continue

            steps.append(decision)

            if decision.get("type") == "final":
                return {
                    "answer": decision.get("answer", ""),
                    "trace_id": trace_id,
                    "steps": steps,
                }

            if decision.get("type") == "tool_call":
                tool_name = decision.get("tool_name")
                params = decision.get("params", {})

                try:
                    tool_result = tool_registry.call(tool_name, params)
                except Exception as e:
                    tool_result = {"error": str(e)}

                messages.append({
                    "role": "assistant",
                    "content": raw_output
                })

                messages.append({
                    "role": "user",
                    "content": f"工具 {tool_name} 的执行结果是：{json.dumps(tool_result, ensure_ascii=False)}。请继续推理或输出最终答案。"
                })

                continue

        return {
            "answer": "任务执行超过最大步骤限制，请简化问题后重试。",
            "trace_id": trace_id,
            "steps": steps,
        }

7. 中间件：app/middleware.py

import time
from starlette.middleware.base import BaseHTTPMiddleware
from app.logger import get_logger

logger = get_logger(__name__)


class AccessLogMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.time()

        response = await call_next(request)

        cost = round((time.time() - start) * 1000, 2)
        logger.info(
            "method=%s path=%s status=%s cost_ms=%s",
            request.method,
            request.url.path,
            response.status_code,
            cost
        )

        return response

8. FastAPI 入口：app/main.py

from fastapi import FastAPI, HTTPException
from app.schemas import AgentRequest, AgentResponse
from app.agent import AgentExecutor
from app.middleware import AccessLogMiddleware
from app.config import settings

app = FastAPI(title=settings.APP_NAME)

app.add_middleware(AccessLogMiddleware)

agent = AgentExecutor()


@app.get("/health")
def health():
    return {"status": "ok", "env": settings.APP_ENV}


@app.post("/agent/run", response_model=AgentResponse)
def run_agent(req: AgentRequest):
    if not req.query.strip():
        raise HTTPException(status_code=400, detail="query cannot be empty")

    result = agent.run(
        user_id=req.user_id,
        query=req.query
    )

    return result

七、依赖文件

requirements.txt

fastapi==0.115.0
uvicorn[standard]==0.30.6
openai==1.43.0
pydantic==2.8.2
pydantic-settings==2.4.0
redis==5.0.8

八、Docker 容器化部署

1. Dockerfile

FROM python:3.11-slim

WORKDIR /app

ENV PYTHONUNBUFFERED=1

COPY requirements.txt .

RUN pip install --no-cache-dir -r requirements.txt

COPY . .

EXPOSE 8000

CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

2. docker-compose.yml

version: "3.9"

services:
  ai-agent:
    build: .
    container_name: ai-agent
    ports:
      - "8000:8000"
    environment:
      APP_ENV: prod
      OPENAI_API_KEY: ${OPENAI_API_KEY}
      OPENAI_BASE_URL: ${OPENAI_BASE_URL}
      OPENAI_MODEL: ${OPENAI_MODEL}
      MAX_AGENT_STEPS: 5
    depends_on:
      - redis
    restart: always

  redis:
    image: redis:7
    container_name: ai-agent-redis
    ports:
      - "6379:6379"
    restart: always

九、本地运行方式

1. 配置环境变量

创建 .env 文件：

OPENAI_API_KEY=你的 API Key
OPENAI_BASE_URL=https://api.openai.com/v1
OPENAI_MODEL=gpt-4o-mini

如果你使用的是兼容 OpenAI API 的国内模型服务，也可以替换为对应的 Base URL 和模型名称。

2. 启动服务

docker compose up -d --build

查看服务状态：

docker ps

访问健康检查接口：

curl http://localhost:8000/health

正常返回：

{
  "status": "ok",
  "env": "prod"
}

3. 调用 Agent 接口

示例一：数学计算

curl -X POST http://localhost:8000/agent/run \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "u001",
    "query": "请帮我计算 128 * 36 + 99"
  }'

可能返回：

{
  "answer": "128 * 36 + 99 的计算结果是 4707。",
  "trace_id": "0f2744ae-1bb2-49ce-84d7-96f86bd7c232",
  "steps": [
    {
      "type": "tool_call",
      "tool_name": "calculator",
      "params": {
        "expression": "128*36+99"
      }
    },
    {
      "type": "final",
      "answer": "128 * 36 + 99 的计算结果是 4707。"
    }
  ]
}

示例二：订单查询

curl -X POST http://localhost:8000/agent/run \
  -H "Content-Type: application/json" \
  -d '{
    "user_id": "u001",
    "query": "帮我查一下订单 1002 的状态"
  }'

可能返回：

{
  "answer": "订单 1002 当前状态为 shipped，商品为 Cloud Service，金额为 299。",
  "trace_id": "c203e9a1-e28d-4bd9-982f-4e367a7f9478",
  "steps": [
    {
      "type": "tool_call",
      "tool_name": "search_order",
      "params": {
        "order_id": "1002"
      }
    },
    {
      "type": "final",
      "answer": "订单 1002 当前状态为 shipped，商品为 Cloud Service，金额为 299。"
    }
  ]
}

十、生产环境必须补强的能力

上面的源码可以作为基础版本，但真正上线前还需要补充更多能力。

1. 鉴权与权限控制

不能让任何人直接调用 Agent 接口。建议增加：

• API Token；
• JWT；
• OAuth2；
• 企业内部 SSO；
• 租户级权限；
• 用户角色权限。

例如：

普通用户：只能查询自己的订单；
客服人员：可以查询授权客户订单；
管理员：可以查看全量审计日志；
Agent：只能调用白名单内工具。

尤其需要注意，Agent 调用工具时也必须带上用户身份上下文，而不是以系统超级管理员身份执行所有操作。

2. 工具参数校验

生产环境中，工具参数不能完全相信模型输出。建议为每个工具定义 Schema。

例如订单查询工具参数：

from pydantic import BaseModel, Field


class SearchOrderParams(BaseModel):
    order_id: str = Field(..., pattern=r"^\d{4,20}$")

调用前先校验：

params = SearchOrderParams(**decision.get("params", {}))

这样可以避免模型生成异常参数、注入参数或越权参数。

3. 敏感操作二次确认

对于高风险操作，不能让 Agent 直接执行，例如：

• 删除数据；
• 退款；
• 转账；
• 修改权限；
• 发送外部邮件；
• 批量导出数据。

正确做法是：

Agent 生成操作计划
   ↓
用户确认
   ↓
后端校验权限
   ↓
执行操作
   ↓
记录审计日志

也就是说，Agent 可以“建议操作”，但不能在没有确认的情况下“直接操作”。

4. 上下文与记忆管理

Agent 的上下文管理一般分为：

生产环境不建议无限制把所有历史对话塞进 Prompt。应使用：

• 最近 N 轮对话；
• 摘要压缩；
• 向量检索；
• 重要信息结构化存储。

5. RAG 知识库增强

如果 Agent 要回答企业内部知识，建议结合 RAG：

用户问题
   ↓
向量检索相关文档
   ↓
拼接上下文
   ↓
模型生成答案
   ↓
返回引用来源

RAG 生产环境重点包括：

• 文档切分策略；
• Embedding 模型选择；
• 向量库召回；
• 关键词检索混合召回；
• 重排序；
• 引用来源；
• 文档权限过滤。

尤其是权限过滤非常关键。
例如员工 A 没有某份内部文档权限，即使向量检索命中了，也不能传给模型。

6. 限流与熔断

Agent 服务成本较高，必须做限流。

常见限流维度：

• IP；
• 用户；
• 租户；
• API Key；
• 接口；
• 模型；
• 时间窗口。

示例策略：

免费用户：每分钟 5 次，每天 100 次；
企业用户：每分钟 100 次；
内部管理员：每分钟 500 次；
单次请求最大推理轮数：5；
单次请求最大 Token：8000。

当模型服务异常时，也需要熔断和降级：

• 自动切换备用模型；
• 返回简化答案；
• 转人工处理；
• 写入异步任务队列稍后处理。

7. 日志审计

生产环境必须记录完整审计链路：

{
  "trace_id": "xxx",
  "user_id": "u001",
  "query": "帮我查询订单 1002",
  "model": "gpt-4o-mini",
  "tool_calls": [
    {
      "tool_name": "search_order",
      "params": {
        "order_id": "1002"
      },
      "result": {
        "status": "shipped"
      }
    }
  ],
  "latency_ms": 1234,
  "token_usage": {
    "prompt_tokens": 1200,
    "completion_tokens": 300
  }
}

日志用途包括：

• 问题排查；
• 成本分析；
• 用户行为分析；
• 安全审计；
• 模型效果评估。

8. Prompt 版本管理

Prompt 不应该散落在代码中。生产环境建议：

• Prompt 独立配置；
• 版本化管理；
• 支持灰度发布；
• 记录每次请求使用的 Prompt 版本；
• 支持回滚。

例如：

agent_customer_service_v1
agent_customer_service_v2
agent_order_query_v3

当某个 Prompt 版本导致错误率升高时，可以快速回滚到稳定版本。

9. 模型输出校验

模型输出必须校验。
建议使用：

• JSON Schema；
• Pydantic；
• Guardrails；
• Instructor；
• 自定义规则校验。

对于不合法输出，可以：

1. 要求模型重新生成；
2. 使用备用模型修复格式；
3. 返回异常并记录日志；
4. 降级为人工处理。

10. 监控指标

建议至少监控以下指标：

十一、Kubernetes 部署建议

如果业务规模较大，建议部署到 Kubernetes。

Deployment 示例

apiVersion: apps/v1
kind: Deployment
metadata:
  name: ai-agent
spec:
  replicas: 3
  selector:
    matchLabels:
      app: ai-agent
  template:
    metadata:
      labels:
        app: ai-agent
    spec:
      containers:
        - name: ai-agent
          image: your-registry/ai-agent:1.0.0
          ports:
            - containerPort: 8000
          env:
            - name: APP_ENV
              value: "prod"
            - name: OPENAI_MODEL
              value: "gpt-4o-mini"
            - name: OPENAI_API_KEY
              valueFrom:
                secretKeyRef:
                  name: ai-agent-secret
                  key: openai-api-key
          resources:
            requests:
              cpu: "500m"
              memory: "512Mi"
            limits:
              cpu: "2"
              memory: "2Gi"
          readinessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 5
            periodSeconds: 10
          livenessProbe:
            httpGet:
              path: /health
              port: 8000
            initialDelaySeconds: 20
            periodSeconds: 20

Service 示例

apiVersion: v1
kind: Service
metadata:
  name: ai-agent-service
spec:
  selector:
    app: ai-agent
  ports:
    - port: 80
      targetPort: 8000
  type: ClusterIP

十二、上线前检查清单

AI Agent 上线前建议逐项检查：

• [ ] 是否开启接口鉴权；
• [ ] 是否配置用户级限流；
• [ ] 是否限制最大推理轮数；
• [ ] 是否限制最大 Token；
• [ ] 是否对模型输出做结构化校验；
• [ ] 是否对工具参数做 Schema 校验；
• [ ] 是否对高风险操作增加二次确认；
• [ ] 是否记录完整审计日志；
• [ ] 是否支持 Trace ID；
• [ ] 是否监控模型调用耗时；
• [ ] 是否监控 Token 成本；
• [ ] 是否有备用模型或降级策略；
• [ ] 是否支持 Prompt 版本回滚；
• [ ] 是否对知识库检索做权限过滤；
• [ ] 是否有异常告警机制；
• [ ] 是否进行灰度发布；
• [ ] 是否进行安全测试和压力测试。

十三、常见问题与优化建议

1. Agent 经常调用错工具怎么办？

可以从以下方面优化：

• 在 Prompt 中明确工具适用场景；
• 减少工具数量，避免选择困难；
• 使用工具分类；
• 对工具参数增加描述；
• 对工具调用结果进行二次判断；
• 使用独立模型进行工具路由。

2. Agent 回答不稳定怎么办？

建议：

• 降低 temperature；
• 固定输出格式；
• 加入 Few-shot 示例；
• 对关键结果进行工具验证；
• 使用评测集持续测试；
• 对 Prompt 做版本管理。

3. Agent 响应太慢怎么办？

可以尝试：

• 使用流式输出；
• 简化推理步骤；
• 并行调用工具；
• 缓存常见问题；
• 对简单任务使用小模型；
• 将复杂任务转为异步任务。

4. 如何降低成本？

建议：

• 使用模型分层；
• 对 FAQ 做缓存；
• 使用向量检索减少上下文；
• 限制最大 Token；
• 限制 Agent 最大步骤；
• 对长对话做摘要；
• 对高频请求使用小模型。

十四、总结

AI Agent 的核心价值在于：它不仅能回答问题，还能调用工具、完成任务、连接真实业务系统。

但也正因为 Agent 具备执行能力，生产环境部署必须比普通 AI 应用更加谨慎。一个合格的生产级 Agent 系统，至少需要具备以下能力：

1. 稳定的模型调用封装；
2. 安全的工具注册与调用机制；
3. 严格的权限控制和参数校验；
4. 完整的日志、审计和链路追踪；
5. 限流、熔断、降级和成本控制；
6. Prompt 版本管理和灰度发布能力；
7. 面向业务场景的持续评测机制。

本文提供的 FastAPI 示例可以作为 AI Agent 工程化的起点。你可以在此基础上继续扩展 RAG 知识库、用户权限、Redis 限流、异步任务队列、Prometheus 监控、Kubernetes 部署和企业内部工具系统。

真正的生产级 AI Agent，不是一个简单的 Prompt，也不是一次模型调用，而是一套完整的软件工程体系。只有把模型能力、安全机制、业务规则和工程架构结合起来，Agent 才能稳定、可靠地在真实业务中创造价值。