Skip to content

Day 7:FastAPI部署——Agent上线

昨天的问题

Agent已经完整了——但只能在你自己的终端里跑。别人用不了,前端接不上,没法演示给老板看。今天是让它对世界开放的日子。

前6天的回顾

你的Agent已经具备了完整的能力:

  1. 骨架(Day 1)——能跑的最简程序
  2. Prompt + MockLLM(Day 2)——有性格的客服
  3. RAG知识库(Day 3)——能查公司政策
  4. 三个工具(Day 4)——能查订单、算退款
  5. 多轮对话记忆(Day 5)——LangGraph + 检查点
  6. 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 1day1.py10行Agent骨架,先跑通
Day 2day2.py+40行Prompt模板,MockLLM
Day 3day3.py+50行RAG,文档切分,向量检索
Day 4day4.py+50行@tool装饰器,工具调用
Day 5day5.py+80行StateGraph,MemorySaver,多轮对话
Day 6tools_server.py+60行MCP Server,工具标准化
Day 7main.py+60行FastAPI,/docs,多用户

总计约350行代码,7天渐进式搭建。 你从零开始,没有跳过任何一步。每天只加一点,但每一天的代码都是完整可运行的。

现在打开 http://localhost:8000/docs,你有一个真正上线的Agent系统了。


🎉 项目一完成。你的智能客服Agent已经上线运行。