把智能体做成服务

前面几章的智能体都在脚本里 run() 一下、打印结果。要接进网页/小程序/IM，得把"跑一次 Agent"变成一个 HTTP 接口：客户端 POST 问题、拿到回答。本章用 FastAPI 把第 18 章的 LangGraph 智能体封装成 /chat 服务——支持普通返回与SSE 流式返回两种模式，含 pydantic 请求模型、CORS 与统一错误码，代码完整可跑。

安装依赖

pip install fastapi uvicorn langgraph langchain-openai pydantic

agent.py：第 18 章图的简写版

有第 18 章现成的图就直接 import；没有的话用下面这份等价的最小图（一个带 add 工具的 ReAct）：

# agent.py —— 简写自第 18 章；结构相同，可整体替换为你的 graph
import os
from typing import Annotated, TypedDict
from langchain_core.messages import HumanMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """计算两个整数的和"""
    return a + b

model = ChatOpenAI(model=os.getenv("LLM_MODEL", "deepseek-chat"),
                   api_key=os.getenv("LLM_API_KEY"),
                   base_url=os.getenv("LLM_BASE_URL"))
model_with_tools = model.bind_tools([add])

class State(TypedDict):
    messages: Annotated[list, add_messages]

def agent(state):
    return {"messages": [model_with_tools.invoke(state["messages"])]}

def should_continue(state):
    last = state["messages"][-1]
    return "tools" if getattr(last, "tool_calls", None) else END

builder = StateGraph(State)
builder.add_node("agent", agent)
builder.add_node("tools", ToolNode([add]))
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, {"tools": "tools", END: END})
builder.add_edge("tools", "agent")
graph = builder.compile()

app.py：/chat 与 /chat/stream

# app.py
import json
from fastapi import FastAPI, HTTPException
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import StreamingResponse
from pydantic import BaseModel, Field
from langchain_core.messages import HumanMessage
from agent import graph

app = FastAPI(title="Agent 服务")
app.add_middleware(CORSMiddleware, allow_origins=["*"],       # 前端域名按需收紧
                   allow_methods=["*"], allow_headers=["*"])

class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=4000, description="用户输入")
    session_id: str | None = None       # 有则复用会话，无则新建
    stream: bool = False                # 简化版可忽略；下面用不同路由实现

class ChatResponse(BaseModel):
    reply: str
    session_id: str

def new_session_id() -> str:
    import uuid
    return str(uuid.uuid4())[:8]

def raise_api(status: int, code: str, message: str):
    # 统一错误体：{"error": {"code": ..., "message": ...}}
    raise HTTPException(status_code=status, detail={"code": code, "message": message})

@app.get("/healthz")
def healthz():
    return {"status": "ok"}

@app.post("/chat", response_model=ChatResponse)
def chat(req: ChatRequest):
    """同步接口：等 Agent 全部跑完再一次性返回。"""
    sid = req.session_id or new_session_id()
    try:
        out = graph.invoke({"messages": [HumanMessage(req.message)]})
        reply = next((m.content for m in reversed(out["messages"]) if m.content), "")
    except Exception as e:
        raise_api(500, "AGENT_ERROR", f"Agent 执行失败：{e}")
    return ChatResponse(reply=reply, session_id=sid)

def sse_events(message: str):
    """同步生成器：LangGraph 逐 token 出内容，包成 SSE 事件。"""
    for _chunk, meta in graph.stream(
        {"messages": [HumanMessage(message)]}, stream_mode="messages"):
        piece = _chunk.content
        if not piece:
            continue
        yield f"data: {json.dumps({'delta': piece}, ensure_ascii=False)}\n\n"
    yield "data: [DONE]\n\n"      # 前端收到即结束

@app.post("/chat/stream")
def chat_stream(req: ChatRequest):
    """SSE 流式接口：一边生成一边推给客户端。"""
    if req.message.startswith("禁止"):       # 演示错误分支；真实校验见第 19 章
        raise_api(400, "BAD_REQUEST", "请求被拒绝")
    return StreamingResponse(sse_events(req.message), media_type="text/event-stream")

说明两点：同步节点跑模型会阻塞事件循环，上面两个端点都写成 def（非 async），FastAPI 会自动丢线程池执行，互不卡死；SSE 用 data: ...\n\n 分隔事件，浏览器 EventSource / 前端 fetch 流都能解析。

启动与验证

uvicorn app:app --host 0.0.0.0 --port 8000 --reload

另开终端验证：

curl http://127.0.0.1:8000/healthz
# 输出：{"status":"ok"}

curl -X POST http://127.0.0.1:8000/chat \
  -H "Content-Type: application/json" \
  -d '{"message": "25 加 17 等于多少？"}'
# 输出：{"reply":"25 加 17 等于 42。","session_id":"3f9a2c1e"}

curl -N -X POST http://127.0.0.1:8000/chat/stream \
  -H "Content-Type: application/json" -d '{"message":"写一句自我介绍"}'
# 输出（逐条推送）：data: {"delta": "你好"}  …  data: [DONE]

浏览器侧消费：new EventSource 不支持 POST，可用 fetch + ReadableStream 逐行读 data: 行，或换 GET + query 参数（把 message 放 query，EventSource 即可直接用，详见第 21 章末尾）。

错误码约定

HTTP	code	含义	处理建议
400	BAD_REQUEST	参数非法/被拦截	客户端修正输入
404	NOT_FOUND	会话不存在等	检查 session_id
429	RATE_LIMITED	触发了限流/配额	退避重试
500	AGENT_ERROR	模型或工具异常	看服务日志，可让客户端重试
所有错误统一为 {"error": {"code": ..., "message": ...}}，前端按 code 分支展示，不猜 HTTP 状态之外的细节。

多轮会话怎么存

示例里没接会话历史，每请求都是新对话。要记住上下文，用第 20 章的做法：compile(checkpointer=MemorySaver())，把 session_id 映射成 thread_id 传进 config，同 session 的后续请求自动带上历史。生产环境把 MemorySaver 换成 Redis/数据库 checkpointer，再给 /chat 加鉴权与限流（可参考第 29 章容器化一起上线）。 小结：FastAPI 封装智能体 = 一个 graph.invoke 的同步端点 + 一个把 graph.stream(stream_mode="messages") 逐 token 包成 SSE 的流式端点，入参出参用 pydantic 建模、CORS 放行前端、错误统一成 code/message 结构；端点为普通 def 时 FastAPI 自动走线程池，不会阻塞事件循环。