conda create -n langgraph python==3.12

conda activate langgraph
pip install -U langgraph
1. 确认成功
pip show langgraph
from langgraph.graph import StateGraph, MessagesState, START, END
def this_is_a_method(params1:int,param2:int):
    print("this is a print")
1、定义节点函数
def mock_llm(state: MessagesState):
""" 模拟调用LLM """
`return {"messages": [{"ole": "ai", "content": "hello world"}]}`
2、定义图
graph = StateGraph(MessagesState)

### 2.2.3、添加节点和边

graph.add_node(mock_llm)

graph.add_edge(START, "mock_llm")

graph.add_edge("mock_llm", END)

# 4、编译图

graph = graph.compile()

# 5、调用图

response = graph.invoke({"messages": [{"role": "user", "content": "hi!"}]})

print(response)

from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
定义输入状态模式
class InputState(TypedDict):
    question: str
定义输出状态模式
class OutputState(TypedDict):
    answer: str
定义整体状态模式，结合输入和输出
class OverallState(InputState, OutputState):
    pass
定义处理节点
def answer_node(state: InputState):
    """
    处理输入并生成答案的节点
    Args:
        state: 输入状态
    Returns:
        dict: 包含答案的字典
    """
    print(f"执行 answer_node 节点:")
    print(f"  输入: {state}")
    # 示例答案
    answer = "再见" if "bye" in state["question"].lower() else "你好"
    result = {"answer": answer, "question": state["question"]}
    print(f"  输出: {result}")
    return result
定义整体状态（这是在节点间共享的公共状态）
class OverallStatePrivate(TypedDict):
    a: str
node_1 的输出包含不属于整体状态的私有数据
class Node1Output(TypedDict):
    private_data: str
私有数据仅在 node_1 和 node_2 之间共享
def node_1(state: OverallStatePrivate) -> Node1Output:
    """
    第一个节点，生成私有数据
    Args:
        state: 整体状态
    Returns:
        dict: 包含私有数据的字典
    """
    output = {"private_data": "由 node_1 设置"}
    print(f"进入 node_1 节点:")
    print(f"  输入: {state}")
    print(f"  返回: {output}")
    return output
节点2的输入只请求node_1之后可用的私有数据
class Node2Input(TypedDict):
    private_data: str
def node_2(state: Node2Input) -> OverallStatePrivate:
    """
    第二个节点，可以访问node_1的私有数据
    Args:
        state: 包含私有数据的输入状态
    Returns:
        dict: 更新的整体状态
    """
    output = {"a": "由 node_2 设置"}
    print(f"进入 node_2 节点:")
    print(f"  输入: {state}")
    print(f"  返回: {output}")
    return output
节点3只能访问整体状态（无法访问node_1的私有数据）
def node_3(state: OverallStatePrivate) -> OverallStatePrivate:
    """
    第三个节点，只能访问整体状态
    Args:
        state: 整体状态
    Returns:
        dict: 更新的整体状态
    """
    output = {"a": "由 node_3 设置"}
    print(f"进入 node_3 节点:")
    print(f"  输入: {state}")
    print(f"  返回: {output}")
    return output
def demo_input_output_schema():
    """演示输入输出模式"""
    print("= 演示输入输出模式 =")
    # 使用指定的输入和输出模式构建图
    builder = StateGraph(OverallState, input_schema=InputState, output_schema=OutputState)
    builder.add_node("answer_node", answer_node)  # 添加答案节点
    builder.add_edge(START, "answer_node")  # 定义起始边
    builder.add_edge("answer_node", END)  # 定义结束边
    graph = builder.compile()  # 编译图
    # 使用输入调用图并打印结果
    result = graph.invoke({"question": "你好"})
    print(f"图调用结果: {result}")
    print()
def demo_private_state():
    """演示私有状态传递"""
    print("= 演示私有状态传递 =")
    # 连接节点序列
    # node_2 接受来自 node_1 的私有数据，而
    # node_3 看不到来自 node_1 的私有数据
    builder = StateGraph(OverallStatePrivate).add_sequence([node_1, node_2, node_3])
    builder.add_edge(START, "node_1")
    graph = builder.compile()
    # 使用初始状态调用图
    response = graph.invoke({
        "a": "在开始时设置",
    })
    print()
    print(f"图调用的输出: {response}")
    print()
