FastAPI 服务化实战：用6大组件把 LangChain Agent 包装成生产级 HTTP API

FastAPI 服务化实战：用6大组件把 LangChain Agent 包装成生产级 HTTP API

承接系列前两篇： ① Python异步编程：从错误示范到asyncio正确姿势 ② LangChain QuickStart 实践：六大核心组件构建心屿树洞最小Agent闭环 系列阅读路线：异步基础 → Agent能力 → API服务化（本篇） → LangGraph状态图生产重构 → 部署上线

1. 问题背景：你一定踩过的 4 类开发痛点

很多同学在前两篇把 Agent 跑通后，一到"接前后端"就频繁碰壁。这一步的问题不是 Agent 本身，而是 Web 框架 + API 契约 + 工程化的组合坑：

本篇目标：用一份可直接复制运行的 FastAPI 完整代码，一次性解决上述 4 类问题，同时把第二篇写的 LangChain Agent 包装成 3 种主流API形态：

• ✅ 一次性对话 POST /api/v1/chat（同步返回）
• ✅ SSE 流式对话 POST /api/v1/chat/stream（打字机体验）
• ✅ WebSocket 双向对话 WS /ws/chat/{thread_id}（实时通信）

2. 前置环境：2 套依赖方案，0 配置即可启动

2.1 最小依赖（推荐新手先跑通这个）

仅需 FastAPI + Uvicorn，无任何 LLM 依赖也能跑，代码内部自动降级 MockAgent：

pip install fastapi uvicorn[standard] pydantic pydantic-settings python-dotenv

2.2 完整依赖（接真实 LangChain Agent + LLM API Key）

把第二篇博客写的 Agent 直接接进来，强烈建议先跑通最小依赖再加：

# 基础
pip install fastapi uvicorn[standard] pydantic pydantic-settings python-dotenv
# LangChain + LangGraph（API版本频繁变更，建议指定以下版本组合）
pip install "langchain-core>=0.3,<0.4" "langgraph>=0.2,<0.3" httpx
# 或用兼容OpenAI协议的包装：
pip install langchain-openai

2.3 环境变量配置（接真实模型时才需要）

Windows PowerShell：

$env:LLM_BASE_URL = "https://api.deepseek.com/v1"
$env:LLM_MODEL = "deepseek-chat"
$env:LLM_API_KEY= "sk-xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"

Linux/macOS Bash：

export LLM_BASE_URL="https://api.deepseek.com/v1"
export LLM_MODEL="deepseek-chat"
export LLM_API_KEY="sk-xxx"

目录结构：

03-fastapi-agent/
├── blog.md ← 本篇文档
├── README.md
└── code/
├── requirements.txt ← 依赖清单（2.1/2.2合集）
└── main.py ← 完整可运行代码（单文件版，180+行）

3. 代码示例：单文件版 FastAPI + Agent 完整闭环

💡 设计原则：单文件可跑 + 分层清晰 + 三级兜底，避免一上来就拆 routers/services/models 把新手劝退。真实项目按 §5 拓展方向拆。

3.0 核心架构总览

前端 (Vue3/Vite:5173)
│ ① POST JSON ② SSE EventSource ③ WS connect
▼
┌─────────────────────────────────────────────────────┐
│ FastAPI App :8000 │
│ ├─ CORS 白名单中间件（解决端口跨域） │
│ ├─ Pydantic 统一请求/响应模型（解决 422） │
│ ├─ 统一异常处理 @app.exception_handler │
│ ├─ run_agent() 入口 ──┬── MockAgent（无依赖兜底） │
│ │ └── LangChain Agent（真实） │
│ └─ 路由层 3 种形态：一次性 / SSE / WS │
└─────────────────────────────────────────────────────┘

3.1 完整可运行代码（复制到 main.py 双击运行）

# -*- coding: utf-8 -*-
"""
运行命令：
A) 开发热重载： uvicorn main:app –reload –port 8000
B) 直接运行： python main.py
启动后打开：
✅ Swagger文档 http://127.0.0.1:8000/docs
✅ 健康检查 http://127.0.0.1:8000/health
"""
import os
import sys
import json
import asyncio
import importlib
from typing import Optional, Any

