LangGraph 实战

https://github.com/webup/langgraph-in-action
Admin
LangGraph 实战 is a practical guide and interactive Jupyter Notebook repository for LangGraph,...
Tokens:100,958
Snippets:238
Trust Score:9.2
Update:1 year ago
Show doc for...
Tokens:
Raw
### 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
Tokens:
Raw