def main():
    """主函数"""
    print("= LangGraph 图输入输出模式和私有状态传递演示 =\n")
    # 演示输入输出模式
    demo_input_output_schema()
    # 演示私有状态传递
    demo_private_state()
    print("= 演示完成 =")
if name == "main":
    main()

“”“

LangGraph Reducer函数演示 - 默认Reducer（覆盖更新）
“”“
from typing import List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
1. 默认Reducer（覆盖更新）
class DefaultReducerState(TypedDict):
    foo: int
    bar: List[str]

def node_default_1(state: DefaultReducerState) -> dict:
    return {"foo": 2}

def node_default_2(state: DefaultReducerState) -> dict:
    return {"bar": ["bye"]}

def run_demo():
print("1. 默认Reducer（覆盖更新）演示:")
builder = StateGraph(DefaultReducerState)
builder.add_node("node1", node_default_1)
    builder.add_node("node2", node_default_2)
    builder.add_edge(START, "node1")
    builder.add_edge("node1", "node2")
    builder.add_edge("node2", END)
    graph = builder.compile()
    result = graph.invoke({"foo": 1, "bar": ["hi"]})
print(f"初始状态: {{'foo': 1, 'bar': ['hi']}}")
print(f"执行结果: {result}\n")

if name == "main":
    run_demo()

"""
LangGraph Reducer函数演示 - add_messages Reducer（消息列表专用）
"""
from typing import Annotated, List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.graph.message import add_messages
2. add_messages Reducer（消息列表专用）
   
class AddMessagesState(TypedDict):
    messages: Annotated[List, add_messages]
    
def chat_node_1(state: AddMessagesState) -> dict:
    return {"messages": [("assistant", "Hello from node 1")]}
    
def chat_node_2(state: AddMessagesState) -> dict:
    return {"messages": [("assistant", "Hello from node 2")]}
    
def run_demo():
    print("2. add_messages Reducer（消息列表专用）演示:")
    
    builder = StateGraph(AddMessagesState)
    builder.add_node("chat1", chat_node_1)
    builder.add_node("chat2", chat_node_2)
    builder.add_edge(START, "chat1")

    builder.add_edge(START, "chat2")  # 并行执行
    builder.add_edge("chat1", END)
    builder.add_edge("chat2", END)
    
    graph = builder.compile()
    
    result = graph.invoke({"messages": [("user", "Hi there!")]})
    print(f"初始状态: {{'messages': [('user', 'Hi there!')]}}")
    print(f"执行结果: {result}\n")
if name == "main":
    run_demo()

"""
LangGraph Reducer函数演示 - operator.add Reducer（列表追加）
"""
import operator
from typing import Annotated, List
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
3. operator.add Reducer（列表追加）
class ListAddState(TypedDict):
    data: Annotated[List[int], operator.add]
def producer_1(state: ListAddState) -> dict:
    return {"data": [1, 2]}
def producer_2(state: ListAddState) -> dict:
    return {"data": [3, 4]}
def run_demo():
    print("3.1 operator.add Reducer（列表追加）演示:")
    builder = StateGraph(ListAddState)
    builder.add_node("producer1", producer_1)
    builder.add_node("producer2", producer_2)
    builder.add_edge(START, "producer1")
    builder.add_edge(START, "producer2")  # 并行执行
    builder.add_edge("producer1", END)
    builder.add_edge("producer2", END)
    graph = builder.compile()
    result = graph.invoke({"data": [0]})
    print(f"初始状态: {{'data': [0]}}")
    print(f"执行结果: {result}\n")
if name == "main":
    run_demo()

"""
LangGraph Reducer函数演示 - 字符串连接Reducer
"""
import operator
from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
6. 字符串连接Reducer
   
class StringConcatState(TypedDict):
    text: Annotated[str, operator.add]
    