from fastapi import FastAPI, Request, WebSocket, WebSocketDisconnect, HTTPException, status
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field

第一段：Pydantic 请求/响应契约（解决 422 的第一道防线，所有字段类型显式声明）：

# ============================================================
# §1. 数据契约：Pydantic V2 模型（替代 Flask 手工 if/else 校验）
# ============================================================
class ChatRequest(BaseModel):
user_input: str = Field(..., description="用户输入文本", examples=["最近压力好大想找人聊聊"])
thread_id: str = Field(default="default_user_001", description="会话隔离ID")
stream: bool = Field(default=False, description="是否流式（一次性接口内部字段）")

class ToolCallRecord(BaseModel):
name: str
args: dict
output: str

class ChatResponse(BaseModel):
"""统一响应 Envelope：{success, data, message}，前端统一解析"""
success: bool = True
data: dict = Field(default_factory=lambda: {
"reply": "", "tool_calls": [], "thread_id": ""
})
message: str = "ok"

class ErrorResponse(BaseModel):
"""统一错误响应：把FastAPI默认格式套进标准 envelope"""
success: bool = False
error: str = "UnknownError"
message: str
detail: Optional[Any] = None

第二段：LangChain Agent 可选导入 + MockAgent 兜底（解决依赖缺失崩溃 + API变更）：

# ============================================================
# §2. Agent 入口：try/except ImportError 分层，缺依赖也不崩
# ============================================================
_HAS_AGENT = False
build_agent = None

try:
# 复用第二篇博客写的 06_agent.py（复制到独立目录也会自动降级）
_THIS_DIR = os.path.dirname(os.path.abspath(__file__))
_PREV_DIR = os.path.abspath(os.path.join(_THIS_DIR, "..", "..", "02-langchain-quickstart", "code"))
sys.path.insert(0, _PREV_DIR)
_m06 = importlib.import_module("06_agent")
build_agent = _m06.build_agent
_HAS_AGENT = True
except Exception as e:
print(f"[WARN] LangChain Agent不可用：{type(e).__name__}: {e}，使用MockAgent\\n")

class MockAgent:
"""零配置MockAgent：FastAPI服务0依赖可启动演示"""
_HOTLINES = {
"四川": "028-85422114（华西心理卫生中心）",
"广东": "020-81899120（广州市心理危机干预中心）",
"全国": "400-161-9995（全国心理援助热线24h）",
}
def _province(self, txt):
for c, p in {"成都":"四川","广州":"广东","深圳":"广东","杭州":"浙江"}.items():
if c in txt: return p
return "全国"

def invoke(self, user_input, thread_id):
crisis = any(k in user_input for k in ["自杀","跳楼","不想活","割腕","活着没意义"])
tool_calls = []
if crisis or ("热线" in user_input and "电话" in user_input):
prov = self._province(user_input)
tool_calls.append(ToolCallRecord(
name="query_crisis_hotline",
args={"province": prov, "urgency": "紧急" if crisis else "一般"},
output=self._HOTLINES.get(prov, self._HOTLINES["全国"]),
))
if tool_calls:
reply = (f"非常担心你，请珍惜生命。{tool_calls[0].args['province']}援助热线："
f"{tool_calls[0].output}，即刻自伤冲动请拨120")
elif any(w in user_input for w in ["压力","焦虑","难过","挂科","失恋"]):
reply = "我在认真听，愿意和我具体说说是什么让你这么难受吗？（MockAgent演示）"
else:
reply = "你好，我是心屿树洞AI陪伴者，有什么想聊的吗？（Mock模式）"
return {"reply": reply, "tool_calls": tool_calls, "thread_id": thread_id}

_agent_inst = None
def get_agent():
"""延迟初始化：首次调用才构建，避免启动时崩溃"""
global _agent_inst
if _agent_inst is None:
try:
_agent_inst = build_agent() if (_HAS_AGENT and build_agent) else MockAgent()
except Exception as e:
print(f"[WARN] Agent构建失败降级Mock：{e}")
_agent_inst = MockAgent()
return _agent_inst

