这一课学 LangGraph 的“条件分支”:检索到资料就回答;没检索到就走兜底分支。关键是add_conditional_edges()——它根据当前状态决定下一步运行哪个节点。官方说明
在langchain-study文件夹新建unit4_lesson2_conditional_graph.py:
from typing import Literal, NotRequired, TypedDict from langgraph.graph import START, END, StateGraph class BookstoreState(TypedDict): question: str context: NotRequired[str] answer: NotRequired[str] def retrieve(state: BookstoreState) -> dict[str, str]: print("执行:retrieve") if "会员" in state["question"]: return {"context": "会员购买图书可以享受八折优惠。"} if "退货" in state["question"]: return {"context": "购买后七天内可凭购物小票申请退货。"} return {"context": ""} def choose_route( state: BookstoreState, ) -> Literal["answer_found", "answer_missing"]: return "answer_found" if state["context"] else "answer_missing" def answer_found(state: BookstoreState) -> dict[str, str]: print("执行:answer_found") return {"answer": state["context"]} def answer_missing(state: BookstoreState) -> dict[str, str]: print("执行:answer_missing") return {"answer": "资料中没有相关信息。"} builder = StateGraph(BookstoreState) builder.add_node("retrieve", retrieve) builder.add_node("answer_found", answer_found) builder.add_node("answer_missing", answer_missing) builder.add_edge(START, "retrieve") builder.add_conditional_edges("retrieve", choose_route) builder.add_edge("answer_found", END) builder.add_edge("answer_missing", END) graph = builder.compile() for question in ["会员买书有什么优惠?", "书店几点关门?"]: print(f"\n问题:{question}") result = graph.invoke({"question": question}) print("回答:", result["answer"]) print("\n图的 Mermaid 表示:") print(graph.get_graph().draw_mermaid())运行:
python unit4_lesson2_conditional_graph.py你应该看到:第一个问题执行answer_found,第二个执行answer_missing。注意这里不要再加retrieve → answer_found的普通边,否则它可能和条件分支一起执行。
这课的核心问题:choose_route是在retrieve之前,还是之后执行?跑完看输出,我们再继续。