def add_text_1(state: StringConcatState) -> dict:
    return {"text": "Hello "}
    
def add_text_2(state: StringConcatState) -> dict:
    return {"text": "World!"}
    
def run_demo():
    print("3.2 字符串连接Reducer演示:")
    builder = StateGraph(StringConcatState)
    builder.add_node("add_text_1", add_text_1)
    builder.add_node("add_text_2", add_text_2)
    builder.add_edge(START, "add_text_1")
    builder.add_edge(START, "add_text_2")  # 并行执行
    builder.add_edge("add_text_1", END)
    builder.add_edge("add_text_2", END)
    graph = builder.compile()
    result = graph.invoke({"text": "Say: "})
    print(f"初始状态: {{'text': 'Say: '}}")
    print(f"执行结果: {result}\n")
if name == "main":
    run_demo()

"""

LangGraph Reducer函数演示 - 数值累加Reducer

"""

import operator

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

# 7. 数值累加Reducer

class NumberAddState(TypedDict):

    count: Annotated[int, operator.add]

def increment_1(state: NumberAddState) -> dict:

    return {"count": 5}

def increment_2(state: NumberAddState) -> dict:

    return {"count": 3}

def run_demo():

    print("3.3 数值累加Reducer演示:")

    builder = StateGraph(NumberAddState)

    builder.add_node("increment_1", increment_1)

    builder.add_node("increment_2", increment_2)

    builder.add_edge(START, "increment_1")

    builder.add_edge(START, "increment_2")  # 并行执行

    builder.add_edge("increment_1", END)

    builder.add_edge("increment_2", END)

    graph = builder.compile()

    result = graph.invoke({"count": 10})

    print(f"初始状态: {{'count': 10}}")

    print(f"执行结果: {result}\n")

if __name__ == "__main__":

    run_demo()

"""

LangGraph Reducer函数演示 - operator.mul Reducer（数值相乘）

"""

import operator

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

# 4. operator.mul Reducer（数值相乘）

class MultiplyState(TypedDict):

    factor: Annotated[float, operator.mul]

def multiplier(state: MultiplyState) -> dict:

    return {"factor": 2.0}

def run_demo():

    """

    """

    print("4. operator.mul Reducer（数值相乘）演示:")

    builder = StateGraph(MultiplyState)

    builder.add_node("multiplier", multiplier)

    builder.add_edge(START, "multiplier")

    builder.add_edge("multiplier", END)

    graph = builder.compile()

    result = graph.invoke({"factor": 5.0})

    print(f"初始状态: {{'factor': 5.0}}")

    print(f"执行结果: {result}\n")

if __name__ == "__main__":

    run_demo()

"""

LangGraph Reducer函数演示 - 自定义Reducer函数

"""

from typing import Annotated, Dict, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

# 5. 自定义Reducer函数

def custom_reducer(current_value: Dict[str, Any], new_value: Dict[str, Any]) -> Dict[str, Any]:

    """合并两个字典，新值会覆盖旧值，但保留旧值中不存在的键"""

    result = current_value.copy()

    result.update(new_value)

    return result

class CustomReducerState(TypedDict):

    metadata: Annotated[Dict[str, Any], custom_reducer]

def update_metadata(state: CustomReducerState) -> dict:

    return {"metadata": {"timestamp": "2025-01-01", "version": "1.0"}}

def run_demo():

    print("5. 自定义Reducer演示:")

    builder = StateGraph(CustomReducerState)

    builder.add_node("update_metadata", update_metadata)

    builder.add_edge(START, "update_metadata")

    builder.add_edge("update_metadata", END)

    graph = builder.compile()

    result = graph.invoke({"metadata": {"user_id": "123", "session": "abc"}})

    print(f"初始状态: {{'metadata': {{'user_id': '123', 'session': 'abc'}}}}")

    print(f"执行结果: {result}\n")

if __name__ == "__main__":

run_demo()

"""

LangGraph Reducer函数演示 - 自定义 mul reducer 实现数值相乘

使用全局变量区分初始化调用和正常调用

"""

