生产级 Agentic AI 系统的 7 层架构详解

现代的代理型 AI 系统，无论是运行在开发、预发布还是生产环境中，都应构建为一组职责明确的架构层，而非单一服务。每一层分别负责代理编排、记忆管理、安全控制、可扩展性、故障处理等具体关注点。一个面向生产的代理系统通常会组合这些层，以确保在真实工作负载下具备可靠性、可观测性与安全性。

Production Grade Agentic System (Created by Fareed Khan)

在一个代理系统中，有两个关键方面需要持续监控：

代理行为：包括推理准确性、工具调用正确性、记忆一致性、安全边界，以及在多轮交互与多代理间的上下文处理。
系统可靠性与性能：涵盖延迟、可用性、吞吐量、成本效率、故障恢复，以及整个架构中各依赖的健康状况。

这两者对于大规模可靠运行多代理系统同样重要。

在本文中，我们将介绍构建一个生产就绪的代理系统所需的核心架构各层，以帮助团队自信地在自有基础设施或为客户部署 AI 代理。

创建模块化代码库

通常，Python 项目从小规模起步，但随着增长容易变得混乱。在构建面向生产的系统时，开发者通常采用模块化架构方法。

这意味着将应用的不同组件拆分为独立模块。这样我们能更容易地维护、测试与更新各部分，而不影响整个系统。

让我们为这个 AI 系统创建一个结构化的目录布局：

├── app/                        # 主应用源代码
│   ├── api/                    # API 路由处理器
│   │   └── v1/                 # 版本化 API (v1 端点)
│   ├── core/                   # 核心应用配置与逻辑
│   │   ├── langgraph/          # AI 代理 / LangGraph 逻辑
│   │   │   └── tools/          # 代理工具 (搜索、操作等)
│   │   └── prompts/            # AI 系统与代理提示词
│   ├── models/                 # 数据库模型 (SQLModel)
│   ├── schemas/                # 数据验证模式 (Pydantic)
│   ├── services/               # 业务逻辑层
│   └── utils/                  # 共享辅助工具
├── evals/                      # AI 评估框架
│   └── metrics/                # 评估指标与标准
│       └── prompts/            # LLM 作为评判者的提示词定义
├── grafana/                    # Grafana 可观测性配置
│   └── dashboards/             # Grafana 仪表盘
│       └── json/               # 仪表盘 JSON 定义
├── prometheus/                 # Prometheus 监控配置
├── scripts/                    # DevOps 与本地自动化脚本
│   └── rules/                  # Cursor 项目规则
└── .github/                    # GitHub 配置
    └── workflows/              # GitHub Actions CI/CD 工作流

这套目录结构遵循了在代理系统或纯软件工程中广泛应用的通用最佳实践模式。每个文件夹都有明确用途：

app/：主应用代码，包括 API 路由、核心逻辑、数据库模型与工具函数。
evals/：用于基于多种指标与提示词评估 AI 性能的框架。
grafana/ 与 prometheus/：监控与可观测性工具的配置。

你会看到很多组件有自己的子目录（例如 langgraph/ 和 tools/），这进一步实现了关注点分离。接下来我们将一步步构建这些模块，并理解每一部分的重要性。

管理依赖项

构建面向生产的 AI 系统，第一步是制定依赖管理策略。小项目通常使用简单的 requirements.txt，而对于复杂项目，建议使用 pyproject.toml，因为它支持更高级的依赖解析、版本管理与构建系统规范。

创建 pyproject.toml 并开始添加依赖与配置：

# ==========================
# Project Metadata
# ==========================
# Basic information about your Python project as defined by PEP 621
[project]
name = "My Agentic AI System"              # The distribution/package name
version = "0.1.0"                          # Current project version (semantic versioning recommended)
description = "Deploying it as a SASS"     # Short description shown on package indexes
readme = "README.md"                       # README file used for long description
requires-python = ">=3.13"                 # Minimum supported Python version

第一部分定义了项目元数据，如名称、版本、描述、Python 版本要求。当你发布到 PyPI 等索引时，这些信息会用到。

随后是核心依赖。由于我们要构建一个面向 ≤10K 活跃用户的 Agentic AI 系统，需要覆盖 Web 框架、数据库、认证、AI 编排、可观测性等多类库。

# ==========================
# Core Runtime Dependencies
# ==========================
# These packages are installed whenever your project is installed
# They define the core functionality of the application

