### Untitled No description -------------------------------- ### Copy Environment Example File Source: https://github.com/webup/langgraph-in-action/blob/main/README.md Copies the example environment file (.env.example) to a new file named .env. This is a standard practice for setting up project-specific configurations. ```bash cp .env.example .env ``` -------------------------------- ### Untitled No description -------------------------------- ### Install Project Dependencies with uv Source: https://github.com/webup/langgraph-in-action/blob/main/README.md Installs all project dependencies, including development environment and Jupyter kernels, using uv sync. The --dev flag ensures development-specific packages are included. ```bash uv sync --dev ``` -------------------------------- ### Untitled No description -------------------------------- ### Get Subgraph State Snapshot and History Source: https://github.com/webup/langgraph-in-action/blob/main/chapter4.ipynb This snippet demonstrates how to get the state snapshot and history for a specific subgraph. Subgraphs can have independent state management, identified by their `thread_id`. ```python # 获取子图状态（子图有独立的thread_id） subgraph_config = {"configurable": {"thread_id": "sub_数据分析任务"}} subgraph_state = compiled_subgraph.get_state(subgraph_config, subgraphs=True) print("Parent Graph State (子图状态):") print(subgraph_state.values) # 打印子图的状态值 print("--------------- ") # 检查子图的历史状态 print("Subgraph State History (子图历史状态):") sub_history = list(compiled_subgraph.get_state_history(subgraph_config)) for i, snapshot in enumerate(sub_history[:3]): print(f" 子图快照 {i+1}: {snapshot.values}") ``` -------------------------------- ### Untitled No description -------------------------------- ### Python: Get Final Graph State Source: https://github.com/webup/langgraph-in-action/blob/main/chapter4.ipynb This snippet retrieves and prints the final state of the approval graph after all operations, including the final approval, have been completed. ```python # print(approval_graph.get_state(config)) ``` -------------------------------- ### Untitled No description -------------------------------- ### Get Main Graph State Snapshot Source: https://github.com/webup/langgraph-in-action/blob/main/chapter4.ipynb Retrieves and prints the state snapshot of the main (grandparent) graph. This is useful for understanding the overall state of the hierarchical agent. ```python print("=== 获取子图状态快照 ===") # 获取主图状态 main_state = hierarchical_graph.get_state(config, subgraphs=True) print("Grandparent State (主图状态):") print(main_state.values) ``` -------------------------------- ### Untitled No description -------------------------------- ### Python: Simulate Final Approval and Get Final Result Source: https://github.com/webup/langgraph-in-action/blob/main/chapter4.ipynb This snippet simulates the human approving the modified action after a rejection. It invokes the graph with an 'approve' command and then prints the final result of the execution. ```python from langgraph.types import Command # 最终批准修改后的操作 # print("\n=== 第三步：批准修改后的操作 ===") # final_result = approval_graph.invoke( # Command(resume={"user_response": "approve"}), # config=config # ) # print("最终执行结果：", final_result["final_result"]) ``` -------------------------------- ### Untitled No description -------------------------------- ### Untitled No description -------------------------------- ### Untitled No description -------------------------------- ### Untitled No description -------------------------------- ### Untitled No description -------------------------------- ### Untitled No description -------------------------------- ### Get and Stream Subgraph State After Node Simulation (Python) Source: https://github.com/webup/langgraph-in-action/blob/main/chapter4.ipynb This snippet demonstrates how to retrieve the state of a compiled LangGraph subgraph after simulating a node's execution using `as_node`. It then shows how to stream the execution from this updated state, allowing for continued processing based on the simulated outcome. This is useful for debugging and testing specific execution paths. ```python # 查看更新后的状态 updated_state_as_node = compiled_enhanced_subgraph.get_state(updated_config_as_node) print("以节点身份更新后的子图状态：", updated_state_as_node.values) print("下一个要执行的节点：", updated_state_as_node.next) print("\n=== 从节点身份更新的状态继续执行 ===") print("从 'process' 节点更新的状态继续执行后续节点...") # 从更新的状态继续执行（会执行format节点） for chunk in compiled_enhanced_subgraph.stream(None, updated_config_as_node, stream_mode="values"): print("继续执行结果：", chunk) ``` -------------------------------- ### Untitled No description -------------------------------- ### Install uv Package Manager Source: https://github.com/webup/langgraph-in-action/blob/main/README.md Installs the uv package manager using a curl command. uv is a fast Python package installer and virtual environment manager. ```bash curl -LsSf https://astral.sh/uv/install.sh | sh ``` -------------------------------- ### Untitled No description -------------------------------- ### LangGraph Weather Subgraph Example (Python) Source: https://github.com/webup/langgraph-in-action/blob/main/chapter4.ipynb This Python code defines a LangGraph subgraph for analyzing weather information. It includes nodes for analyzing requests, querying weather, formatting results, and a parent graph structure to orchestrate tasks. This example demonstrates how to build and compile reusable subgraphs within LangGraph. ```python # 创建一个更真实的天气查询子图示例 from typing import TypedDict, List from langchain_core.messages import BaseMessage, HumanMessage, AIMessage class WeatherState(TypedDict): location: str messages: List[BaseMessage] weather_result: str class ParentTaskState(TypedDict): task_name: str location: str weather_info: str final_report: str # 天气子图节点函数 def weather_analyzer(state: WeatherState): """分析天气请求""" location = state.get("location", "unknown") return { "messages": state["messages"] + [HumanMessage(content=f"请查询{location}的天气")], "location": location } def weather_node(state: WeatherState): """天气查询节点 - 这是我们要模拟的关键节点""" location = state.get("location", "unknown") # 模拟天气API调用 weather_result = f"{location}的天气是晴朗的" return { "messages": state["messages"] + [AIMessage(content=weather_result)], "weather_result": weather_result } def weather_formatter(state: WeatherState): """格式化天气结果""" weather_result = state.get("weather_result", "无天气数据") return { "weather_result": f"[格式化] {weather_result}" } # 构建天气子图 (weather_graph) weather_graph = StateGraph(WeatherState) weather_graph.add_node("weather_analyzer", weather_analyzer) weather_graph.add_node("weather_node", weather_node) # 关键的weather_node weather_graph.add_node("weather_formatter", weather_formatter) weather_graph.add_edge(START, "weather_analyzer") weather_graph.add_edge("weather_analyzer", "weather_node") weather_graph.add_edge("weather_node", "weather_formatter") weather_graph.add_edge("weather_formatter", END) compiled_weather_graph = weather_graph.compile(checkpointer=MemorySaver()) # 父图节点函数 def prepare_weather_task(state: ParentTaskState): """准备天气查询任务""" return { "task_name": f"天气查询任务: {state.get('task_name', '默认任务')}", "location": state.get("location", "beijing") } def call_weather_subgraph(state: ParentTaskState): """调用天气子图""" subgraph_config = {"configurable": {"thread_id": f"weather_{state['location']}"}} # 调用天气子图 subgraph_input = { "location": state["location"], "messages": [HumanMessage(content="开始天气查询")], "weather_result": "" } subgraph_result = compiled_weather_graph.invoke(subgraph_input, subgraph_config) return { "weather_info": subgraph_result["weather_result"] } def generate_final_report(state: ParentTaskState): """生成最终报告""" return { "final_report": f"任务完成: {state['task_name']}, 天气信息: {state.get('weather_info', '无')}" } ``` -------------------------------- ### Untitled No description