Day 7:FastAPI部署——Agent上线
昨天的问题
Agent已经完整了——但只能在你自己的终端里跑。别人用不了,前端接不上,没法演示给老板看。今天是让它对世界开放的日子。
前6天的回顾
你的Agent已经具备了完整的能力:
- ✅ 骨架(Day 1)——能跑的最简程序
- ✅ Prompt + MockLLM(Day 2)——有性格的客服
- ✅ RAG知识库(Day 3)——能查公司政策
- ✅ 三个工具(Day 4)——能查订单、算退款
- ✅ 多轮对话记忆(Day 5)——LangGraph + 检查点
- ✅ MCP标准化(Day 6)——工具变成标准服务
但这一切都只在你自己的终端里跑。今天,让它对世界开放。
今天做什么
用FastAPI把Day 5的LangGraph Agent包装成RESTful API。完成之后:
- 前端可以发HTTP请求调用Agent
- 每个用户有独立的
thread_id,互不干扰 - 自动生成API文档(Swagger UI)
- 一条命令启动/停止
你需要安装
bash
pip install fastapi uvicorn📁 今天的代码目录
customer_service_agent/
├── day1~6.py (前6天代码不动)
├── tools.py
├── tools_server.py
├── main.py ← 新增!FastAPI入口
└── test_client.py ← 测试脚本python
# main.py — FastAPI 包装 Day5 的 LangGraph Agent
import sys
sys.path.insert(0, '.') # 确保能导入Day5的模块
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage
import uuid
# ===== Day5的Agent(假设保存在 day5_agent.py 中)=====
# 实际使用时,把Day5的 graph 和 memory 导入
from day5 import app as agent # Day5编译好的StateGraph
# ===== FastAPI 应用 =====
api = FastAPI(
title="智能客服Agent API",
description="7天从零搭建的智能客服系统。支持查订单、退货政策、多轮对话。",
version="1.0.0"
)
# 允许跨域(前端调用需要)
api.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_methods=["*"],
allow_headers=["*"],
)
# ===== 请求/响应模型 =====
class ChatRequest(BaseModel):
"""用户发来的消息"""
message: str = Field(..., description="用户消息", example="ORD001")
user_id: str = Field(default="guest", description="用户ID,用于区分会话", example="user-alice")
class ChatResponse(BaseModel):
"""Agent的回复"""
answer: str = Field(..., description="Agent的回答")
session_id: str = Field(..., description="会话ID")
# ===== API接口 =====
@api.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
"""
用户发送消息,Agent回复。
同一个 user_id 的多次请求共享会话上下文(多轮对话记忆)。
不同 user_id 完全隔离。
"""
try:
# thread_id 让不同用户的会话隔离
config = {"configurable": {"thread_id": request.user_id}}
# 调用Day5的LangGraph Agent
result = agent.invoke(
{"messages": [HumanMessage(content=request.message)]},
config
)
# 提取最后一条AI消息
last_message = result["messages"][-1]
answer = last_message.content if hasattr(last_message, 'content') else str(last_message)
return ChatResponse(
answer=answer,
session_id=request.user_id
)
except Exception as e:
raise HTTPException(status_code=500, detail=f"Agent处理出错: {str(e)}")
@api.get("/health")
async def health_check():
"""健康检查"""
return {"status": "ok", "service": "智能客服Agent"}
@api.get("/")
async def root():
"""根路径"""
return {
"service": "智能客服Agent API",
"version": "1.0.0",
"docs": "/docs",
"endpoints": {
"chat": "POST /chat",
"health": "GET /health"
}
}
# ===== 启动 =====
if __name__ == "__main__":
import uvicorn
print("🚀 智能客服 Agent 启动中...")
print("📍 访问 http://localhost:8000 查看服务信息")
print("📖 访问 http://localhost:8000/docs 查看API文档")
uvicorn.run(api, host="0.0.0.0", port=8000)运行和测试
bash
# 启动服务
python main.py
# 输出:
# 🚀 智能客服 Agent 启动中...
# 📍 访问 http://localhost:8000 查看服务信息
# 📖 访问 http://localhost:8000/docs 查看API文档打开浏览器,访问 http://localhost:8000/docs——你会看到自动生成的Swagger文档,可以直接在网页上测试API。
命令行测试及预期结果:
bash
# 健康检查
curl http://localhost:8000/health
# → {"status":"ok","service":"智能客服Agent"}
# Alice的第一轮对话
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message":"ORD001","user_id":"alice"}'
# → {"answer":"📦 订单ORD001 | 无线耳机 | ¥299 | 运输中","session_id":"alice"}
# Alice的第二轮——Agent记得上一轮!
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message":"那这个能退货吗?","user_id":"alice"}'
# → {"answer":"订单ORD001:状态为'运输中',请先确认收货。","session_id":"alice"}
# Bob完全独立的会话
curl -X POST http://localhost:8000/chat \
-H "Content-Type: application/json" \
-d '{"message":"退货需要什么条件?","user_id":"bob"}'
# → {"answer":"根据公司政策:7天内可退货...","session_id":"bob"}🎬 最终效果演示
启动服务后,模拟一段真实的多轮对话场景——Alice在查订单,Bob在问政策:
============================================================
🚀 智能客服Agent v1.0 已上线
📍 http://localhost:8000/docs
============================================================
——— Alice的会话(thread_id: alice)———
👤 Alice: 你好
🤖 小智: 您好!我是小智。可以帮您查订单、退货政策。有什么需要?
👤 Alice: ORD001
🤖 小智: 📦 订单ORD001 | 无线耳机 | ¥299 | 运输中
👤 Alice: 那这个能退货吗?
🤖 小智: ⏳ 订单当前状态为'运输中'。请先确认收货后再申请退货。
👤 Alice: 好的,那退款流程是什么?
🤖 小智: 在"我的订单"页面点击"申请退款",客服24小时内审核。
审核通过后3-5个工作日到账。
——— Bob的会话(thread_id: bob,完全独立)———
👤 Bob: 退货需要什么条件?
🤖 小智: 根据公司政策:顾客收到商品后7天内可申请退货。
商品需保持原包装,邮费由公司承担。
👤 Bob: ORD002
🤖 小智: 📦 ORD002 | 手机壳 | ¥39 | 已签收
👤 Bob: 这个能退吗?
🤖 小智: ❌ 订单ORD002已签收15天,超过了7天退货期,无法退货。关键证明:Alice和Bob的对话完全隔离。Alice问"这个"指向ORD001,Bob问"这个"指向ORD002——各记各的,互不干扰。多用户并发场景下,7天搭建的Agent系统运行稳定。
你学到了什么
FastAPI是最简单的Agent部署方案。 几行装饰器就把Agent变成了HTTP服务。而且自动生成API文档——你不需要额外写任何前端文档。
thread_id 是生产级的多用户隔离方案。 Alice和Bob的对话完全独立,互不干扰。同一个用户多次请求共享上下文——这就是真实客服系统的运行方式。
7天项目总结
| 天 | 文件 | 增量 | 核心概念 |
|---|---|---|---|
| Day 1 | day1.py | 10行 | Agent骨架,先跑通 |
| Day 2 | day2.py | +40行 | Prompt模板,MockLLM |
| Day 3 | day3.py | +50行 | RAG,文档切分,向量检索 |
| Day 4 | day4.py | +50行 | @tool装饰器,工具调用 |
| Day 5 | day5.py | +80行 | StateGraph,MemorySaver,多轮对话 |
| Day 6 | tools_server.py | +60行 | MCP Server,工具标准化 |
| Day 7 | main.py | +60行 | FastAPI,/docs,多用户 |
总计约350行代码,7天渐进式搭建。 你从零开始,没有跳过任何一步。每天只加一点,但每一天的代码都是完整可运行的。
现在打开 http://localhost:8000/docs,你有一个真正上线的Agent系统了。
🎉 项目一完成。你的智能客服Agent已经上线运行。

