Машина состояний LangGraph: оркестрация сложных AI-агентов на практике

from langgraph.graph import StateGraph, END from typing import TypedDict class AgentState(TypedDict): messages: list iteration: int def node_a(state): return {"messages": ["Node A done"]} def node_b(state): return {"messages": ["Node B done"]} # 构建图 graph = StateGraph(AgentState) graph.add_node("node_a", node_a) graph.add_node("node_b", node_b) # 固定边 graph.add_edge("node_a", "node_b") graph.add_edge("node_b", END) # 编译并运行 app = graph.compile() result = app.invoke({"messages": ["start"], "iteration": 0}) print(result)

from langgraph.graph import StateGraph, END from typing import TypedDict class CodeReviewState(TypedDict): code: str messages: list approved: bool revision_count: int max_revisions: int = 3 def coding_node(state): """生成代码""" return {"messages": ["Code generated"]} def review_node(state): """审查代码，模拟通过/失败""" revision_count = state["revision_count"] if revision_count < 2: return {"approved": False, "revision_count": revision_count + 1} return {"approved": True, "revision_count": revision_count} def revise_node(state): """修复问题""" return {"messages": [f"Revision {state['revision_count']} applied"]} def should_continue(state): """条件分支：审查通过则结束，否则返回重写""" if state["approved"]: return "END" if state["revision_count"] >= state["max_revisions"]: return "END" # 超过最大迭代次数，强制结束 return "revise" # 构建图 graph = StateGraph(CodeReviewState) graph.add_node("code", coding_node) graph.add_node("review", review_node) graph.add_node("revise", revise_node) graph.add_edge("code", "review") graph.add_conditional_edges( "review", should_continue, {"revise": "revise", "END": END} ) graph.add_edge("revise", "code") # 修完后重新生成代码 app = graph.compile() result = app.invoke({ "code": "def add(a,b): return a+b", "messages": [], "approved": False, "revision_count": 0 })

from langgraph.graph import MessagesState def supervisor_node(state): """主管分析任务，决定分派给哪个Agent""" messages = state["messages"] last_msg = messages[-1]["content"] if "код" in last_msg.lower(): return {"next": "coder"} elif "данные" in last_msg.lower() or "анализ" in last_msg.lower(): return {"next": "analyst"} elif "поиск" in last_msg.lower() or "найти" in last_msg.lower(): return {"next": "researcher"} return {"next": "END"} def coder_node(state): return {"messages": [{"role": "assistant", "content": "Код написан"}]} def analyst_node(state): return {"messages": [{"role": "assistant", "content": "Анализ выполнен"}]} def researcher_node(state): return {"messages": [{"role": "assistant", "content": "Информация найдена"}]} def should_continue(state): return state.get("next", "END") # 建图 graph = StateGraph(MessagesState) graph.add_node("supervisor", supervisor_node) graph.add_node("coder", coder_node) graph.add_node("analyst", analyst_node) graph.add_node("researcher", researcher_node) # 主管决定后分派到对应Agent graph.add_edge("supervisor", "coder", condition=lambda s: s.get("next") == "coder") graph.add_edge("supervisor", "analyst", condition=lambda s: s.get("next") == "analyst") graph.add_edge("supervisor", "researcher", condition=lambda s: s.get("next") == "researcher") # Agent完成后回到主管 graph.add_edge("coder", "supervisor") graph.add_edge("analyst", "supervisor") graph.add_edge("researcher", "supervisor") graph.add_conditional_edges("supervisor", should_continue, {"END": END}) app = graph.compile()

from langgraph.checkpoint.memory import MemorySaver class CustomerServiceState(TypedDict): messages: list customer_id: str session_history: list intent: str checkpointer = MemorySaver() # Agent间共享记忆 graph = StateGraph(CustomerServiceState) graph.add_node("understand", understand_node) graph.add_node("search_kb", search_knowledge_base_node) graph.add_node("generate", generate_response_node) graph.add_node("validate", validate_response_node) graph.add_edge("understand", "search_kb") graph.add_edge("search_kb", "generate") graph.add_edge("generate", "validate") def route_validation(state): if state.get("valid", False): return "generate" # 重写 return END # 通过，结束 graph.add_conditional_edges("validate", route_validation, {"generate": "generate", "END": END}) # 带记忆的图 app = graph.compile( checkpointer=checkpointer, # 状态持久化 interrupt_before=["validate"] # 可暂停让人工审核 ) # 恢复会话 config = {"configurable": {"thread_id": "customer-12345"}} result = app.invoke({"messages": [input_msg]}, config=config)

def query_transform_node(state): """将用户问题分解为多个子查询""" query = state["messages"][-1]["content"] sub_queries = [q.strip() for q in query.split("?")] return {"sub_queries": sub_queries} def retrieve_node(state): """并行检索多个子查询""" # 每个子查询 → 独立向量检索 results = [] for q in state["sub_queries"]: r = vector_db.similarity_search(q, k=5) results.append(r) return {"retrieved_docs": results} def synthesize_node(state): """综合多个检索结果生成答案""" context = "\n".join([doc.page_content for doc in state["retrieved_docs"][0]]) answer = llm.invoke(f"基于以下内容回答：{context}\n问题：{state['messages'][-1]['content']}") return {"messages": [{"role": "assistant", "content": answer}]} graph = StateGraph(AgentState) graph.add_node("transform", query_transform_node) graph.add_node("retrieve", retrieve_node) graph.add_node("synthesize", synthesize_node) graph.add_edge("transform", "retrieve") graph.add_edge("retrieve", "synthesize") graph.add_edge("synthesize", END)

def classify_intent(state): """意图分类：技术支持/销售/投诉/其他""" msg = state["messages"][-1]["content"].lower() if "не работает" in msg or "ошибк" in msg: return "tech_support" elif "купить" in msg or "цена" in msg or "стоимост" in msg: return "sales" elif "жалоб" in msg or "возврат" in msg: return "complaint" return "general" def tech_support_flow(state): return {"next_agent": "tech_agent", "priority": "high"} def sales_flow(state): return {"next_agent": "sales_agent", "priority": "medium"} def complaint_flow(state): return {"next_agent": "manager_agent", "priority": "high"} # 多条件分支 def route(state): intent = classify_intent(state) routes = { "tech_support": "tech", "sales": "sales", "complaint": "manager", "general": "general" } return routes.get(intent, "general") graph.add_conditional_edges("classifier", route, { "tech": "tech_agent", "sales": "sales_agent", "manager": "manager_agent", "general": "general_agent" })