Skip to content

cyKron613/agent-fastapi

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

5 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Agent-FastAPI

通用 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_iduser_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 配置

快速开始

1. 克隆项目

git clone <your-repo-url>
cd agent-fastapi

2. 安装依赖

pip install -r requirements.txt

3. 配置环境变量

创建 .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=

4. 初始化数据库

psql -U postgres -d agent_db -f create_chat_tables.sql

5. 启动服务

python main.py

服务启动后访问:

  • API 文档: http://localhost:18176/api-doc.html
  • ReDoc: http://localhost:18176/api-redoc.html

API 接口一览

所有接口前缀: /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} 删除消息

Agent 信息

方法 路径 说明
GET /agent/info 获取 Agent 基本信息
GET /agent/models 获取可用模型列表

系统

方法 路径 说明
GET /health 健康检查

SLS 日志分析

方法 路径 说明
GET /sls/v1/analysis 读取 SLS llm_stream 日志并返回实时聚合指标

接口详细说明

POST /chat/completions — 对话

请求体:

{
  "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]

GET /sessions — 获取会话列表

参数:

参数 类型 默认 说明
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"
    }
  ]
}

DELETE /sessions/{session_id} — 删除会话

参数:

参数 类型 默认 说明
hard bool false 是否硬删除 (不可恢复)

POST /sessions/{session_id}/stop — 停止响应

停止指定会话正在进行的流式输出。前端点击"停止生成"按钮时调用。


开发指南

创建自定义 Agent

核心步骤: 继承 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 实现

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)
    ),
):
    ...

数据库设计

agent_chat_sessions_test (会话表)

字段 类型 说明
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 更新时间

agent_chat_messages_test (消息表)

字段 类型 说明
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 配置

变量名 说明 示例
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 流式日志

项目可选接入阿里云 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_idrequest_idsession_iduser_idmodelchat_typepromptimage_countrequest_body
response 流式回答结束、取消、断连或异常时 stream_idresponse_bodyanswerduration_msttft_mschunk_countfinish_reasonerror

stream_id 用于关联同一次请求的 request/response 两条记录。SLS 中绑定字段以 extra. 前缀保存,例如 extra.event=llm_streamextra.phase=responseextra.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.eventextra.phaseextra.finish_reason,不依赖 SLS key-value 索引。details[].raw 会保留原始 SLS 字段,details[].response_body 会优先尝试按 JSON 解析,便于前端检查完整响应。


部署指南

Docker 部署

# 构建镜像
docker build -f deploy/docker/Dockerfile -t agent-fastapi .

# 运行容器
docker run -d \
  --name agent-fastapi \
  -p 8000:8000 \
  --env-file .env \
  agent-fastapi

Docker Compose

cd deploy/docker
docker-compose up -d

Kubernetes

# 测试环境
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          │
└──────────────────────────────────────────────────────────────┘

核心设计原则:

  1. 关注点分离 — API 路由 / 服务逻辑 / 数据访问 / 模型定义 各层职责清晰
  2. 开放-封闭原则 — 框架处理基础设施,开发者通过继承扩展业务逻辑
  3. 兼容性 — OpenAI Chat Completions API 格式,前端/SDK 可无缝对接
  4. 渐进式 — 可从 DefaultAgentService 开始,逐步添加 RAG、Tool Use 等能力

License

MIT

About

基于python+fastapi 快速搭建一个多轮对话集成的agent框架

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

 
 
 

Contributors

Languages