fromfastapiimportFastAPIfromfastapi.responsesimportResponsefromlangserveimportadd_routesfromlanggraph.graphimportStateGraph,ENDfromtypingimportTypedDict,List# 定义状态结构classAgentState(TypedDict):input:stroutput:List[str]# 创建节点defnode1(state:AgentState):return{"output":[f"处理:{state['input']}"]}defnode2(state:AgentState):return{"output":state["output"]+["追加处理"]}# 构建工作流graph=StateGraph(AgentState)graph.add_node("node1",node1)graph.add_node("node2",node2)graph.set_entry_point("node1")graph.add_edge("node1","node2")graph.add_edge("node2",END)graph=graph.compile()# 编译为可运行对象app=FastAPI(title="My LangServer",version="0.1.0",description="暴露 LangGraph explain为 REST API",)add_routes(app,graph,path="/explain",input_type=AgentState,playground_type="default")@app.get("/hello")asyncdefhello():returnResponse("hello, world")if__name__=="__main__":importuvicorn uvicorn.run(app,host="0.0.0.0",port=8000)运行以上服务并在浏览器里请求:http://localhost:8000/explain/playground/ 进行测试
一行代码部署为 API:
add_routes(app, graph, path=“/explain”, input_type=AgentState, playground_type=“default”)
启动后自动获得：
POST /explain/invoke → 同步调用 POST /explain/stream → 流式返回（SSE） POST /explain/batch → 批量调用 GET /explain/playground/ → 内置测试页面核心作用:
- 没有 LangServe：自己写 FastAPI 路由 → 手动处理输入输出 → 手动实现流式 → 手动写文档
- 有 LangServe：LangChain/LangGraph 写好 → add_routes() → 自动生成 invoke/stream/batch/playground