from typing import Annotated

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

# 使用全局变量来跟踪是否是第一次调用（初始化阶段）

_is_initial_call = True

def my_mul_reducer(current_value: float, new_value: float) -> float:

    """

    自定义乘法reducer，使用全局变量区分初始化调用和正常调用

    Args:

        current_value: 当前状态值

        new_value: 新值

    Returns:

        计算后的结果

    """

    global _is_initial_call

    print(f"Reducer被调用: current_value={current_value}, new_value={new_value}, is_initial_call={_is_initial_call}")

    # 如果是初始化调用，直接返回new_value，避免默认值0的影响

    if _is_initial_call:

        _is_initial_call = False  # 重置标志

        return new_value

    else:

        # 正常的乘法操作，包括乘以0的情况

        return current_value * new_value

class MultiplyState(TypedDict):

    factor: Annotated[float, my_mul_reducer]

def multiplier_by_two(state: MultiplyState) -> dict:

    """将factor乘以2"""

    return {"factor": 2.0}

def multiplier_by_zero(state: MultiplyState) -> dict:

    """将factor乘以0"""

    return {"factor": 0.0}

def run_demo():

    """

    演示增强版乘法reducer的使用

    """

    global _is_initial_call

    print("=== operator.mul 增强版解决方案演示 ===\n")

    # 演示1: 正常乘法操作

    print("1. 正常乘法操作演示:")

    _is_initial_call = True  # 重置初始化标志

    builder = StateGraph(MultiplyState)

    builder.add_node("multiplier_by_two", multiplier_by_two)

    builder.add_edge(START, "multiplier_by_two")

    builder.add_edge("multiplier_by_two", END)

    graph = builder.compile()

    result = graph.invoke({"factor": 5.0})

    print(f"初始状态: {{'factor': 5.0}}")

    print(f"执行结果: {result}")

    print(f"预期结果: 10.0 (5.0 * 2.0)\n")

    # 演示2: 乘以0的操作

    print("2. 乘以0的操作演示:")

    _is_initial_call = True  # 重置初始化标志

    builder2 = StateGraph(MultiplyState)

    builder2.add_node("multiplier_by_zero", multiplier_by_zero)

    builder2.add_edge(START, "multiplier_by_zero")

    builder2.add_edge("multiplier_by_zero", END)

    graph2 = builder2.compile()

    result2 = graph2.invoke({"factor": 5.0})

    print(f"初始状态: {{'factor': 5.0}}")

    print(f"执行结果: {result2}")

    print(f"预期结果: 0.0 (5.0 * 0.0)\n")

    # 演示3: 连续乘法操作

    print("3. 连续乘法操作演示:")

    _is_initial_call = True  # 重置初始化标志

    builder3 = StateGraph(MultiplyState)

    builder3.add_node("multiplier_by_two_1", multiplier_by_two)

    builder3.add_node("multiplier_by_zero", multiplier_by_zero)

    builder3.add_node("multiplier_by_two_2", multiplier_by_two)

    builder3.add_edge(START, "multiplier_by_two_1")

    builder3.add_edge("multiplier_by_two_1", "multiplier_by_zero")

    builder3.add_edge("multiplier_by_zero", "multiplier_by_two_2")

    builder3.add_edge("multiplier_by_two_2", END)

    graph3 = builder3.compile()

    result3 = graph3.invoke({"factor": 3.0})

    print(f"初始状态: {{'factor': 3.0}}")

    print(f"执行结果: {result3}")

    print(f"预期过程: 3.0 -> 6.0 -> 0.0 -> 0.0")

    print(f"预期结果: 0.0\n")

if __name__ == "__main__":

    run_demo()

from typing import Annotated, List

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

from langgraph.graph.message import add_messages

import operator

class ChatState(TypedDict):

    messages: Annotated[list, add_messages]  # 消息历史

    tags: Annotated[List[str], operator.add]  # 标签列表

    score: Annotated[float, operator.add]     # 累计分数

