Marico's space

最近折腾了 FastAPI + AI Agent 的组合，踩了几个坑，这篇把问题说清楚。

用 FastAPI 构建 AI Agent 其实没那么复杂，关键是把几个核心模块拆清楚：LLM 调用、工具系统、记忆层。把这些理顺了，剩下的就是排列组合。

Python 环境怎么搭

先搞个干净的 Python 环境，装好依赖库。我一般用 pyenv 管理 Python 版本，配合 venv 做虚拟隔离。在 Ubuntu 上跑的话，基本配置如下：

# 安装指定 Python 版本（可选）
pyenv install 3.11.5
pyenv global 3.11.5 # 创建并激活虚拟环境
python -m venv .venv
source .venv/bin/activate # 升级 pip 并安装核心依赖
pip install --upgrade pip
pip install fastapi uvicorn python-dotenv

AI 这边加上 OpenAI SDK、LangChain、LangGraph 和 SQLAlchemy（持久化用）：

pip install openai langchain langgraph[all] sqlalchemy

如果要在 GPU 上跑，加个 torch 和 transformers。版本号最好锁死，不然 CI 跑着跑着可能突然挂掉：

fastapi==0.110.0
uvicorn[standard]==0.27.0
openai==1.12.0
langchain==0.2.0
langgraph==0.0.12
sqlalchemy==2.0.29
python-dotenv==1.0.1

容易踩的坑：

• Docker 构建时忘了激活 venv
• OpenAI SDK 版本不匹配，接口签名变了

加个简单的导入测试就能提前发现这两个问题：

# tests/test_imports.py
def test_imports(): import fastapi, openai, langchain, langgraph, sqlalchemy

选哪个 Agent 框架

简单说：看你的 Agent 要多复杂。如果只是简单调个 ChatCompletion，原生 SDK 够用。想要可组合的链、内置的工具选择和记忆系统，LangChain 是折中方案。想要精细控制工具调用的循环流程，LangGraph 最灵活。

OpenAI SDK——轻量级

import openai def simple_chat(prompt: str) -> str: resp = openai.ChatCompletion.create( model="gpt-4o-mini", messages=[{"role": "user", "content": prompt}], temperature=0.2, ) return resp.choices[0].message.content

优点：依赖少，容易排查问题。缺点：记忆、工具调用、重试逻辑全得自己写。

LangChain——功能齐全

from langchain import OpenAI, LLMChain, PromptTemplate
from langchain.memory import ConversationBufferMemory llm = OpenAI(model="gpt-4o-mini", temperature=0.2)
prompt = PromptTemplate.from_template("{history}\nUser: {input}\nAI:")
memory = ConversationBufferMemory(k=5)
chain = LLMChain(llm=llm, prompt=prompt, memory=memory) def chat_with_memory(user_input: str) -> str: return chain.run({"input": user_input})

优点：内置记忆、工具封装、LLM 提供商切换方便。缺点：依赖树大，有时候封装层会把延迟问题藏起来。

LangGraph——显式图控制

from langgraph.graph import GraphBuilder
from langchain import OpenAI
from langgraph.prebuilt import tool_node llm = OpenAI(model="gpt-4o-mini")
builder = GraphBuilder() @builder.node
def start(state): return {"msg": "What do you need help with?"} @builder.node
def tool_use(state): # 示例工具调用：简单计算器 if "calc:" in state["msg"]: expr = state["msg"].split("calc:")[1] result = eval(expr) # 生产环境千万别这么写！ return {"msg": f"The answer is {result}"} return {"msg": state["msg"]} builder.add_edge("start", "tool_use")
graph = builder.compile()

优点：流程一目了然，可以加循环、调试每个节点。缺点：样板代码多，学习曲线陡。

什么时候别用重型框架？
如果 Agent 只是单轮查询或者低流量的 webhook，原生 SDK 延迟低，Docker 镜像也小。

Agent 架构设计：工具、记忆、循环

