通用 Agent 开发框架 — 基于 FastAPI 构建的 AI Agent 后端脚手架,开箱即用,专为快速构建对话式 AI Agent 应用而设计。
Agent-FastAPI 是一个 通用的 Agent 开发框架,提供了构建对话式 AI Agent 所需的全部基础设施:
- 会话管理 — 创建、查询、重命名、删除会话
- 对话接口 — 兼容 OpenAI Chat Completions API 格式 (流式/非流式)
- 多模态图片输入 — 支持 OpenAI
image_url消息格式,统一走/chat/completions - 消息持久化 — 自动保存对话历史到 PostgreSQL
- 停止响应 — 支持中断正在进行的流式生成
- SLS 流式观测 — 对每次流式回答记录请求和终态响应,便于线上追踪、统计和排障
- Agent 信息 — 标准化的 Agent 元信息和能力声明
- 可扩展架构 — 通过继承
BaseAgentService快速实现自定义 Agent
开发者无需关心对话管理、Session 持久化、流式传输等基础设施,只需专注于实现自己的 Agent 逻辑。
本版本重点增强了图片对话、多模态消息和会话安全性:
- 统一图片对话入口:图片 URL 直接放入
messages[*].content[*].image_url.url,仍然调用POST /chat/completions,不再单独维护/upload接口。 - OpenAI 多模态消息兼容:
Message.content支持字符串和 OpenAI content parts 数组,例如text+image_url。 - 快捷字段:支持
query+images快捷请求格式;当不传messages时,框架会自动构造一条用户多模态消息。 - 视觉模型自动切换:检测到
image_url后,如果当前模型不是视觉模型,会自动切到LLM_VISION_MODEL_NAME,默认qwen-vl-max。 - 多模态历史持久化:图片消息会以 JSON 字符串保存到数据库,读取历史时会自动还原为多模态结构。
- 图片元数据回显:用户消息会把图片 URL 写入消息
metadata.images,并保留chat_type。 - 会话用户隔离:恢复已有会话时要求同时传
session_id和user_id,避免跨用户串会话。 - Swagger 多示例展示:对话接口使用
openapi_examples,可展示基础对话、多轮对话、图片 URL 对话多个示例。
| 特性 | 说明 |
|---|---|
| 🤖 OpenAI 兼容 | Chat Completions API 格式,前端可直接对接 |
| 🖼️ 多模态图片 | 支持 content 数组和 image_url 图片输入 |
| 👁️ 视觉模型路由 | 检测到图片后自动切换到 LLM_VISION_MODEL_NAME |
| 🌊 流式/非流式 | 同时支持 SSE 流式和一次性 JSON 响应 |
| 📝 自动会话管理 | 自动创建会话、生成标题、持久化消息 |
| 🔌 插件式架构 | 继承 BaseAgentService 实现自定义 Agent |
| ⏹️ 停止响应 | 支持中断流式生成 |
| 📡 SLS 日志 | 可选接入阿里云 SLS,记录 llm_stream 请求/响应生命周期 |
| 🗑️ 软删除 | 会话和消息支持软删除,数据可恢复 |
| 🎯 生命周期钩子 | on_before_chat / on_after_chat 扩展点 |
| 🛡️ 异常处理 | 全局异常处理、参数校验、业务异常 |
| 📊 数据库监控 | 连接池监控、自动重试机制 |
| 🔄 Redis 支持 | 内置 Redis 缓存客户端和中间件(可选) |
| 组件 | 技术 |
|---|---|
| Web 框架 | FastAPI |
| ORM | SQLAlchemy 2.0 (AsyncIO) |
| 数据库 | PostgreSQL + asyncpg |
| 缓存 | Redis (可选) |
| HTTP 客户端 | httpx |
| 配置管理 | pydantic-settings + python-decouple |
| 日志 | loguru + Aliyun SLS 可选上报 |
agent-fastapi/
├── main.py # 应用入口
├── requirements.txt # Python 依赖
├── create_chat_tables.sql # 数据库建表脚本
├── .env # 环境变量配置 (需自行创建)
├── system_prompt.txt # 系统提示词模板 (可选)
│
├── src/
│ └── main/
│ ├── api/ # 🔵 API 路由层
│ │ ├── endpoints.py # 路由注册入口
│ │ └── agent/
│ │ └── agent_router.py # ⭐ Agent 核心路由 (所有 API 定义)
│ │
│ ├── service/ # 🟢 服务层 (业务逻辑)
│ │ └── agent/
│ │ ├── base_agent_service.py # ⭐ Agent 服务基类 (抽象)
│ │ ├── default_agent_service.py # ⭐ 默认 Agent 实现
│ │ └── example_custom_agent.py # 📖 自定义 Agent 示例
│ │
│ ├── schema/ # 🟡 数据模型 (Pydantic)
│ │ ├── chat.py # 对话相关请求/响应模型
│ │ └── agent.py # Agent 信息模型
│ │
│ ├── models/ # 🔴 ORM 模型
│ │ └── chat.py # 会话 & 消息表定义
│ │
│ ├── repository/ # 🟣 数据访问层
│ │ └── chat_repository.py # 会话/消息 CRUD
│ │
│ ├── config/ # ⚙️ 配置
│ │ ├── manager.py # 配置工厂
│ │ ├── settings/ # 环境配置 (dev/prod)
│ │ ├── handler/ # 异常/中间件/生命周期/SLS 日志处理器
│ │ └── middleware/ # Redis 缓存中间件
│ │
│ └── core/ # 🧱 核心基础设施
│ ├── orm/ # ORM 基类 (DB/Session/Repository/Service)
│ ├── schema/ # 通用响应模型
│ └── util/ # 工具类 (Redis/异常/格式化)
│
├── deploy/ # 部署配置
│ ├── docker/ # Docker 配置
│ └── k8s/ # Kubernetes 配置
git clone <your-repo-url>
cd agent-fastapipip install -r requirements.txt创建 .env 文件:
# ===== 服务器配置 =====
ENVIRONMENT=DEV
BACKEND_SERVER_HOST=0.0.0.0
BACKEND_SERVER_PORT=18176
BACKEND_SERVER_WORKERS=1
API_PREFIX=/api
DOCS_URL=/api-doc.html
OPENAPI_URL=/api.json
REDOC_URL=/api-redoc.html
DOCS_AUTH_USERNAME=admin
DOCS_AUTH_PASSWORD=admin123
# ===== 数据库配置 (主库) =====
POSTGRES_CONNECT=postgresql
POSTGRES_HOST=localhost
POSTGRES_PORT=5432
POSTGRES_DB=agent_db
POSTGRES_USERNAME=postgres
POSTGRES_PASSWORD=postgres
# ===== 数据库配置 (副库,可与主库相同) =====
POSTGRES_CONNECT_ANOTHER=postgresql
POSTGRES_HOST_ANOTHER=localhost
POSTGRES_PORT_ANOTHER=5432
POSTGRES_DB_ANOTHER=agent_db
POSTGRES_USERNAME_ANOTHER=postgres
POSTGRES_PASSWORD_ANOTHER=postgres
# ===== 数据库连接池 =====
DB_MAX_POOL_CON=10
DB_POOL_SIZE=5
DB_POOL_OVERFLOW=10
DB_TIMEOUT=30
DB_POOL_RECYCLE=3600
DB_POOL_TIMEOUT=30
DB_POOL_RESET_ON_RETURN=rollback
DB_RETRY_ATTEMPTS=3
DB_RETRY_DELAY=1
DB_RETRY_BACKOFF=2
IS_DB_ECHO_LOG=false
IS_DB_FORCE_ROLLBACK=false
IS_DB_EXPIRE_ON_COMMIT=false
# ===== Redis 配置 =====
PROD_REDIS_HOST=localhost
PROD_REDIS_PORT=6379
PROD_REDIS_PASSWORD=
PROD_REDIS_DB=0
PROD_REDIS_CLUSTER=false
PROD_REDIS_NODES=
TEST_REDIS_HOST=localhost
TEST_REDIS_PORT=6379
TEST_REDIS_PASSWORD=
TEST_REDIS_DB=0
# ===== LLM 配置 =====
LLM_API_KEY=sk-your-api-key
LLM_BASE_URL=https://api.openai.com/v1
LLM_MODEL_NAME=gpt-4o
LLM_VISION_MODEL_NAME=gpt-4o
# ===== Aliyun SLS 日志配置,可留空关闭 =====
ALIYUN_LOGGING_SERVER=
ALIYUN_LOGGING_AKID=
ALIYUN_LOGGING_AKSECRET=
ALIYUN_LOGGING_PROJECT=
ALIYUN_LOGGING_LOGSTORE=psql -U postgres -d agent_db -f create_chat_tables.sqlpython main.py服务启动后访问:
- API 文档:
http://localhost:18176/api-doc.html - ReDoc:
http://localhost:18176/api-redoc.html
所有接口前缀: /api/agent/v1
| 方法 | 路径 | 说明 |
|---|---|---|
POST |
/chat/completions |
对话接口 (流式/非流式) |
| 方法 | 路径 | 说明 |
|---|---|---|
GET |
/sessions |
获取会话列表 (支持分页) |
POST |
/sessions |
创建新会话 |
GET |
/sessions/{session_id} |
获取会话详情及消息记录 |
PUT |
/sessions/{session_id} |
更新会话信息 |
DELETE |
/sessions/{session_id} |
删除会话 (软删除/硬删除) |
PATCH |
/sessions/{session_id}/rename |
重命名会话 |
| 方法 | 路径 | 说明 |
|---|---|---|
POST |
/sessions/{session_id}/stop |
停止流式响应 |
| 方法 | 路径 | 说明 |
|---|---|---|
DELETE |
/sessions/{session_id}/messages/{message_id} |
删除消息 |
| 方法 | 路径 | 说明 |
|---|---|---|
GET |
/agent/info |
获取 Agent 基本信息 |
GET |
/agent/models |
获取可用模型列表 |
| 方法 | 路径 | 说明 |
|---|---|---|
GET |
/health |
健康检查 |
| 方法 | 路径 | 说明 |
|---|---|---|
GET |
/sls/v1/analysis |
读取 SLS llm_stream 日志并返回实时聚合指标 |
请求体:
{
"messages": [
{"role": "user", "content": "你好"}
],
"user_id": "user_001",
"session_id": null,
"stream": true,
"model": "gpt-4o",
"temperature": 0.7,
"max_tokens": 2048,
"chat_type": "chat",
"metadata": {}
}session_id为空时自动创建新会话- 传
session_id恢复历史时建议同时传user_id,框架会按用户隔离会话 stream=true返回 SSE 流式响应stream=false返回标准 JSON 响应 (OpenAI 格式)content可为普通字符串,也可为 OpenAI 多模态 content parts 数组- 不传
messages时,可用query+images快捷字段自动构造用户消息
图片 URL 请求体 (OpenAI 多模态格式):
{
"messages": [
{
"role": "user",
"content": [
{"type": "text", "text": "请描述这张图片"},
{
"type": "image_url",
"image_url": {
"url": "https://example.com/image.png"
}
}
]
}
],
"chat_type": "vision",
"stream": true
}图片 URL 快捷请求体:
{
"query": "请描述这张图片",
"images": ["https://example.com/image.png"],
"chat_type": "vision",
"stream": true
}检测到
image_url后,默认实现会自动使用LLM_VISION_MODEL_NAME。例如默认文本模型是qwen3-max时,图片请求会自动切到qwen-vl-max。
流式响应格式 (SSE):
data: {"type": "session_info", "session_id": "xxxx-xxxx"}
data: {"choices": [{"delta": {"content": "你"}}]}
data: {"choices": [{"delta": {"content": "好"}}]}
data: [DONE]
参数:
| 参数 | 类型 | 默认 | 说明 |
|---|---|---|---|
limit |
int | 20 | 每页数量 (1-100) |
offset |
int | 0 | 偏移量 |
user_id |
string | null | 按用户筛选 |
响应:
{
"total": 42,
"items": [
{
"id": "uuid",
"title": "关于天气的对话",
"user_id": "user_001",
"model": "gpt-4o",
"created_at": "2026-02-24T10:00:00",
"updated_at": "2026-02-24T10:05:00"
}
]
}参数:
| 参数 | 类型 | 默认 | 说明 |
|---|---|---|---|
hard |
bool | false | 是否硬删除 (不可恢复) |
停止指定会话正在进行的流式输出。前端点击"停止生成"按钮时调用。
核心步骤: 继承 BaseAgentService,实现 process_chat() 方法。
# src/main/service/agent/my_agent_service.py
from typing import AsyncGenerator, List, Dict, Any, Optional
from src.main.service.agent.base_agent_service import BaseAgentService
from src.main.schema.chat import ChatRequest
from src.main.schema.agent import AgentInfoResponse
class MyAgentService(BaseAgentService):
"""我的自定义 Agent"""
def get_agent_info(self) -> AgentInfoResponse:
return AgentInfoResponse(
name="My Agent",
description="我的自定义 AI Agent",
version="1.0.0",
)
def get_system_prompt(self, request=None) -> Optional[str]:
return "你是一个专业的助手。"
async def process_chat(
self,
messages: List[Dict[str, Any]],
request: ChatRequest,
session_id: str,
) -> AsyncGenerator[str, None]:
"""
在这里实现你的 Agent 核心逻辑:
- RAG 检索增强
- Function Calling
- 多模型路由
- 业务数据注入
- ...
"""
# 示例: 调用 LLM
import httpx, json
from src.main.config.manager import settings
async with httpx.AsyncClient(timeout=120) as client:
async with client.stream(
"POST",
f"{settings.LLM_BASE_URL}/chat/completions",
json={"model": settings.LLM_MODEL_NAME, "messages": messages, "stream": True},
headers={"Authorization": f"Bearer {settings.LLM_API_KEY}"},
) as response:
async for line in response.aiter_lines():
if self.is_stream_cancelled(session_id):
break
if line.strip():
yield f"{line}\n\n"BaseAgentService 提供以下可覆盖的方法:
| 方法 | 必须实现 | 说明 |
|---|---|---|
process_chat() |
✅ | 核心对话处理逻辑 — 接收上下文消息,流式返回 SSE 格式响应 |
get_agent_info() |
❌ | 返回 Agent 元信息 (名称、能力、版本等) |
get_system_prompt() |
❌ | 返回系统提示词 (支持根据 request 动态生成) |
get_available_models() |
❌ | 返回可用模型列表 |
on_before_chat() |
❌ | 对话前钩子 — 预处理、校验、审计日志 |
on_after_chat() |
❌ | 对话后钩子 — 统计、评估、异步任务 |
框架自动处理的事项 (开发者无需关心):
- ✅ 会话创建/恢复
- ✅ 历史消息上下文构建
- ✅ 用户消息持久化
- ✅ 助手回复持久化
- ✅ 会话标题自动生成
- ✅ 流式传输管理
- ✅ 停止响应支持
- ✅ 异常捕获和错误响应
在 agent_router.py 中,将 DefaultAgentService 替换为你的实现:
# src/main/api/agent/agent_router.py
# 替换前:
from src.main.service.agent.default_agent_service import DefaultAgentService
# 替换后:
from src.main.service.agent.my_agent_service import MyAgentService as DefaultAgentService或者更精细地控制:
# 只替换 chat_completions 端点中的依赖
@router.post("/chat/completions")
async def chat_completions(
request: ChatRequest = Body(...),
agent_service: MyAgentService = Depends(
get_async_service(ser_type=MyAgentService, repo_type=ChatRepository)
),
):
...| 字段 | 类型 | 说明 |
|---|---|---|
id |
UUID | 主键 |
user_id |
VARCHAR(64) | 用户ID |
title |
VARCHAR(255) | 会话标题 (自动生成) |
model |
VARCHAR(100) | 使用的模型名称 |
system_prompt |
TEXT | 会话级系统提示词 |
metadata |
JSONB | 自定义元数据 |
is_deleted |
BOOLEAN | 软删除标记 |
created_at |
TIMESTAMP | 创建时间 |
updated_at |
TIMESTAMP | 更新时间 |
| 字段 | 类型 | 说明 |
|---|---|---|
id |
UUID | 主键 |
session_id |
UUID | 所属会话ID (外键, CASCADE) |
user_id |
VARCHAR(64) | 用户ID |
role |
VARCHAR(20) | 角色: system / user / assistant / tool |
content |
TEXT | 消息内容 |
token_count |
VARCHAR(20) | Token 消耗量 |
metadata |
JSONB | 自定义元数据 |
is_deleted |
BOOLEAN | 软删除标记 |
created_at |
TIMESTAMP | 创建时间 |
所有配置通过 .env 文件或环境变量注入。
| 变量名 | 说明 | 示例 |
|---|---|---|
ENVIRONMENT |
运行环境 | DEV / PROD |
BACKEND_SERVER_HOST |
服务地址 | 0.0.0.0 |
BACKEND_SERVER_PORT |
服务端口 | 18176 |
API_PREFIX |
API 路径前缀 | /api |
| 变量名 | 说明 | 示例 |
|---|---|---|
LLM_API_KEY |
API 密钥 | sk-xxxx |
LLM_BASE_URL |
API 地址 (兼容 OpenAI 格式) | https://api.openai.com/v1 |
LLM_MODEL_NAME |
默认模型 | gpt-4o |
LLM_VISION_MODEL_NAME |
图片/多模态请求默认视觉模型 | qwen-vl-max / gpt-4o |
支持任何兼容 OpenAI API 格式的 LLM 服务,如: OpenAI、Azure OpenAI、通义千问、DeepSeek、Ollama 等。 图片理解需要模型本身支持视觉能力;纯文本模型即使收到
image_url也无法分析图片。
| 变量名 | 说明 |
|---|---|
POSTGRES_HOST |
数据库地址 |
POSTGRES_PORT |
数据库端口 |
POSTGRES_DB |
数据库名 |
POSTGRES_USERNAME |
用户名 |
POSTGRES_PASSWORD |
密码 |
DB_POOL_SIZE |
连接池大小 |
项目可选接入阿里云 SLS。未配置 ALIYUN_LOGGING_* 时,SLS sink 会自动跳过,不影响服务启动和本地日志。
| 变量名 | 说明 |
|---|---|
ALIYUN_LOGGING_SERVER |
SLS endpoint,例如 cn-hangzhou.log.aliyuncs.com |
ALIYUN_LOGGING_AKID |
AccessKey ID |
ALIYUN_LOGGING_AKSECRET |
AccessKey Secret |
ALIYUN_LOGGING_PROJECT |
SLS Project |
ALIYUN_LOGGING_LOGSTORE |
SLS Logstore |
流式回答由 BaseAgentService.stream_chat() 统一记录,每次对话最多写两条 llm_stream 日志:
| phase | 触发时机 | 关键字段 |
|---|---|---|
request |
用户消息入库后、调用模型前 | stream_id、request_id、session_id、user_id、model、chat_type、prompt、image_count、request_body |
response |
流式回答结束、取消、断连或异常时 | stream_id、response_body、answer、duration_ms、ttft_ms、chunk_count、finish_reason、error |
stream_id 用于关联同一次请求的 request/response 两条记录。SLS 中绑定字段以 extra. 前缀保存,例如 extra.event=llm_stream、extra.phase=response、extra.finish_reason=complete。
- 只发送
logger.bind(event="llm_stream", ...)等白名单事件,普通logger.info()保持本地输出。 - 上传使用后台线程批量发送,业务请求不会等待 SLS 写入。
- 队列满时丢弃最旧日志,避免流式回答被日志系统拖慢。
- SDK 缺失、配置为空或上传失败都不会阻塞应用启动;上传失败只打印到
stderr,避免日志递归。
GET /api/sls/v1/analysis 会从 SLS 读取最近一段时间的 llm_stream 终态响应日志,并返回成功、拦截、失败、取消/断连等聚合指标。
| 参数 | 默认 | 范围 | 说明 |
|---|---|---|---|
window_minutes |
5 |
1-1440 |
查询最近多少分钟 |
limit |
1000 |
1-5000 |
最多读取多少条 SLS 日志 |
detail_limit |
50 |
1-200 |
返回多少条失败/拦截/取消明细 |
分析接口使用全文查询 "llm_stream" 后在本地过滤 extra.event、extra.phase 和 extra.finish_reason,不依赖 SLS key-value 索引。details[].raw 会保留原始 SLS 字段,details[].response_body 会优先尝试按 JSON 解析,便于前端检查完整响应。
# 构建镜像
docker build -f deploy/docker/Dockerfile -t agent-fastapi .
# 运行容器
docker run -d \
--name agent-fastapi \
-p 8000:8000 \
--env-file .env \
agent-fastapicd deploy/docker
docker-compose up -d# 测试环境
kubectl apply -f deploy/k8s/test/
# 生产环境
kubectl apply -f deploy/k8s/prod/┌──────────────────────────────────────────────────────────────┐
│ Client (前端/SDK) │
│ POST /chat/completions (SSE Stream) │
└─────────────────────────┬────────────────────────────────────┘
│
┌─────────────────────────▼────────────────────────────────────┐
│ API Router Layer │
│ (agent_router.py) │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────────────┐ │
│ │ /chat/ │ │ /sessions │ │ /agent/info │ │
│ │ completions │ │ CRUD │ │ /agent/models │ │
│ └──────┬───────┘ └──────┬───────┘ │ /health │ │
│ │ │ └──────────┬───────────┘ │
└─────────┼────────────────┼────────────────────┼──────────────┘
│ │ │
┌─────────▼────────────────▼────────────────────▼──────────────┐
│ Service Layer │
│ ┌─────────────────────────────────────────────────────────┐ │
│ │ BaseAgentService (抽象基类) │ │
│ │ ┌───────────────────────────────────────────────────┐ │ │
│ │ │ stream_chat() ← 框架核心流程 (自动管理) │ │ │
│ │ │ non_stream_chat() ← 非流式封装 │ │ │
│ │ │ stop_stream() ← 停止响应 │ │ │
│ │ └───────────────────────────────────────────────────┘ │ │
│ │ ┌───────────────────────────────────────────────────┐ │ │
│ │ │ process_chat() ← ⭐ 开发者实现 (核心逻辑) │ │ │
│ │ │ get_system_prompt() ← 开发者可选覆盖 │ │ │
│ │ │ on_before_chat() ← 前置钩子 │ │ │
│ │ │ on_after_chat() ← 后置钩子 │ │ │
│ │ └───────────────────────────────────────────────────┘ │ │
│ └─────────────────────────────────────────────────────────┘ │
│ ↓ │
│ ┌──────────────────────────────────┐ │
│ │ DefaultAgentService (默认实现) │ ← 直接代理 LLM │
│ │ ExampleCustomAgent (示例) │ ← 参考实现 │
│ │ YourAgentService (你的实现) │ ← 开发者自定义 │
│ └──────────────────────────────────┘ │
└──────────────────────────┬───────────────────────────────────┘
│
┌──────────────────────────▼───────────────────────────────────┐
│ Repository Layer │
│ (chat_repository.py) │
│ Session CRUD | Message CRUD │
└──────────────────────────┬───────────────────────────────────┘
│
┌──────────────────────────▼───────────────────────────────────┐
│ Database (PostgreSQL) │
│ agent_chat_sessions_test | agent_chat_messages_test │
└──────────────────────────────────────────────────────────────┘
核心设计原则:
- 关注点分离 — API 路由 / 服务逻辑 / 数据访问 / 模型定义 各层职责清晰
- 开放-封闭原则 — 框架处理基础设施,开发者通过继承扩展业务逻辑
- 兼容性 — OpenAI Chat Completions API 格式,前端/SDK 可无缝对接
- 渐进式 — 可从 DefaultAgentService 开始,逐步添加 RAG、Tool Use 等能力
MIT