def process_user_message(state: ChatState) -> dict:

    user_message = state["messages"][-1]  # 获取最新消息

    # 修复：正确访问消息内容

    return {

        "messages": [("assistant", f"Echo: {user_message.content}")],

        "tags": ["processed"],

        "score": 1.0

    }

def add_sentiment_tag(state: ChatState) -> dict:

    return {

        "tags": ["positive"],

        "score": 0.5

    }

# 构建图

builder = StateGraph(ChatState)

builder.add_node("process", process_user_message)

builder.add_node("sentiment", add_sentiment_tag)

builder.add_edge(START, "process")

builder.add_edge(START, "sentiment")

builder.add_edge("process", END)

builder.add_edge("sentiment", END)

graph = builder.compile()

# 使用示例 -使用正确的消息格式

result = graph.invoke({

    "messages": [{"role": "user", "content": "Hello, how are you?"}],

    "tags": ["greeting"],

    "score": 0.0

})

print(result)

def node(state:State, config:RunnableConfig,*,store)

import time

from typing_extensions import TypedDict

from langgraph.graph import StateGraph

from langgraph.cache.memory import InMemoryCache

from langgraph.types import CachePolicy

# 定义状态

class State(TypedDict):

    x: int

    result: int

# 创建图

builder = StateGraph(State)

# 定义节点

def expensive_node(state: State) -> dict[str, int]:

    # expensive computation

    time.sleep(2)

    return {"result": state["x"] * 2}

# 添加节点，并设置缓存策略

builder.add_node("expensive_node", expensive_node, cache_policy=CachePolicy(ttl=3))

# 设置入口和出口

builder.set_entry_point("expensive_node")

builder.set_finish_point("expensive_node")

# 编译图

graph = builder.compile(cache=InMemoryCache())

# 执行图

print(graph.invoke({"x": 5}, stream_mode='updates'))

# [{'expensive_node': {'result': 10}}]

# 第二次运行利用缓存并快速返回

print(graph.invoke({"x": 5}, stream_mode='updates'))

# [{'expensive_node': {'result': 10}, '__metadata__': {'cached': True}}]

RetryPolicy(NameTuple):
	max_attempts:int
	retry_on:callable | Function

"""

LangGraph 节点重试策略演示

"""

import random

from typing import Dict, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

from langgraph.types import RetryPolicy

# 定义状态

class State(TypedDict):

    result: str

# 模拟不稳定的API调用，使用全局变量跟踪尝试次数

attempt_counter = 0

def unstable_api_call(state: State) -> Dict[str, Any]:

    """

    模拟一个不稳定的API调用，有一定概率失败

    """

    global attempt_counter

    attempt_counter += 1

    print(f"尝试调用API，这是第 {attempt_counter} 次尝试")

    # 模拟前几次尝试失败，最后一次成功

    if attempt_counter < 3:

        raise Exception(f"模拟API调用失败 (尝试 {attempt_counter})")

    else:

        # 第三次尝试成功

        return {

            "result": f"API调用成功，经过 {attempt_counter} 次尝试"

        }

# 自定义重试策略

def custom_retry_on(exception: Exception) -> bool:

    """

    自定义重试条件：只对特定错误进行重试

    """

    # 只对包含"模拟API调用失败"的异常进行重试

    if "模拟API调用失败" in str(exception):

        print(f"捕获到可重试异常: {exception}")

        return True

    # 对其他异常不重试

    print(f"捕获到不可重试异常: {exception}")

    return False

# 模拟抛出 ValueError 的节点

def value_error_call(state: State) -> Dict[str, Any]:

    """

    模拟抛出 ValueError 的节点（不会被默认重试策略重试）

    """

    print("调用会抛出 ValueError 的节点")

    raise ValueError("模拟 ValueError 异常")