架构说白了就三块：LLM 核心、可调用的工具集合、跨轮次的记忆层。

1. LLM 核心

把模型封装成薄薄一层，方便以后换 provider。暴露一个 generate 函数，接收消息列表，返回助手回复。

def generate(messages: list[dict]) -> str: resp = openai.ChatCompletion.create( model="gpt-4o-mini", messages=messages, temperature=0.0, ) return resp.choices[0].message.content

2. 工具接口

定义一个简单的协议，每个工具都得实现：

from typing import Protocol, Any class Tool(Protocol): name: str description: str def run(self, input: str) -> Any: ...

然后注册到 dict 里：

tools: dict[str, Tool] = {} def register(tool: Tool): tools[tool.name] = tool

来个具体例子——天气查询：

import httpx class WeatherTool: name = "weather" description = "Fetches current weather for a city." async def run(self, city: str) -> str: url = f"https://api.open-meteo.com/v1/forecast?city={city}¤t_weather=true" async with httpx.AsyncClient() as client: data = await client.get(url) return data.json()["current_weather"]["temperature"] # 简化版

3. 记忆层

我倾向混合方案：短期记忆放 Redis（快，多 worker 共享），长期事实存 SQLite + SQLAlchemy。短期存储保留最近 k 条消息，长期存储存用户偏好。

# redis_mem.py
import redis
import json
r = redis.from_url("redis://localhost:6379/0") def push_message(user_id: str, role: str, content: str): key = f"mem:{user_id}" r.rpush(key, json.dumps({"role": role, "content": content})) r.ltrim(key, -10, -1) # 保留最近 10 条消息 def get_history(user_id: str) -> list[dict]: key = f"mem:{user_id}" raw = r.lrange(key, 0, -1) return [json.loads(m) for m in raw]

4. 循环

把这些串起来：

async def agent_loop(user_id: str, user_input: str) -> str: # 1. 加载最近的历史 history = get_history(user_id) # 2. 加入当前用户消息 history.append({"role": "user", "content": user_input}) # 3. 让 LLM 决定是否调用工具 response = generate(history) # 简单判断：如果回复包含工具标签如 "<tool:weather>" if response.startswith("<tool:"): tool_name, arg = response[6:].split(">", 1) tool = tools.get(tool_name) if tool: tool_result = await tool.run(arg.strip()) # 4. 把工具结果喂回 LLM history.append({"role": "assistant", "content": f"Tool result: {tool_result}"}) final_reply = generate(history) else: final_reply = "I don't know that tool." else: final_reply = response # 5. 持久化最终回复 push_message(user_id, "assistant", final_reply) return final_reply

权衡点：

• 用 Redis 做短期记忆有网络开销；如果延迟敏感，单 worker 部署可以放内存里
• 每轮都写 SQLite 会让数据库膨胀；需要定期清理或归档历史会话

持久化记忆和状态管理

持久化记忆是让 Agent 像个一致助手的关键。我把状态分成三层：

层级	用途	技术选型
临时（每请求）	构造 prompt	Python 对象
短期（最近几轮）	对话上下文	Redis（TTL 24 小时）
长期（用户画像、偏好）	知识库	SQLite + SQLAlchemy

SQLite schema 示例

# models.py
from sqlalchemy import Column, Integer, String, JSON, create_engine
from sqlalchemy.orm import declarative_base, sessionmaker Base = declarative_base() class UserProfile(Base): __tablename__ = "user_profiles" id = Column(Integer, primary_key=True) user_id = Column(String, unique=True, index=True) prefs = Column(JSON) # 比如 {"language": "en", "units": "metric"} engine = create_engine("sqlite:///agents.db", future=True)
SessionLocal = sessionmaker(bind=engine, expire_on_commit=False) def init_db(): Base.metadata.create_all(engine)

Agent 需要长期信息时：

def get_user_prefs(user_id: str) -> dict: with SessionLocal() as db: profile = db.query(UserProfile).filter_by(user_id=user_id).first() return profile.prefs if profile else {}