dependencies = [
    # --- Web framework & server ---
    "fastapi>=0.121.0",
    "uvicorn>=0.34.0",
    "asgiref>=3.8.1",
    "uvloop>=0.22.1",

    # --- LangChain / LangGraph ecosystem ---
    "langchain>=1.0.5",
    "langchain-core>=1.0.4",
    "langchain-openai>=1.0.2",
    "langchain-community>=0.4.1",
    "langgraph>=1.0.2",
    "langgraph-checkpoint-postgres>=3.0.1",

    # --- Observability & tracing ---
    "langfuse==3.9.1",
    "structlog>=25.2.0",

    # --- Authentication & security ---
    "passlib[bcrypt]>=1.7.4",
    "bcrypt>=4.3.0",
    "python-jose[cryptography]>=3.4.0",
    "email-validator>=2.2.0",

    # --- Database & persistence ---
    "psycopg2-binary>=2.9.10",
    "sqlmodel>=0.0.24",
    "supabase>=2.15.0",

    # --- Configuration & environment ---
    "pydantic[email]>=2.11.1",
    "pydantic-settings>=2.8.1",
    "python-dotenv>=1.1.0",

    # --- API utilities ---
    "python-multipart>=0.0.20",
    "slowapi>=0.1.9",

    # --- Metrics & monitoring ---
    "prometheus-client>=0.19.0",
    "starlette-prometheus>=0.7.0",

    # --- Search & external tools ---
    "duckduckgo-search>=3.9.0",
    "ddgs>=9.6.0",

    # --- Reliability & utilities ---
    "tenacity>=9.1.2",
    "tqdm>=4.67.1",
    "colorama>=0.4.6",

    # --- Memory / agent tooling ---
    "mem0ai>=1.0.0",
]

你会注意到我们针对每个依赖都指定了版本（使用 >= 运算符），这在生产系统中极为重要，可以避免“依赖地狱”，即不同库对同一包的版本要求冲突。

然后是开发依赖。多人在同一代码库协作时，要保证代码质量与一致性，需要 Linters、Formatters、Type Checkers 等开发工具。

# ==========================
# Optional Dependencies
# ==========================
# Extra dependency sets that can be installed with:
#   pip install .[dev]

[project.optional-dependencies]
dev = [
    "black",
    "isort",
    "flake8",
    "ruff",
    "djlint==1.36.4",
]

接着我们为测试定义依赖分组，便于逻辑归类，例如将所有测试相关库归为 `test` 组。