def run_demo():

    print("=== LangGraph 节点重试策略演示 ===\n")

    # 重置全局计数器

    global attempt_counter

    attempt_counter = 0

    # 演示1: 使用默认重试策略

    print("1. 使用默认重试策略:")

    print("   默认策略会对除特定异常外的所有异常进行重试")

    print("   不会重试的异常包括: ValueError, TypeError, ArithmeticError, ImportError,")

    print("                     LookupError, NameError, SyntaxError, RuntimeError,")

    print("                     ReferenceError, StopIteration, StopAsyncIteration, OSError\n")

    builder1 = StateGraph(State)

    # 添加节点，使用默认重试策略

    builder1.add_node(

        "unstable_call",

        unstable_api_call,

        retry_policy=RetryPolicy(max_attempts=5)  # 允许最多5次尝试

    )

    builder1.add_edge(START, "unstable_call")

    builder1.add_edge("unstable_call", END)

    graph1 = builder1.compile()

    print("测试默认重试策略:")

    try:

        result = graph1.invoke({"result": ""})

        print(f"最终结果: {result}\n")

    except Exception as e:

        print(f"最终失败: {type(e).__name__}: {e}\n")

    # 演示2: 使用自定义重试策略

    print("2. 使用自定义重试策略:")

    print("   自定义策略只对特定错误进行重试\n")

    # 重置全局计数器

    attempt_counter = 0

    builder2 = StateGraph(State)

    # 添加节点，使用自定义重试策略

    builder2.add_node(

        "custom_retry_call",

        unstable_api_call,

        retry_policy=RetryPolicy(max_attempts=5, retry_on=custom_retry_on)

    )

    builder2.add_edge(START, "custom_retry_call")

    builder2.add_edge("custom_retry_call", END)

    graph2 = builder2.compile()

    print("测试自定义重试策略:")

    try:

        result = graph2.invoke({"result": ""})

        print(f"最终结果: {result}\n")

    except Exception as e:

        print(f"最终失败: {type(e).__name__}: {e}\n")

    # 演示3: 不会重试的异常类型

    print("3. 测试不会重试的异常类型:")

    builder3 = StateGraph(State)

    # 添加节点，使用默认重试策略

    builder3.add_node(

        "value_error_call",

        value_error_call,

        retry_policy=RetryPolicy(max_attempts=3)

    )

    builder3.add_edge(START, "value_error_call")

    builder3.add_edge("value_error_call", END)

    graph3 = builder3.compile()

    print("测试 ValueError（默认策略不会重试）:")

    try:

        result = graph3.invoke({"result": ""})

        print(f"最终结果: {result}\n")

    except Exception as e:

        print(f"最终失败: {type(e).__name__}: {e}\n")

if __name__ == "__main__":

    run_demo()

"""

LangGraph 延迟节点执行演示

本示例展示了如何使用defer=True来实现节点延迟执行，确保该节点等待所有其他并行分支任务完成后才执行。

"""

import operator

from typing import Annotated, Any

from typing_extensions import TypedDict

from langgraph.graph import StateGraph, START, END

  

class State(TypedDict):

    """

    状态类型定义

    aggregate: 使用operator.add reducer使这个列表为追加模式，确保每个节点的结果都能被正确合并

    """

    # The operator.add reducer fn makes this append-only

    aggregate: Annotated[list, operator.add]

def a(state: State):

    """

    节点a：启动分支

    此节点是工作流的起点，负责初始化流程并分发到不同的分支。

    Args:

        state: 当前状态

    Returns:

        dict: 包含新结果的状态更新

    """

    print(f'Adding "A" to {state["aggregate"]}')

    return {"aggregate": ["A"]}

def b(state: State):

    """

    节点b：第一个分支

    此节点处理第一个分支的任务，与节点c并行执行。

    Args:

        state: 当前状态

    Returns:

        dict: 包含新结果的状态更新

    """

    print(f'Adding "B" to {state["aggregate"]}')

    return {"aggregate": ["B"]}

def b_2(state: State):

    """

    节点b2：第二个分支

    此节点处理第二个分支的任务，在节点b完成后执行。

    Args:

        state: 当前状态

    Returns:

        dict: 包含新结果的状态更新

    """

    print(f'Adding "B_2" to {state["aggregate"]}')

    return {"aggregate": ["B_2"]}