def run_agent(user_input: str, thread_id: str):
"""统一调用入口：Mock / LangChain 走同一返回结构"""
agent = get_agent()
if isinstance(agent, MockAgent):
return agent.invoke(user_input, thread_id)

# 真实 LangChain Agent：解析 graph.invoke 结果
try:
from langchain_core.messages import HumanMessage, ToolMessage
result = agent.invoke(
{"messages": [HumanMessage(content=user_input)]},
config={"configurable": {"thread_id": thread_id}},
)
msgs = result["messages"]
reply = msgs[–1].content if hasattr(msgs[–1], "content") else str(msgs[–1])
tool_calls = []
for m in msgs:
for tc in getattr(m, "tool_calls", None) or []:
output = ""
for m2 in msgs:
if isinstance(m2, ToolMessage) and getattr(m2, "tool_call_id", None) == tc["id"]:
output = m2.content if hasattr(m2, "content") else str(m2)
break
tool_calls.append(ToolCallRecord(name=tc["name"], args=tc["args"], output=str(output)))
return {"reply": reply, "tool_calls": tool_calls, "thread_id": thread_id}
except Exception as e:
print(f"[ERROR] Agent调用失败：{e}，降级Mock")
return MockAgent().invoke(user_input, thread_id)

第三段：FastAPI App 初始化 + 中间件 + 三种路由形态：

# ============================================================
# §3. FastAPI App + CORS 中间件（解决跨域问题）
# ============================================================
app = FastAPI(
title="心屿树洞 · AI陪伴 API",
version="1.0.0",
responses={404: {"model": ErrorResponse}, 422: {"model": ErrorResponse}},
)

app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:5173", "http://localhost:3000", "http://localhost:8080"],
allow_credentials=True, # 前端 fetch 要带 credentials:'include'
allow_methods=["*"], # 允许所有HTTP方法（含OPTIONS预检）
allow_headers=["*"], # 允许所有请求头（含Content-Type, Authorization）
)

# ============================================================
# §4. 路由层：一次性 / SSE流式 / WebSocket
# ============================================================
@app.get("/health", tags=["系统"])
def health_check():
"""健康检查：K8s探针/部署脚本必调"""
agent_type = "Mock" if isinstance(get_agent(), MockAgent) else "LangChain"
return {"success": True, "status": "ok", "agent": agent_type}

@app.post("/api/v1/chat", response_model=ChatResponse, tags=["对话"])
async def chat_once(req: ChatRequest):
"""① 一次性对话：最常用，同步返回完整回复"""
try:
result = run_agent(req.user_input, req.thread_id)
return ChatResponse(data=result)
except Exception as e:
return ChatResponse(
success=False, data={"reply": "", "tool_calls": [], "thread_id": req.thread_id},
message=f"Agent调用失败: {type(e).__name__}"
)

@app.post("/api/v1/chat/stream", tags=["对话"])
async def chat_stream(req: ChatRequest):
"""② SSE 流式接口：前端 new EventSource() 或 fetch + ReadableStream
– 消息体按 SSE 标准：以 data: xxx\\n\\n 分隔
– 结束标记固定字符串：data: [DONE]
"""
result = run_agent(req.user_input, req.thread_id)

async def _gen():
if result["tool_calls"]:
yield f"data: {json.dumps({'type':'tools','data':[t.model_dump() for t in result['tool_calls']]}, ensure_ascii=False)}\\n\\n"
await asyncio.sleep(0.05)
for ch in result["reply"]:
yield f"data: {json.dumps({'type':'delta','data':ch}, ensure_ascii=False)}\\n\\n"
await asyncio.sleep(0.02)
yield "data: [DONE]\\n\\n"

return StreamingResponse(_gen(), media_type="text/event-stream")

