LangGraph 写完了怎么对外提供服务？我用 FastAPI 把它包成了一套能上线的接口

agent-service/
├── main.py          # FastAPI 入口，只管路由
├── graph.py         # 图的全部定义（State / 节点 / 边）
├── deps.py          # checkpointer、LLM 实例
└── schemas.py       # Pydantic 请求响应模型

# graph.py
from typing import TypedDict, Annotated
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
from langgraph.types import interrupt, StreamWriter, Command
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage
import anyio

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)


class TicketState(TypedDict):
    messages: Annotated[list, add_messages]   # 追加历史，别覆盖
    intent: str                               # refund / logistics / consult
    amount: float                             # 退款金额
    retry_count: int                          # 重写次数，防死循环
    approved: bool                            # 人工审批结果
    final_answer: str


# ① 意图识别 + 抽金额
async def classify(state: TicketState) -> dict:
    text = state["messages"][-1].content
    prompt = f"把工单归到 refund/logistics/consult 之一，只输出一个词。\n工单：{text}"
    intent = (await llm.ainvoke(prompt)).content.strip().lower()
    if intent not in ("refund", "logistics", "consult"):
        intent = "consult"          # 兜底，别让图抛异常
    amount = 0.0
    for tok in text.replace("元", " ").split():
        try:
            amount = float(tok)
            break
        except ValueError:
            continue
    return {"intent": intent, "amount": amount}


# ② 退款：超过阈值就挂起等人
async def handle_refund(state: TicketState, writer: StreamWriter) -> dict:
    if state["amount"] > 500 and not state.get("approved"):
        writer({"step": "waiting_approval", "msg": "金额超阈值，已推送人工审批"})
        decision = interrupt({
            "reason": f"工单涉及 {state['amount']} 元退款，超过 500 元阈值",
            "amount": state["amount"],
        })
        return {"approved": bool(decision.get("approved"))}

    writer({"step": "refunding", "msg": "正在生成退款答复"})
    prompt = f"用户申请退款 {state['amount']} 元，给出专业、简短的客服回复。"
    return {"final_answer": (await llm.ainvoke(prompt)).content}


# ③ 物流查询（同步 SDK 用线程池包一层）
async def handle_logistics(state: TicketState) -> dict:
    order_id = extract_order_id(state["messages"][-1].content)
    tracking = await anyio.to_thread.run_sync(query_tracking_api, order_id)
    prompt = f"物流状态：{tracking}。用客服口吻向用户说明。"
    return {"final_answer": (await llm.ainvoke(prompt)).content}


# ④ 技术咨询
async def handle_consult(state: TicketState) -> dict:
    docs = await anyio.to_thread.run_sync(retrieve_kb, state["messages"][-1].content)
    prompt = f"参考资料：{docs}\n回答：{state['messages'][-1].content}"
    return {"final_answer": (await llm.ainvoke(prompt)).content}


# ⑤ 质检：不合格重写，超三次转人工
def quality_check(state: TicketState) -> str:
    ans = state.get("final_answer", "")
    if state["retry_count"] >= 3:
        return "to_human"
    if len(ans) < 15 or "不知道" in ans:
        return "regenerate"
    return "pass"


async def regenerate(state: TicketState) -> dict:
    prompt = f"上一版回答太简略，补充细节重写。原始问题：{state['messages'][0].content}"
    return {
        "final_answer": (await llm.ainvoke(prompt)).content,
        "retry_count": state["retry_count"] + 1,
    }


async def to_human(state: TicketState) -> dict:
    return {"final_answer": "已转交人工客服，预计 30 分钟内联系您。"}


def build_graph(checkpointer):
    b = StateGraph(TicketState)
    b.add_node("classify", classify)
    b.add_node("refund", handle_refund)
    b.add_node("logistics", handle_logistics)
    b.add_node("consult", handle_consult)
    b.add_node("regenerate", regenerate)
    b.add_node("human", to_human)

    b.add_edge(START, "classify")
    b.add_conditional_edges(
        "classify",
        lambda s: s["intent"],
        {"refund": "refund", "logistics": "logistics", "consult": "consult"},
    )
    # 三条线都汇到质检
    for n in ("refund", "logistics", "consult"):
        b.add_conditional_edges(
            n, quality_check,
            {"regenerate": n, "pass": END, "to_human": "human"},
        )
    b.add_edge("human", END)
    return b.compile(checkpointer=checkpointer)