def c(state: State):

    """

    节点c：另一个分支

    此节点处理另一个分支的任务，与节点b并行执行。

    Args:

        state: 当前状态

    Returns:

        dict: 包含新结果的状态更新

    """

    print(f'Adding "C" to {state["aggregate"]}')

    return {"aggregate": ["C"]}

def d(state: State):

    """

    节点d：延迟执行的汇总节点

    此节点设置了defer=True，因此会等待所有其他任务完成后才执行。

    它负责汇总所有分支的结果。

    Args:

        state: 当前状态

    Returns:

        dict: 包含新结果的状态更新

    """

    print(f'Adding "D" to {state["aggregate"]}')

    return {"aggregate": ["D"]}

# 创建图

builder = StateGraph(State)

# 添加节点

builder.add_node("a", a)

builder.add_node("b", b)

builder.add_node("b_2", b_2)

builder.add_node("c", c)

builder.add_node("d", d, defer=True)  # 设置defer=True延迟执

# 添加边

builder.add_edge(START, "a")

builder.add_edge("a", "b")

builder.add_edge("a", "c")

builder.add_edge("b", "b_2")

builder.add_edge("b_2", "d")

builder.add_edge("c", "d")

builder.add_edge("d", END)

# 编译图

graph = builder.compile()

# 执行图

print("=== 开始执行工作流 ===")

result = graph.invoke({})

print("=== 执行结果 ===")

print(result)

"""
LangGraph普通边演示
普通边是直接连接两个节点的边，表示无条件地从一个节点跳转到另一个节点。
"""
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
定义状态
class GraphState(TypedDict):
    value: int
    step: str
定义节点函数
def node_a(state: GraphState) -> dict:
    """节点A"""
    print("执行节点A")
    return {"value": state["value"] + 1, "step": "A执行完毕"}
def node_b(state: GraphState) -> dict:
    """节点B"""
    print("执行节点B")
    return {"value": state["value"] * 2, "step": "B执行完毕"}
def node_c(state: GraphState) -> dict:
    """节点C"""
    print("执行节点C")
    return {"value": state["value"] - 1, "step": "C执行完毕"}
def main():
    """演示普通边"""
    print("= 普通边演示 =")
    # 创建图
    builder = StateGraph(GraphState)
    # 添加节点
    builder.add_node("node_a", node_a)
    builder.add_node("node_b", node_b)
    builder.add_node("node_c", node_c)
    # 添加普通边
    builder.add_edge(START, "node_a")  # 从开始到A
    builder.add_edge("node_a", "node_b")  # 从A到B
    builder.add_edge("node_b", "node_c")  # 从B到C
    builder.add_edge("node_c", END)       # 从C到结束
    # 编译图
    graph = builder.compile()
    # 执行图
    result = graph.invoke({"value": 1})
    print(f"执行结果: {result}\n")
if name == "main":
    main()

"""
LangGraph条件边演示
条件边根据当前状态动态决定下一个要执行的节点。
"""
from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

定义状态
class GraphState(TypedDict):
    value: int
    step: str
    
定义节点函数
def node_a(state: GraphState) -> dict:
    """节点A"""
    print("执行节点A")
    return {"value": state["value"] + 1, "step": "A执行完毕"}
def node_b(state: GraphState) -> dict:
    """节点B"""
    print("执行节点B")
    return {"value": state["value"] * 2, "step": "B执行完毕"}
def node_c(state: GraphState) -> dict:
    """节点C"""
    print("执行节点C")
    return {"value": state["value"] - 1, "step": "C执行完毕"}

条件边的路由函数
def route_condition(state: GraphState) -> Literal["node_b", "node_c"]:
    """根据value值决定路由到哪个节点"""
    if state["value"] % 2 == 0:
        return "node_b_alias"  # 偶数路由到节点B
    else:
        return "node_c_alias"  # 奇数路由到节点C