@app.websocket("/ws/chat/{thread_id}")
async def ws_chat(websocket: WebSocket, thread_id: str):
"""③ WebSocket 双向对话：前端 ws://127.0.0.1:8000/ws/chat/user_001
– 客户端 → 服务端：纯文本消息（用户输入）
– 服务端 → 客户端：JSON（type: delta/done/error）
"""
await websocket.accept()
try:
while True:
data = await websocket.receive_text()
result = run_agent(data, thread_id)
for ch in result["reply"]:
await websocket.send_json({"type": "delta", "data": ch})
await asyncio.sleep(0.02)
await websocket.send_json({"type": "done", "tool_calls": [t.model_dump() for t in result["tool_calls"]]})
except WebSocketDisconnect:
print(f"[WS] 会话 {thread_id} 断开")
except Exception as e:
await websocket.send_json({"type": "error", "message": str(e)})

第四段：统一异常处理（FastAPI 默认 422/404 格式套回标准 Envelope）：

# ============================================================
# §5. 统一异常处理：响应格式始终一致
# ============================================================
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
"""把 FastAPI 默认 422 格式转成我们的标准 envelope"""
return JSONResponse(status_code=422, content=ErrorResponse(
error="ValidationError",
message="请求参数校验失败，请检查Swagger文档里的字段名和类型",
detail=exc.errors(), # 详细定位哪个字段错了
).model_dump())

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(status_code=exc.status_code, content=ErrorResponse(
error=f"HTTP{exc.status_code}", message=exc.detail
).model_dump())

# ============================================================
# 启动入口：python main.py 直接运行
# ============================================================
if __name__ == "__main__":
import uvicorn
print("\\n" + "="*60)
print("🚀 心屿树洞 FastAPI 服务")
print(" Swagger: http://127.0.0.1:8000/docs")
print(" ReDoc: http://127.0.0.1:8000/redoc")
print(" 健康检查: http://127.0.0.1:8000/health")
print("="*60 + "\\n")
uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=False)

3.2 启动与调用验证

① 启动服务：

cd code
uvicorn main:app –reload –port 8000
# 或直接：python main.py

② 用 Swagger 验证（最推荐）：浏览器打开 http://127.0.0.1:8000/docs → 选 /api/v1/chat → Try it out → Execute

③ 用 curl 验证：

# 一次性对话（Windows PowerShell 把引号改成双引号）
curl -X POST http://127.0.0.1:8000/api/v1/chat `
-H "Content-Type: application/json" `
-d '{"user_input":"最近期末考试压力好大，挂了两门课","thread_id":"user_test_001"}'

# 危机场景
curl -X POST http://127.0.0.1:8000/api/v1/chat `
-H "Content-Type: application/json" `
-d '{"user_input":"我觉得活着没意义想跳楼，现在在成都上学"}'

预期输出：

{
"success": true,
"data": {
"reply": "非常担心你，请珍惜生命。四川援助热线：028-85422114…",
"tool_calls": [{
"name": "query_crisis_hotline",
"args": {"province": "四川", "urgency": "紧急"},
"output": "028-85422114（华西心理卫生中心）"
}],
"thread_id": "user_test_001"
},
"message": "ok"
}

4. 报错复盘：Top 5 高频报错 + 定位技巧

💡 下面 5 类报错是我在项目重构 + 读者群答疑中积累的最高频案例，每一条都带"复现路径"和"源码级根因 + 解决方案"，配合 §3 代码里故意埋的两个坑位对照看效果最好。

🚨 报错 1：接口 404 反复排查无效

复现路径：

# 明明在代码里写了 @app.post("/api/v1/chat2/ping")，调它返回 404
curl -X POST http://127.0.0.1:8000/api/v1/chat2/ping
# → {"detail":"Not Found"}

根因（按出现概率排序）：

解决方案（二选一并全局强制执行）：

# 方案A（推荐）：Router只写业务名，version前缀统一在 main 注册时加
# routers/chat.py
router = APIRouter(prefix="/chat", tags=["对话"])
# main.py
app.include_router(router, prefix="/api/v1") # → 最终路径 /api/v1/chat

# 方案B：Router写完整前缀，main注册时不再加
# routers/chat.py
router = APIRouter(prefix="/api/v1/chat", tags=["对话"])
# main.py
app.include_router(router) # → 不再传 prefix

快速定位技巧：启动后直接访问 Swagger /docs，看左侧列表里有没有你这个接口——没有就是根本没注册上，先排查导入和注册；有就看路径是不是和你 curl 的一样。

🚨 报错 2：422 Unprocessable Entity（参数校验失败）

复现路径A：

curl http://127.0.0.1:8000/api/v1/qa/audit/stats
# → 422 ValidationError，字段提示 log_id 不是有效 integer

复现路径B：

curl -X POST http://127.0.0.1:8000/api/v1/chat `
-d '{"user_message":"test"}' # ← 字段名应该是 user_input，故意写错
# → 422