# main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from fastapi.responses import StreamingResponse
from pydantic import BaseModel
from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
from langgraph.types import Command
from graph import build_graph
import json, asyncio

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 演示用 SQLite；生产换 AsyncPostgresSaver
    cp = AsyncSqliteSaver.from_conn_string("tickets.db")
    app.state.graph = build_graph(cp)
    app.state.locks = {}            # 每个会话一把锁，后面讲
    yield
    # 关服务时释放连接

app = FastAPI(title="售后工单 Agent 服务", lifespan=lifespan)


class TicketIn(BaseModel):
    user_id: str
    text: str

class ApproveIn(BaseModel):
    approved: bool
    comment: str | None = None


def cfg(thread_id: str):
    return {"configurable": {"thread_id": thread_id}}


def get_lock(app, thread_id: str) -> asyncio.Lock:
    return app.state.locks.setdefault(thread_id, asyncio.Lock())


@app.post("/tickets")
async def create_ticket(body: TicketIn):
    thread_id = f"{body.user_id}-{int(asyncio.get_running_loop().time() * 1000)}"
    return {"thread_id": thread_id}

def sse(event: str, data: dict) -> str:
    return f"event: {event}\ndata: {json.dumps(data, ensure_ascii=False)}\n\n"


@app.get("/tickets/{thread_id}/stream")
async def stream_ticket(thread_id: str, text: str):
    g = app.state.graph
    config = cfg(thread_id)

    async def gen():
        yield ": connected\n\n"        # 先占住连接，顺带防代理缓冲
        try:
            async for mode, chunk in g.astream(
                {"messages": [{"role": "user", "content": text}],
                 "retry_count": 0, "approved": False},
                config,
                stream_mode=["messages", "custom", "updates"],
            ):
                if mode == "messages":
                    token, meta = chunk
                    if token.content:
                        yield sse("token", {"text": token.content})
                elif mode == "custom":
                    # 节点里 writer({...}) 推出来的进度
                    yield sse("step", chunk)
                elif mode == "updates":
                    node = list(chunk.keys())[0]
                    yield sse("node_done", {"node": node})

            # 跑完之后检查一下是不是卡在人工审批了
            snap = await g.aget_state(config)
            for task in snap.tasks:
                if task.interrupts:
                    yield sse("need_approval", task.interrupts[0].value)
            yield sse("done", {"thread_id": thread_id})
        except Exception as e:
            yield sse("error", {"msg": str(e)})

    return StreamingResponse(
        gen(),
        media_type="text/event-stream",
        headers={
            "Cache-Control": "no-cache",
            "X-Accel-Buffering": "no",   # 关键！否则 Nginx 会把流攒起来
        },
    )

@app.post("/tickets/{thread_id}/approve")
async def approve(thread_id: str, body: ApproveIn):
    g = app.state.graph
    config = cfg(thread_id)

    # 先确认这个会话确实卡在审批上
    snap = await g.aget_state(config)
    if not snap.tasks or not any(t.interrupts for t in snap.tasks):
        raise HTTPException(409, "该工单当前不在审批等待中")

    result = await g.ainvoke(
        Command(resume={"approved": body.approved, "comment": body.comment}),
        config,
    )
    return {"final_answer": result.get("final_answer"), "approved": body.approved}

@app.get("/tickets/{thread_id}/state")
async def get_state(thread_id: str):
    snap = await app.state.graph.aget_state(cfg(thread_id))
    return {
        "values": snap.values,
        "next": snap.next,                       # 下一步要走哪些节点
        "waiting": bool(snap.tasks and any(t.interrupts for t in snap.tasks)),
    }

lock = get_lock(app, thread_id)
if lock.locked():
    raise HTTPException(429, "该会话正在处理中，请稍后再试")
async with lock:
    result = await g.ainvoke(Command(resume=...), config)