def main():
    """演示条件边"""
    print("= 条件边演示 =")
    # 创建图
    builder = StateGraph(GraphState)
    # 添加节点
    builder.add_node("node_a", node_a)
    builder.add_node("node_b", node_b)
    builder.add_node("node_c", node_c)
    # 添加边
    builder.add_edge(START, "node_a")  # 入口点
    # 添加条件边
    builder.add_conditional_edges(
        "node_a",  # 源节点
        route_condition,  # 路由函数
        {  # 路由映射
            "node_b_alias": "node_b",
            "node_c_alias": "node_c"
        }
    )
    # 从B和C到结束
    builder.add_edge("node_b", END)
    builder.add_edge("node_c", END)
    # 编译图
    graph = builder.compile()
    # 执行图 - 偶数情况
    print("输入值为偶数:")
    result = graph.invoke({"value": 2})
    print(f"执行结果: {result}")
    # 执行图 - 奇数情况
    print("\n输入值为奇数:")
    result = graph.invoke({"value": 1})
    print(f"执行结果: {result}\n")

if name == "main":
    main()

"""
LangGraph入口点演示
入口点定义了图开始执行的第一个节点。
"""
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
定义状态
class GraphState(TypedDict):
    value: int
    step: str
定义节点函数
def node_a(state: GraphState) -> dict:
    """节点A"""
    print("执行节点A")
    return {"value": state["value"] + 1, "step": "A执行完毕"}
def node_b(state: GraphState) -> dict:
    """节点B"""
    print("执行节点B")
    return {"value": state["value"] * 2, "step": "B执行完毕"}
def main():
    """演示入口点"""
    print("= 入口点演示 =")
    # 创建图
    builder = StateGraph(GraphState)
    # 添加节点
    builder.add_node("node_a", node_a)
    builder.add_node("node_b", node_b)
    # 设置入口点和出口点
    builder.add_edge(START, "node_a")
    builder.add_edge("node_a", "node_b")
    builder.add_edge("node_b", END)
    # 编译图
    graph = builder.compile()
    # 执行图
    result = graph.invoke({"value": 0})
    print(f"执行结果: {result}\n")
if name == "main":
    main()

"""
LangGraph条件入口点演示
条件入口点允许根据输入状态动态决定从哪个节点开始执行。
"""
from typing import Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END
定义状态
class GraphState(TypedDict):
    value: int
    step: str
定义节点函数
def node_a(state: GraphState) -> dict:
    """节点A"""
    print("执行节点A")
    return {"value": state["value"] + 1, "step": "A执行完毕"}
def node_b(state: GraphState) -> dict:
    """节点B"""
    print("执行节点B")
    return {"value": state["value"] * 2, "step": "B执行完毕"}
def node_d(state: GraphState) -> dict:
    """节点D"""
    print("执行节点D")
    return {"value": state["value"] + 10, "step": "D执行完毕"}
条件入口点的路由函数
def entry_condition(state: GraphState) -> Literal["node_a", "node_d"]:
    """根据输入值决定从哪个节点开始"""
    if state.get("value", 0) > 5:
        return "node_d"  # 大于5从节点D开始
    else:
        return "node_a"  # 否则从节点A开始
def main():
    """演示条件入口点"""
    print("= 条件入口点演示 =")
    # 创建图
    builder = StateGraph(GraphState)
    # 添加节点
    builder.add_node("node_a", node_a)
    builder.add_node("node_d", node_d)
    builder.add_node("node_b", node_b)
    # 添加条件入口点
    builder.add_conditional_edges(
        START,  # 起始点
        entry_condition,  # 路由函数
        {  # 路由映射
            "node_a": "node_a",
            "node_d": "node_d"
        }
    )
    # 添加普通边
    builder.add_edge("node_a", "node_b")
    builder.add_edge("node_d", "node_b")
    builder.add_edge("node_b", END)
    # 编译图
    graph = builder.compile()
    # 执行图 - 小于等于5的情况
    print("输入值小于等于5:")
    result = graph.invoke({"value": 3})
    print(f"执行结果: {result}")
    # 执行图 - 大于5的情况
    print("\n输入值大于5:")
    result = graph.invoke({"value": 10})
    print(f"执行结果: {result}\n")
if name == "main":
    main()