根因：

A类解决方案（路由顺序）：静态路由永远写在动态路由之前：

# ❌ 错误（§3代码坑位1，故意这么写的）
router = APIRouter(prefix="/api/v1")

@router.get("/qa/audit/{log_id:int}") # ← 写在前，先匹配
def audit_detail(log_id: int): ...

@router.get("/qa/audit/stats") # ← 写在后，永远匹配不到
def audit_stats(): ...

# ✅ 正确
@router.get("/qa/audit/stats") # 1. 先写静态
def audit_stats(): ...

@router.get("/qa/audit/{log_id:int}") # 2. 后写动态
def audit_detail(log_id: int): ...

B类解决方案：

🚨 报错 3：CORS 跨域失败（简单请求通，POST JSON不通）

复现路径：

• 浏览器控制台：Access to fetch at 'http://127.0.0.1:8000/api/v1/chat' from origin 'http://localhost:5173' has been blocked by CORS policy
• 同时 Network 栏显示 OPTIONS 请求返回 405 / 没有 CORS 头

根因排序：

正确配置（对应 §3 代码）：

app.add_middleware(
CORSMiddleware,
allow_origins=["http://localhost:5173", "http://localhost:3000"], # ✅ 具体origin，不要*
allow_credentials=True, # ✅ 带Cookie/Authorization时打开
allow_methods=["*"], # ✅ 允许所有方法（自动包含OPTIONS预检）
allow_headers=["*"], # ✅ 允许Content-Type等自定义头
)

🚨 报错 4：端口占用 10048 / Address already in use

复现路径：uvicorn main:app –port 8000 启动失败，提示端口被占。

Windows 一键查杀（我日常常用的两条）：

# 方案A：精确杀占用8000的进程
Get-NetTCPConnection –LocalPort 8000 –ErrorAction SilentlyContinue |
ForEach-Object { Stop-Process –Id $_.OwningProcess –Force }

# 方案B：用 netstat + taskkill 老命令（兼容低版本PowerShell）
netstat –ano | findstr :8000
# 输出最后一列是PID，假设是 24680
taskkill /PID 24680 /F

🚨 报错 5：ImportError: cannot import name 'create_react_agent' + 启动崩溃

复现路径：python main.py 时 import 06_agent.py 过程中抛出异常 → 整个 FastAPI app 起不来。

根因：LangGraph V1.0 把 create_react_agent 从 langgraph.prebuilt 移到了 langchain.agents，同时参数名从 state_modifier 改成了 instructions。如果你 pip install langgraph 拉到了最新版，旧版代码 import 就炸。

解决方案：

5. 拓展延伸：从 Demo 到生产的 5 个进阶方向

§3 单文件版能跑，但真实生产落地需要按下面方向分层拆：

6. 文末互动 & 系列预告

📝 互动小任务（任选其一做，进步最快）：

🔜 第四篇预告（LangGraph 生产重构）：

前三篇我们完成了异步基础 → Agent能力 → API服务化的最小闭环，但真实业务流程（倾诉 → 情绪分析 → 危机分级 → 转人工 → 推荐测评 → 存档）不是单轮 ReAct Agent 能描述的。

下一篇我会用 LangGraph StateGraph 状态图把这个完整多分支流程显式画出来，每一个状态节点对应一个工具/模型调用，条件边处理"高危转人工、中危推荐测评、低危继续对话"，真正把 LangChain Agent 从 Demo 级提升到可落地业务级。

如果这篇对你有用，欢迎 👍点赞 / ⭐收藏 / 💬评论，你的反馈是我更新的最大动力！