langGraph--2--langServe+langGraph示例

from fastapi import FastAPI
from fastapi.responses import Response
from langserve import add_routes
from langgraph.graph import StateGraph, END
from typing import TypedDict, List# 定义状态结构
class AgentState(TypedDict):input: stroutput: List[str]# 创建节点
def node1(state: AgentState):return {"output": [f"处理: {state['input']}"]}def node2(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 workflow 为 REST API",
)add_routes(app,graph,path="/workflow",input_type=AgentState,playground_type="default"
)@app.get("/hello")
async def hello():return Response("hello, world")if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

运行以上服务并在浏览器里请求:http://localhost:8000/workflow/playground/ 进行测试