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][追加处理]}# 构建工作流graphStateGraph(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)graphgraph.compile()# 编译为可运行对象appFastAPI(titleMy LangServer,version0.1.0,description暴露 LangGraph explain为 REST API,)add_routes(app,graph,path/explain,input_typeAgentState,playground_typedefault)app.get(/hello)asyncdefhello():returnResponse(hello, world)if__name____main__:importuvicorn uvicorn.run(app,host0.0.0.0,port8000)运行以上服务并在浏览器里请求:http://localhost:8000/explain/playground/ 进行测试一行代码部署为 API:add_routes(app, graph, path“/explain”, input_typeAgentState, playground_type“default”)启动后自动获得POST /explain/invoke → 同步调用 POST /explain/stream → 流式返回SSE POST /explain/batch → 批量调用 GET /explain/playground/ → 内置测试页面核心作用:没有 LangServe自己写 FastAPI 路由 → 手动处理输入输出 → 手动实现流式 → 手动写文档有 LangServeLangChain/LangGraph 写好 → add_routes() → 自动生成 invoke/stream/batch/playground