容易挂的点：

• Redis 淘汰策略可能意外清掉短期记忆；设置合理的 maxmemory-policy
• SQLite 并发写入会锁表；QPS 超过 10 的话考虑 Postgres 或者开 WAL 模式

什么时候可以不要持久化？
如果 Agent 跑的是批处理（比如生成报告），不会跟同一个用户聊第二次，长期层可以直接砍掉。

测试、调试和部署

测 AI Agent 跟测 CRUD 接口不太一样，因为 LLM 回复是非确定性的。我用三层测试：

1. 单元测试：工具逻辑和记忆辅助函数（纯 Python）
2. 集成测试：用 responses 或 pytest-mock mock 掉 OpenAI SDK
3. 端到端测试：用 TestClient 起 FastAPI 应用，跑几个真实 prompt，用便宜模型（比如 gpt-4o-mini）

单元测试示例

def test_redis_memory_roundtrip(): user = "test_user" push_message(user, "user", "Hello") hist = get_history(user) assert len(hist) == 1 assert hist[0]["content"] == "Hello"

Mock LLM

from unittest.mock import patch @patch("openai.ChatCompletion.create")
def test_agent_calls_tool(mock_create): mock_create.return_value = SimpleNamespace( choices=[SimpleNamespace(message=SimpleNamespace(content="<tool:weather>London"))] ) # 注册一个假的天气工具 class FakeWeather: name = "weather" async def run(self, city): return "15°C" register(FakeWeather()) reply = asyncio.run(agent_loop("u1", "What's the weather?")) assert "15°C" in reply

FastAPI 端点

# main.py
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel app = FastAPI() class ChatRequest(BaseModel): user_id: str message: str @app.post("/chat")
async def chat(req: ChatRequest): try: reply = await agent_loop(req.user_id, req.message) return {"reply": reply} except Exception as e: raise HTTPException(status_code=500, detail=str(e))

本地跑：

uvicorn main:app --reload

Docker 部署

# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY . .
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

镜像推送到容器服务，跑在阿里云容器服务或者腾讯云容器上都行。

调试技巧：

• 每次 LLM 调用前把完整消息列表打日志，方便复现问题
• 抓取 OpenAI 响应的 header，里面有 rate limit 信息
• 容器反复重启的话，检查内存限制——模型 payload 在边缘情况下可能超过 2GiB

成本：

• 单次 gpt-4o-mini 调用大约 0.00015 美元。100 QPS 的话一天 13 美元，不算 Redis 和数据库
• logprobs 只在需要细粒度调试时打开，会增加延迟

FAQ

LangChain 能不能不用 Redis？
可以。LangChain 自带内存 buffer，单进程开发完全够用。要水平扩展的时候再切 Redis。

需要 GPU 吗？
用 OpenAI 的托管模型不需要。GPU 成本只在跑本地 LLM（如 Llama-3）时才需要，会增加不少复杂度。

怎么加新工具？
实现 Tool 协议，注册到 register()，在循环里加个检测 <tool:name> 标签的正则就行。核心 LLM 代码不用动。

FastAPI 端点怎么做用户认证？
用 OAuth2 + JWT Token。FastAPI 的 Depends 系统提取 token 里的 user_id 传给 agent_loop 很方便。

总结

• 隔离 Python 环境，锁定版本号，防止 SDK 突然挂掉
• 选框架要看 Agent 复杂度：简单用原生 SDK，中等用 LangChain，复杂用 LangGraph
• 短期记忆（Redis）和长期记忆（SQLite）分开，保持低延迟和数据持久化
• 单元测试和集成测试 mock LLM，端到端测试抓 prompt 级的问题
• Docker 部署到容器平台，监控好 token 成本

把这些模块搭好，从零构建 AI Agent 就不是什么难事了，迭代快，生产环境也 hold 得住。祝开发顺利。