```toml
# ==========================
# Dependency Groups (PEP 735-style)
# ==========================
# Logical grouping of dependencies, commonly used with modern tooling

[dependency-groups]
test = [
    "httpx>=0.28.1",
    "pytest>=8.3.5",
]

# ... 其余工具配置保持不变 ...

这些配置的作用如下：

Dependency Groups：创建逻辑依赖分组，例如 test 分组。
Pytest Configuration：定制 pytest 的测试发现与运行方式。
Black：统一代码格式。
Flake8：风格检查与潜在错误提示。
Radon：监控圈复杂度以保持代码可维护性。
isort：自动排序 import 语句。

我们还定义了 Pylint 与 Ruff 等其他检查工具。它们虽然是可选的，但强烈建议在生产系统中使用，因为随着代码库增长，没有这些工具将难以管理。

环境配置管理

接下来设置最常见的配置，即配置管理。

小型项目通常直接使用 .env 文件存储环境变量；更规范的做法是使用 .env.example 文件并提交到版本库，同时使用不同环境的配置文件：

# 不同环境的配置文件
.env.[development|staging|production] # 例如 .env.development

为什么不使用单一的 .env 文件？因为这样可以针对不同环境（例如开发环境开启 debug，生产环境关闭）保留独立的配置，而无需频繁修改同一个文件。

创建 .env.example 文件，添加必要环境变量的占位值：

# ==================================================
# Application Settings
# ==================================================
APP_ENV=development
PROJECT_NAME="Project Name"
VERSION=1.0.0
DEBUG=true

接着是 API 设置、CORS 以及 Langfuse 观测性配置：

# ==================================================
# API Settings
# ==================================================
API_V1_STR=/api/v1

# ==================================================
# CORS (Cross-Origin Resource Sharing) Settings
# ==================================================
ALLOWED_ORIGINS="http://localhost:3000,http://localhost:8000"

# ==================================================
# Langfuse Observability Settings
# ==================================================
LANGFUSE_PUBLIC_KEY="your-langfuse-public-key"
LANGFUSE_SECRET_KEY="your-langfuse-secret-key"
LANGFUSE_HOST=https://cloud.langfuse.com

API_V1_STR 便于进行 API 版本化；CORS 设置用于控制哪些前端域名可以访问后端 API（从而集成 AI Agents）。我们还将使用 Langfuse 标准化监控 LLM 交互，因此需要设置其密钥与主机地址。

接着是 LLM、JWT 与数据库（PostgreSQL）配置，以及连接池参数：

# LLM 设置
OPENAI_API_KEY="your-llm-api-key"
DEFAULT_LLM_MODEL=gpt-4o-mini
DEFAULT_LLM_TEMPERATURE=0.2

# JWT 设置
JWT_SECRET_KEY="your-jwt-secret-key"
JWT_ALGORITHM=HS256
JWT_ACCESS_TOKEN_EXPIRE_DAYS=30

# PostgreSQL 设置与连接池参数
POSTGRES_HOST=db
POSTGRES_DB=mydb
POSTGRES_USER=myuser
POSTGRES_PORT=5432
POSTGRES_PASSWORD=mypassword
POSTGRES_POOL_SIZE=5
POSTGRES_MAX_OVERFLOW=10

最后是速率限制与日志设置：

# Rate Limiting Settings (SlowAPI)
RATE_LIMIT_DEFAULT="1000 per day,200 per hour"
RATE_LIMIT_CHAT="100 per minute"
RATE_LIMIT_CHAT_STREAM="100 per minute"
RATE_LIMIT_MESSAGES="200 per minute"
RATE_LIMIT_LOGIN="100 per minute"

# Logging
LOG_LEVEL=DEBUG
LOG_FORMAT=console

至此，我们已完成依赖与配置管理策略。接下来，将在应用代码中使用 Pydantic Settings Management 来加载这些配置（app/core/config.py）。后续的代码逻辑保持不变，其核心是加载环境变量、解析列表或字典、定义 Settings 类、按环境覆写关键配置，并初始化全局 settings 对象。

容器化策略

我们将创建 docker-compose.yml 文件来定义应用所需的全部服务。在生产系统中，数据库、监控工具与 API 并非孤立运行，而是需要相互通信。Docker Compose 是编排多容器应用的标准方式。

Building Data Persistence Layer

我们已有数据库，但还没有结构。AI 系统高度依赖结构化数据：需要严格定义 Users、Chat Sessions 与 AI State 之间的关系。我们使用 SQLModel（整合 SQLAlchemy 与 Pydantic）来实现。

Structured Modeling

首先定义 BaseModel，抽取公共字段（如 created_at），以遵循 DRY 原则。随后定义 User 实体（包含 email、hashed_password，并封装 verify_password/hash_password 方法），再定义 Session（代表特定会话，关联用户，使用 UUID 以增强安全性）。

Entity Definition

User 与 Session 的表结构与关系如原文图示。注意将密码哈希逻辑封装到模型中，防止安全误用。

Data Transfer Objects (DTOs)

为 LangGraph Persistence 定义 Thread 模型，用于校验线程存在。然后在 app/models/database.py 中聚合导出，方便统一导入。最后定义用于 API 输入/输出契约的 Schemas：auth.py（注册、Token、公共用户信息）、chat.py（消息、请求/响应、流式响应）、graph.py（LangGraph 的 State 对象，包含 messages 与 long_term_memory）。这实现了数据库层与 API 层的类型安全，避免敏感字段外泄与坏数据污染。

Security & Safeguards Layer

生产环境下不能信任用户输入，也不能允许无限访问。类似 together.ai 等 API 提供方会对每分钟请求数进行限流，以防止滥用、保护基础设施并控制成本。

Security Layer (Created by Fareed Khan)

若没有防护会发生两件事：
1. 滥用：爬虫或脚本狂刷 API，导致 OpenAI 账单飙升。
2. 安全漏洞：恶意用户尝试注入攻击。

Rate Limiting Feature

使用 SlowAPI 集成 FastAPI 进行限流，基于 IP 或按需改用 X-Forwarded-For。通过 @limiter.limit(...) 可对不同路由施加精细限额。

Sanitization Check Logic

即使前端框架做了 XSS 防护，后端也不能盲信字符串。实现 sanitize_string 与 sanitize_email 等工具，进行 HTML 转义、去除 script 标签、校验格式等。

接着实现 JWT 工具，用于无状态认证：create_access_token（签发）、verify_token（验证）。这样无需每次命中数据库即可确认登录状态。

Context Management

扩展上下文窗口管理。随着消息累积，容易触发模型 token 上限或增加成本。实现 prepare_messages 智能裁剪历史，保留系统提示与最近消息；process_llm_response 规范化模型响应（合并 reasoning blocks 与文本）。这样即便历史消息达到 500 条，也不会溢出。

The Service Layer for AI Agents

在合理的架构下，API 路由应保持简洁，复杂业务逻辑与数据库访问属于服务层，便于测试、复用与维护。

Connection Pooling

实现数据库连接池，开启 pool_pre_ping 避免长时间空闲后连接被关闭导致的“Broken Pipe”错误；设置 pool_recycle 以适应云数据库的连接空闲策略；提供用户与会话的 CRUD 操作。

LLM Unavailability Handling

单一模型存在风险：服务宕机或限流。我们实现韧性与回退机制：使用 tenacity 实现自动重试（指数退避）、循环回退（例如从 gpt-4o 切换到 gpt-4o-mini），并封装在 LLMService 中。遇到 RateLimitError、APITimeoutError 或 APIError 则重试，重试耗尽后自动切换到下一个模型，以保障高可用性。

Circuit Breaking

在 LLM 层绑定工具，使模型可在 Agent 场景中调用工具。创建全局 llm_service 实例便于全局复用。若提供方出现大规模故障，tenacity 会触发回退，尽量避免 500 错误。

Multi-Agentic Architecture

使用 LangGraph 构建有状态代理。代理可循环、重试、调用工具、记忆历史、持久化状态（重启可恢复到中断位置）。用户也期望跨会话的长期记忆。

Long-Term Memory Integration

集成 mem0ai 作为长期记忆。将 prompts 作为资产与代码分离，使用变量注入 {long_term_memory} 与当前时间等。

Tool Calling Feature

将工具模块化。构建 app/core/langgraph/graph.py，包含：
1. 状态管理：将对话状态持久化到 Postgres。
2. 记忆检索：从 mem0ai 获取相关用户事实并注入系统提示。
3. 执行循环：调用 LLM、解析工具调用、执行并反馈。
4. 流式传输：实时向用户推送 tokens。

核心方法 get_response：检索长记忆 → 执行图 → 异步更新记忆 → 返回消息列表。也添加了 Langfuse CallbackHandler 以追踪每一步。

Building The API Gateway

需要认证与授权。先构建认证端点，用 FastAPI 依赖注入来统一校验。

定义 get_current_user 依赖：从 Authorization 头提取并验证 JWT，查询用户、绑定日志上下文；get_current_session 用于基于会话的校验，统一审计与风控。

实时流式传输

构建聊天 API。系统支持两类交互模式：

标准聊天：单次请求，阻塞式等待，返回完整响应。
流式聊天：非阻塞式，以 Server-Sent Events (SSE) 格式实时推送生成的 token，提供更优的用户体验。

API 端点设计如下：
* /chatbot/chat：返回完整的聊天响应。
* /chatbot/chat/stream：使用异步生成器流式推送响应，并在末尾发送 done 标记。
* /messages：用于获取或清空用户的历史消息。

最后，通过路由聚合器 app/api/v1/api.py 整合所有模块化子路由，并提供 /health 端点用于健康检查。

可观测性与运维测试

对于一个服务上万用户的系统，可观测性至关重要。我们通过 Prometheus 指标和上下文感知的日志来跟踪系统性能、用户行为和错误。

创建评估指标

在 app/core/metrics.py 中定义并暴露 Prometheus 指标，包括 HTTP 请求总数与延迟直方图、数据库连接数以及 LLM 推理耗时（使用自定义桶分布）。通过 setup_metrics(app) 函数注册 /metrics 端点。

基于中间件的测试

在 app/core/middleware.py 中实现中间件，统一记录请求时延与状态码，并将用户/会话 ID 注入日志上下文：
* MetricsMiddleware：统计所有请求（排除 /metrics 和 /health）的数量与耗时。
* LoggingContextMiddleware：从 JWT 中提取用户标识并绑定到日志上下文，确保即使在认证失败的情况下也能保持一致的日志追踪。

流式端点交互

如前所述，/chat 与 /chat/stream 端点已实现。流式端点以 SSE 格式 (data: {json}nn) 持续向前端推送数据。发生错误时，返回结构化错误信息并结束流。

此外，结合 LangGraph 的检查点恢复能力，系统支持在页面刷新后仍能查看历史会话。

使用异步进行上下文管理

app/main.py 负责应用的装配与生命周期管理。采用现代的 asynccontextmanager 替代旧的 @app.on_event 装饰器：启动时记录元数据，关闭时清理资源（如刷新 Langfuse 数据）。

配置中间件的执行顺序至关重要：
1. LoggingContextMiddleware（最外层，优先绑定上下文）
2. MetricsMiddleware（计时）
3. CORS（安全）

同时注册 SlowAPI 的异常处理，并自定义 RequestValidationError 以返回更友好的 JSON 错误信息。根路由 / 和健康检查端点 /health 均受限流保护，其中 /health 会实际检查数据库连接状态，并据此返回 200 或 503 状态码。

DevOps 自动化

生产系统需要卓越的运维能力，涵盖部署、监控以及确保依赖服务就绪。

Dockerfile：构建安全、轻量的镜像，使用非 root 用户，缓存依赖层，并通过入口脚本 scripts/docker-entrypoint.sh 检查关键环境变量。
Prometheus：在 prometheus/prometheus.yml 中配置，抓取 FastAPI 应用和 cAdvisor 的指标。
Grafana：采用“仪表盘即代码”模式，在 grafana/dashboards/dashboards.yml 中配置，自动加载 JSON 仪表盘定义。
Makefile：封装常用命令，如安装依赖、开发热重载、Docker 启动和评估等。
GitHub Actions：在 .github/workflows/deploy.yaml 中配置 CI/CD 流水线，当代码推送到主分支时自动构建并推送镜像至 Docker Hub。

通过以上步骤，完成了从部署、监控到运维自动化的“生产级”闭环。

评估框架

AI 系统具有概率性，其表现不像传统软件那样确定。对提示词的更新可能修复一个边界问题，却引入新的回归。因此，我们需要在接近生产环境的场景下持续评估 Agent 的表现，尽早发现问题。

构建评估框架，采用 LLM-as-a-Judge 模式：由更强的 LLM 基于 Langfuse 记录的追踪数据自动进行评分。

LLM 作为评审员

定义评分标准，使用结构化输出模式强制模型给出分数和理由（ScoreSchema）。为不同评估维度（如幻觉、毒性、相关性、帮助性、简洁性）定义指标提示词（如 hallucination.md、toxicity.md），并通过 evals/metrics/__init__.py 动态加载所有 .md 文件。

实现 evals/evaluator.py：
1. 从 Langfuse 拉取近期的追踪数据。
2. 过滤尚未评分的记录。
3. 对每条记录，遍历所有评估指标，使用强模型（如 GPT-4o）作为评审员进行打分。
4. 将评分结果回传至 Langfuse，以便进行趋势可视化和监管。

自动化评分

evals/main.py 提供 CLI 入口，可手动触发或由 CI/CD 的定时任务自动执行。这形成了一个自我监控的反馈回路：如果一次提示词更新导致更多幻觉，次日“幻觉分数”指标将显著升高。

这也是区分“简单项目”与“生产级 AI 平台”的关键环节之一。

架构压力测试

原型系统与生产系统的一个重要区别在于负载承受能力。我们需要验证系统的并发处理能力，以避免数据库连接耗尽、OpenAI 限流碰撞或延迟飙升等问题。

我们将模拟 1,500 个并发用户同时访问聊天端点，类似营销活动后的流量高峰场景。

模拟流量

压力测试不应在本地开发机上进行（网络和 CPU 瓶颈会干扰结果）。选择云主机（如 AWS m6i.xlarge，4 vCPU/16 GiB RAM），开放端口，安装 Docker 环境。使用 make docker-run-env ENV=development 启动开发环境验证连通性，然后编写 tests/stress_test.py 脚本，模拟完整流程（登录 → 创建会话 → 聊天）进行并发压测。

性能分析

运行压力测试后，日志显示大量 200 成功响应。在流量高峰时，观察到 switching_model_fallback 日志，表明当主模型触发限流时，系统自动切换到备用模型（如 gpt-4o-mini）以保持服务不中断。整体成功率约为 98.4%，平均延迟约 1.2 秒，失败请求多为 429 错误（由 OpenAI 限流导致）。

随后使用 Prometheus 查询每秒请求率（RPS），并通过 Grafana/Prometheus 仪表板展示 /auth/login、/chatbot/chat、/auth/session 等关键端点的 RPS 峰值。同时，利用 Langfuse 查看详细的调用链追踪（traces），分析延迟与成本数据。观测结果显示，延迟在 0.98 秒至 2.10 秒的合理区间内波动，成本数据清晰可审计。

为进一步确保系统健壮性，可以进行逐步加压测试与长时间高压测试，以检测潜在的内存泄漏等问题。更深入的性能压测与监控实践，可参考相关 GitHub 项目。

关注“鲸栖”小程序，掌握最新AI资讯

本文来自网络搜集，不代表鲸林向海立场，如有侵权，联系删除。转载请注明出处：http://www.itsolotime.com/archives/14756