diff --git a/08_streaming.py b/08_streaming.py new file mode 100644 index 0000000..7dd0a29 --- /dev/null +++ b/08_streaming.py @@ -0,0 +1,254 @@ +"""LangGraph Streaming 示例:节点更新 + 模型 Token 流。 + +图结构: + + START -> model + | 有 tool_calls + v + tools + | + +---------> model + + model -- 无 tool_calls --> END + +本课重点不是工具本身,而是三种 stream_mode: + +1. updates —— 每个节点执行完后,推送该节点的状态增量 +2. messages —— 模型生成时,逐 token 推送 AIMessageChunk +3. 同时开两者 —— 一边打字机输出,一边看到节点何时完成 + +运行: + uv run --with langgraph --with langchain-openai --with python-dotenv \ + python 08_streaming.py +""" + +import os +from typing import Literal + +from dotenv import load_dotenv +from langchain_core.tools import tool +from langchain_openai import ChatOpenAI +from langgraph.graph import END, START, MessagesState, StateGraph +from langgraph.prebuilt import ToolNode + +load_dotenv() + + +# ----------------------------------------------------------------------------- +# 1. 工具与模型(沿用上一课的天气查询,方便观察节点切换) +# ----------------------------------------------------------------------------- + + +@tool +def get_weather(city: str) -> str: + """查询指定城市当前的天气。城市名称使用中文,例如北京、上海。""" + weather_data = { + "北京": "晴,25°C,北风 2 级", + "上海": "多云,28°C,东南风 3 级", + "广州": "雷阵雨,30°C,南风 2 级", + "深圳": "阴,29°C,东风 2 级", + } + return weather_data.get(city, f"暂时没有 {city} 的天气数据") + + +tools = [get_weather] +model = ChatOpenAI( + model=os.getenv("OPENAI_MODEL", "openai/gpt-5.4"), + api_key=os.getenv("OPENAI_API_KEY"), + base_url=os.getenv("OPENAI_BASE_URL", "https://api.qnaigc.com/v1"), + temperature=0, + max_retries=2, +) +model_with_tools = model.bind_tools(tools) + + +def call_model(state: MessagesState) -> dict: + # 即使这里写的是 .invoke,stream_mode="messages" 仍能拿到 token 流。 + # LangGraph 会自动把内部 LLM 调用的流式事件透传出来。 + response = model_with_tools.invoke(state["messages"]) + return {"messages": [response]} + + +def route_after_model(state: MessagesState) -> Literal["tools", "__end__"]: + last_message = state["messages"][-1] + if last_message.tool_calls: + return "tools" + return END + + +tool_node = ToolNode(tools) + +builder = StateGraph(MessagesState) +builder.add_node("model", call_model) +builder.add_node("tools", tool_node) +builder.add_edge(START, "model") +builder.add_conditional_edges( + "model", + route_after_model, + {"tools": "tools", END: END}, +) +builder.add_edge("tools", "model") +graph = builder.compile() + + +# ----------------------------------------------------------------------------- +# 2. 演示 A:stream_mode="updates" —— 看节点级增量 +# ----------------------------------------------------------------------------- + + +def demo_updates(question: str) -> None: + """每个节点跑完才出一包数据:{节点名: 该节点返回的状态增量}。""" + print("\n--- stream_mode='updates' ---") + + for event in graph.stream( + {"messages": [{"role": "user", "content": question}]}, + stream_mode="updates", + ): + # event 形如:{"model": {"messages": [AIMessage(...)]}} + # 或:{"tools": {"messages": [ToolMessage(...)]}} + for node_name, update in event.items(): + print(f"\n[{node_name}] 节点完成,增量:") + for message in update.get("messages", []): + message.pretty_print() + + +# ----------------------------------------------------------------------------- +# 3. 演示 B:stream_mode="messages" —— 看模型 Token 流 +# ----------------------------------------------------------------------------- + + +def demo_messages(question: str) -> None: + """模型一边生成,一边吐 AIMessageChunk;适合做打字机效果。""" + print("\n--- stream_mode='messages' ---") + print("助手:", end="", flush=True) + + for chunk in graph.stream( + {"messages": [{"role": "user", "content": question}]}, + stream_mode="messages", + version="v2", + ): + # v2 统一格式:{"type": "messages", "data": (message_chunk, metadata)} + if chunk["type"] != "messages": + continue + + message_chunk, metadata = chunk["data"] + + # 只打印 model 节点吐出的文本 token;工具调用阶段 content 常为空。 + if metadata.get("langgraph_node") != "model": + continue + + content = message_chunk.content + if content: + # content 可能是 str,也可能是 content blocks 列表,这里只处理文本。 + if isinstance(content, str): + print(content, end="", flush=True) + elif isinstance(content, list): + for block in content: + if isinstance(block, dict) and block.get("type") == "text": + print(block.get("text", ""), end="", flush=True) + elif isinstance(block, str): + print(block, end="", flush=True) + + print() # 换行 + + +# ----------------------------------------------------------------------------- +# 4. 演示 C:同时开 updates + messages —— 最实用的组合 +# ----------------------------------------------------------------------------- + + +def demo_both(question: str) -> None: + """一边流式打印 token,一边在节点结束时打印节点名。""" + print("\n--- stream_mode=['updates', 'messages'] ---") + print("助手:", end="", flush=True) + printed_tokens = False + + for chunk in graph.stream( + {"messages": [{"role": "user", "content": question}]}, + stream_mode=["updates", "messages"], + version="v2", + ): + if chunk["type"] == "messages": + message_chunk, metadata = chunk["data"] + if metadata.get("langgraph_node") != "model": + continue + + content = message_chunk.content + if isinstance(content, str) and content: + print(content, end="", flush=True) + printed_tokens = True + elif isinstance(content, list): + for block in content: + text = "" + if isinstance(block, dict) and block.get("type") == "text": + text = block.get("text", "") + elif isinstance(block, str): + text = block + if text: + print(text, end="", flush=True) + printed_tokens = True + + elif chunk["type"] == "updates": + # 节点结束后再换行并标注,避免和 token 混在同一行。 + for node_name in chunk["data"]: + if printed_tokens: + print() + printed_tokens = False + print(f" ✓ 节点 [{node_name}] 完成") + if node_name == "model": + print("助手:", end="", flush=True) + + if printed_tokens: + print() + + +# ----------------------------------------------------------------------------- +# 5. 入口:依次跑三种模式,对比输出差异 +# ----------------------------------------------------------------------------- + + +def main() -> None: + print("=== LangGraph Streaming:节点更新 + Token 流 ===") + print("建议先问:北京天气怎么样? (会触发 tools -> model 两跳)") + print("或问:用一句话介绍 LangGraph。 (只走 model)") + print("输入 quit 退出。\n") + print("命令:") + print(" /u 只看 updates") + print(" /m 只看 messages(token)") + print(" /b 同时看两者(默认)") + + mode = "both" + + while True: + question = input(f"\n用户[{mode}]:").strip() + if question.lower() in {"quit", "exit", "q"}: + break + if not question: + continue + + if question == "/u": + mode = "updates" + print("已切换到 updates") + continue + if question == "/m": + mode = "messages" + print("已切换到 messages") + continue + if question == "/b": + mode = "both" + print("已切换到 both") + continue + + try: + if mode == "updates": + demo_updates(question) + elif mode == "messages": + demo_messages(question) + else: + demo_both(question) + except Exception as exc: + print(f"\n执行失败:{type(exc).__name__}: {exc}") + + +if __name__ == "__main__": + main() diff --git a/09_retry_and_errors.py b/09_retry_and_errors.py new file mode 100644 index 0000000..6e7a75e --- /dev/null +++ b/09_retry_and_errors.py @@ -0,0 +1,185 @@ +"""LangGraph 容错示例:重试、错误恢复与循环保护。 + +本文件不调用真实 LLM/API,可以直接运行并稳定复现实验: + +1. transient:前两次 ConnectionError,第三次成功 +2. permanent:ConnectionError 重试耗尽后进入 error_handler +3. invalid:ValueError 不重试,直接进入 error_handler +4. loop:故意无限循环,由 recursion_limit 终止 + +需要 langgraph >= 1.2(节点级 error_handler / NodeError 是 1.2+ 能力)。 + +运行: + python 09_retry_and_errors.py +""" + +from typing import Literal, TypedDict + +from langgraph.errors import GraphRecursionError, NodeError +from langgraph.graph import END, START, StateGraph +from langgraph.runtime import Runtime +from langgraph.types import Command, RetryPolicy + + +# ----------------------------------------------------------------------------- +# 1. 节点重试 + 重试耗尽后的错误恢复 +# ----------------------------------------------------------------------------- + + +class ApiState(TypedDict): + mode: Literal["transient", "permanent", "invalid"] + result: str + status: str + error: str + + +def call_unstable_api( + state: ApiState, + runtime: Runtime, +) -> Command[Literal["finalize"]]: + """模拟一个不稳定的外部 API。 + + 注意:失败尝试产生的 State 更新不会提交,因此不要用 State 字段记录 + 当前重试次数。LangGraph 已通过 execution_info 提供 node_attempt。 + """ + attempt = runtime.execution_info.node_attempt + mode = state["mode"] + print(f" call_unstable_api:第 {attempt} 次尝试,mode={mode}") + + if mode == "transient" and attempt < 3: + raise ConnectionError("模拟网络抖动") + + if mode == "permanent": + raise ConnectionError("模拟上游服务持续不可用") + + if mode == "invalid": + # ValueError 通常代表参数/业务错误,重复请求不会自行恢复。 + raise ValueError("模拟无效请求参数") + + return Command( + update={ + "result": "API 调用成功", + "status": "success", + "error": "", + }, + goto="finalize", + ) + + +def api_error_handler( + state: ApiState, + error: NodeError, +) -> Command[Literal["finalize"]]: + """在不重试或重试耗尽后,将异常转成可控的业务结果。""" + print( + f" error_handler:节点 {error.node} 最终失败:" + f"{type(error.error).__name__}: {error.error}" + ) + return Command( + update={ + "result": "已启用降级结果,请稍后重试", + "status": "recovered", + "error": f"{type(error.error).__name__}: {error.error}", + }, + goto="finalize", + ) + + +def finalize(state: ApiState) -> dict: + """统一收尾节点;成功和降级路径都会到达这里。""" + print(f" finalize:status={state['status']}, result={state['result']}") + return {} + + +api_builder = StateGraph(ApiState) +api_builder.add_node( + "call_api", + call_unstable_api, + retry_policy=RetryPolicy( + max_attempts=3, # 包含第一次,共最多执行 3 次 + initial_interval=0.1, # 第一次重试前等待 0.1 秒 + backoff_factor=2.0, # 后续等待时间指数增长 + max_interval=1.0, + jitter=False, # 教学输出保持可预测 + retry_on=ConnectionError, + ), + error_handler=api_error_handler, +) +api_builder.add_node("finalize", finalize) +api_builder.add_edge(START, "call_api") +api_builder.add_edge("finalize", END) +api_graph = api_builder.compile() + + +def run_api_case(mode: Literal["transient", "permanent", "invalid"]) -> None: + """运行一个重试/错误恢复场景,并打印最终状态。""" + print(f"\n=== 场景:{mode} ===") + result = api_graph.invoke( + { + "mode": mode, + "result": "", + "status": "pending", + "error": "", + } + ) + print(" 最终状态:", result) + + +# ----------------------------------------------------------------------------- +# 2. 循环保护:recursion_limit +# ----------------------------------------------------------------------------- + + +class LoopState(TypedDict): + count: int + + +def loop_forever(state: LoopState) -> dict: + """故意不提供终止条件,用于演示 GraphRecursionError。""" + next_count = state["count"] + 1 + print(f" loop:count={next_count}") + return {"count": next_count} + + +loop_builder = StateGraph(LoopState) +loop_builder.add_node("loop", loop_forever) +loop_builder.add_edge(START, "loop") +loop_builder.add_edge("loop", "loop") +loop_graph = loop_builder.compile() + + +def run_loop_case() -> None: + """用较小的递归限制安全终止错误设计的循环图。""" + print("\n=== 场景:loop / recursion_limit ===") + try: + loop_graph.invoke( + {"count": 0}, + config={"recursion_limit": 4}, + ) + except GraphRecursionError as exc: + print(f" 已捕获 {type(exc).__name__}") + print(" 原因:图在限制步数内没有到达 END。") + print(" 正确做法:为循环增加业务终止条件;recursion_limit 只是最后防线。") + + +# ----------------------------------------------------------------------------- +# 3. 入口 +# ----------------------------------------------------------------------------- + + +def main() -> None: + print("=== LangGraph Retry & Errors ===") + run_api_case("transient") + run_api_case("permanent") + run_api_case("invalid") + run_loop_case() + + print("\n=== 观察结论 ===") + print("1. ConnectionError 会按照 RetryPolicy 自动重试。") + print("2. ValueError 不符合 retry_on,第一次失败后立即交给 error_handler。") + print("3. 重试耗尽后,error_handler 可用 Command 更新状态并走降级路径。") + print("4. recursion_limit 能阻止失控循环,但不能替代正确的终止条件。") + + +if __name__ == "__main__": + main()