1、图的基础API

1.1、从节点更新图状态

更新节点状态有三种分别是
1、自定义添加消息函数

# 直接导入add运算符
from typing_extensions import TypedDict, Annotated
from operator import add
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add]
    extra_field: int
# 或者自定义
def add(left, right):
    return left + right

2、直接导入API的添加消息函数（内置规约器使用）

# 内置规约器使用
from langgraph.graph.message import add_messages
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    extra_field: int

3、继承消息系统状态管理类 MessagesState

# 状态管理
from langgraph.graph import MessagesState
class State(MessagesState):
    extra_field: int

使用状态管理的完整代码

"""加载环境变量"""
from dotenv import load_dotenv
load_dotenv()

"""定义状态"""
from langchain_core.messages import AnyMessage, HumanMessage, 
AIMessage
# 此处可以用上面三种方式替换
from langgraph.graph import MessagesState
class State(MessagesState):
    extra_field: int

def node(state: State):
    new_message = AIMessage("Hello!")
    """注意点"""
    return {"messages": [new_message], "extra_field": 10}

# 构建图
from langgraph.graph import START, StateGraph
graph = StateGraph(State).add_node(node).add_edge(START, "node").compile()

result = graph.invoke({"messages": [HumanMessage("Hi")]})
print(result)

1.2、创建节点序列

如果是顺序创建图，可以使用add_sequence，代替原来的add_node、add_edge

from typing_extensions import TypedDict

class State(TypedDict):
    value_1: str
    value_2: int

def step_1(state: State):
    return {"value_1": "a"}

def step_2(state: State):
    current_value_1 = state["value_1"]
    return {"value_1": f"{current_value_1} + b"}

def step_3(state: State):
    return {"value_2": 10}

from langgraph.graph import START, StateGraph
# graph_builder = StateGraph(State)
# Add nodes
# graph_builder.add_node(step_1)
# graph_builder.add_node(step_2)
# graph_builder.add_node(step_3)
#
# # Add edges
# graph_builder.add_edge(START, "step_1")
# graph_builder.add_edge("step_1", "step_2")
# graph_builder.add_edge("step_2", "step_3")
# # 编译
# graph = graph_builder.compile()

#  add_sequence替换上面的add_edge和add_node
graph_builder = StateGraph(State).add_sequence([step_1, step_2, step_3])
graph_builder.add_edge(START, "step_1")
graph = graph_builder.compile()
print(graph.invoke({"value_1": "c"}))

1.3、并行序列创建

多分支节点需要设置thread_id，这种并行化是通过扇出和扇入机制实现的，利用标准边和条件边。

import operator
import time
from typing import Annotated, Any
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

def b_2(state: State):
    print(f'Adding "B_2" to {state["aggregate"]}')
    return {"aggregate": ["B_2"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(b_2)
builder.add_node(c)
builder.add_node(d)

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", "c"], "d")
builder.add_edge("d", END)
graph = builder.compile()
// 配置thread_id实现并行执行
print(graph.invoke({"aggregate": []}, {"configurable": {"thread_id": "10086"}}))

1.4、条件分支

import operator
from typing import Annotated, Sequence
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]
    which: str

def a(state: State):
    print(f'Adding "A" to {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Adding "B" to {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Adding "C" to {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Adding "D" to {state["aggregate"]}')
    return {"aggregate": ["D"]}

def e(state: State):
    print(f'Adding "E" to {state["aggregate"]}')
    return {"aggregate": ["E"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)
builder.add_node(e)
builder.add_edge(START, "a")

"""可自定义路由走单边"""
def route_bc_or_cd(state: State) -> Sequence[str]:
    if state["which"] == "cd":
        return ["c", "d"]
    return ["b", "c"]

intermediates = ["b", "c", "d"]
# 添加a到b、c、d节点的边, 并添加条件边
builder.add_conditional_edges(
    "a",
    route_bc_or_cd,
    path_map = intermediates,
)

# 添加到e节点的边
for node in intermediates:
    print('intermediates',intermediates)
    builder.add_edge(node, "e")

builder.add_edge("e", END)
graph = builder.compile()

# print(graph.invoke({"aggregate": [], "which": "bc"}))
print(graph.invoke({"aggregate": [], "which": "cd"}))

1.5、创建用于并行执行的MapReduce分支

Map-reduce 操作对于高效的任务分解和并行处理至关重要。这种方法涉及将任务分解成更小的子任务，并行处理每个子任务，然后汇总所有已完成子任务的结果。

代码示例图

import operator
from typing import Annotated
from langchain_openai import ChatOpenAI
from typing_extensions import TypedDict
from langgraph.types import Send
from langgraph.graph import END, StateGraph, START
from pydantic import BaseModel, Field

"""
生成与{topic}相关的1到3个示例的逗号分隔列表。
生成一个关于{subject}的笑话
下面是一些关于{topic}的笑话。选择最好的一个！返回最佳的ID。
{jokes}
"""
subjects_prompt = """Generate a comma separated list of between 1 and 3 examples related to: {topic}."""
joke_prompt = """Generate a joke about {subject}"""
best_joke_prompt = """Below are a bunch of jokes about {topic}. Select the best one! Return the ID of the best one.
{jokes}"""

class Subjects(BaseModel):
    subjects: list[str]

class Joke(BaseModel):
    joke: str

class BestJoke(BaseModel):
    id: int = Field(description="Index of the best joke, starting with 0")

from dotenv import load_dotenv
load_dotenv()
model = ChatOpenAI(model="gpt-4o-mini")

class OverallState(TypedDict):
    topic: str
    subjects: list
    jokes: Annotated[list, operator.add]
    best_selected_joke: str

class JokeState(TypedDict):
    subject: str

def generate_topics(state: OverallState):
    prompt = subjects_prompt.format(topic=state["topic"])
    # 结构化输出
    response = model.with_structured_output(Subjects).invoke(prompt)
    return {"subjects": response.subjects}

def generate_joke(state: JokeState):
    prompt = joke_prompt.format(subject=state["subject"])
    response = model.with_structured_output(Joke).invoke(prompt)
    return {"jokes": [response.joke]}

def continue_to_jokes(state: OverallState):
    return [Send("generate_joke", {"subject": s}) for s in state["subjects"]]
"""for循环与列表推导式 以时间换空间的方式   python没有真正意义上的多线程"""

def best_joke(state: OverallState):
    jokes = "\n\n".join(state["jokes"])
    prompt = best_joke_prompt.format(topic=state["topic"], jokes=jokes)
    response = model.with_structured_output(BestJoke).invoke(prompt)
    return {"best_selected_joke": state["jokes"][response.id]}

graph = StateGraph(OverallState)
graph.add_node("generate_topics", generate_topics)
graph.add_node("generate_joke", generate_joke)
graph.add_node("best_joke", best_joke)

graph.add_edge(START, "generate_topics")
graph.add_conditional_edges("generate_topics", continue_to_jokes, ["generate_joke"])
graph.add_edge("generate_joke", "best_joke")
graph.add_edge("best_joke", END)
app = graph.compile()

# Call the graph: here we call it to generate a list of jokes
for s in app.stream({"topic": "animals"}):
    print(s)

1.6、循环

1、设置最大循环次数recursion_limit（默认24次退出循环）
2、添加循环条件add_conditional_edges

import operator
from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import StateGraph, START, END

class State(TypedDict):
    aggregate: Annotated[list, operator.add]

def a(state: State):
    print(f'Node A sees {state["aggregate"]}')
    return {"aggregate": ["A"]}

def b(state: State):
    print(f'Node B sees {state["aggregate"]}')
    return {"aggregate": ["B"]}

def c(state: State):
    print(f'Node C sees {state["aggregate"]}')
    return {"aggregate": ["C"]}

def d(state: State):
    print(f'Node D sees {state["aggregate"]}')
    return {"aggregate": ["D"]}

builder = StateGraph(State)
builder.add_node(a)
builder.add_node(b)
builder.add_node(c)
builder.add_node(d)

def route(state: State) -> Literal["b", END]:
    print(len(state["aggregate"]))
    if len(state["aggregate"]) < 100:
        return "b"
    else:
        return END

builder.add_edge(START, "a")
builder.add_conditional_edges("a", route)
builder.add_edge("b", "c")
builder.add_edge("b", "d")
builder.add_edge(["c", "d"], "a")
graph = builder.compile()

from langgraph.errors import GraphRecursionError
try:
    # 不设置默认最多24步退出循环，需要try catch
    print(graph.invoke({"aggregate": []}, {"recursion_limit": 25}))
except GraphRecursionError:
    print("Recursion Error")

1.7、可视化图

1、使用 Pyppeteer（在线方式，需要翻墙）
2、安装 Graphviz库（不同系统需要版本适配）

from IPython.display import Image, display
from langchain_core.runnables.graph import CurveStyle, MermaidDrawMethod, NodeStyles

display(
    Image(
        builder.compile().get_graph().draw_png(
            draw_method=MermaidDrawMethod.API,
            output_file_path='./img/示例6.png'
        )
    )
)

2、对图的细粒度控制

2.1、在图和子图中更新状态

可以将控制流（边）和状态更新（节点）结合起来使用，可以通过 Command 对象来实现

import random
from typing_extensions import TypedDict, Literal
from langgraph.graph import StateGraph, START
from langgraph.types import Command
import operator
from typing_extensions import Annotated

class State(TypedDict):
    foo: Annotated[str, operator.add]

def node_a(state: State):
    print("Called A")
    value = random.choice(["a", "b"])
    print(value)
    if value == "a":
        goto = "node_b"
    else:
        goto = "node_c"

    return Command(
        update={"foo": value},
        goto=goto,
        graph=Command.PARENT, # parent 返回到父图继续执行
    )

subgraph = StateGraph(State).add_node(node_a).add_edge(START, "node_a").compile()

def node_b(state: State):
    print("Called B")
    return {"foo": "b"}

def node_c(state: State):
    print("Called C")
    return {"foo": "c"}

builder = StateGraph(State)
builder.add_edge(START, "subgraph")
builder.add_node("subgraph", subgraph)
builder.add_node(node_b)
builder.add_node(node_c)
graph = builder.compile()

print(graph.invoke({"foo": "bc"}))

2.2、对图添加配置

1、不同能力模型的配置
2、不同提示词面板的配置

import operator
from typing import Annotated, Sequence
from typing_extensions import TypedDict, Optional
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import END, StateGraph, START
from langchain_openai import ChatOpenAI
from langchain_core.runnables.config import RunnableConfig

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
# 模型配置
openai_model = ChatOpenAI(model="gpt-4o-mini",api_key='',base_url="")
Anthropic_model = ChatOpenAI(model="claude-3-5-sonnet-latest",api_key="")
models = {
    "anthropic": Anthropic_model,
    "openai": openai_model,
}

from langchain_core.messages import SystemMessage
# We can define a config schema to specify the configuration options for the graph
# A config schema is useful for indicating which fields are available in the configurable dict inside the config
class ConfigSchema(TypedDict):
    model: Optional[str]
    system_message: Optional[str]

def _call_model(state: AgentState, config: RunnableConfig):
    model_name = config["configurable"].get("model", "anthropic")
    model = models[model_name]
    messages = state["messages"]
    if "system_message" in config["configurable"]:
        messages = [SystemMessage(content=config["configurable"]["system_message"])] + messages
    response = model.invoke(messages)
    return {"messages": [response]}
# Define a new graph
builder = StateGraph(AgentState)
builder.add_node("model", _call_model)
builder.add_edge(START, "model")
builder.add_edge("model", END)

graph = builder.compile()
# anthropic模型
config = {"configurable": {"model": "anthropic"}}
print(graph.invoke({"messages": [HumanMessage(content="who are you?")]}, config=config))
# openai模型
config = {"configurable": {"system_message": "respond in Chinese","model": "openai"}}
print(graph.invoke({"messages": [HumanMessage(content="who are you?")]}, config=config))

2.3、节点重试策略

一般调用API超时，或者查询数据库失败，或者调用大模型失败的情况下使用
正对决策性的节点有几种处理方式
1、针对不同的场景，需要评估哪个模型更适用这个任务
2、通过写提示词的方式进行干预（需要多次迭代）
3、对已经获得的结果进行过滤（正则）
4、重试策略

import operator
import sqlite3
from typing import Annotated, Sequence
from typing_extensions import TypedDict
from langchain_core.messages import BaseMessage, HumanMessage
from langgraph.graph import END, StateGraph, START
from langchain_community.utilities import SQLDatabase
from langchain_core.messages import AIMessage
from langgraph.pregel import RetryPolicy
# 重试策略
RetryPolicy()

db = SQLDatabase.from_uri("sqlite:///:memory:")
# 创建表
db.run("CREATE TABLE Artist (ArtistId INTEGER PRIMARY KEY, Name NVARCHAR(120));")
# 表中添加数据
db.run("""
INSERT INTO Artist (Name) VALUES 
('Louis Armstrong'),
('Duke Ellington'),
('Ella Fitzgerald'),
('Charlie Parker'),
""")
from langchain_openai import ChatOpenAI
from dotenv import load_dotenv

load_dotenv()
model = ChatOpenAI(model_name="gpt-4o-mini")
class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], operator.add]
# 正则表达式兜底
def query_database(state):
    query_result = db.run(state['messages'][1].content)
    print('query_result', query_result)
    '''```sql\nSELECT * FROM Artist LIMIT 10;\n```
    正则表达式的形式
    '''
    # query_result = db.run("SELECT * FROM Artist LIMIT 5;")
    print('query_result:', query_result)
    return {"messages": [AIMessage(content=query_result)]}

def call_model(state):
    response = model.invoke(state["messages"])
    print('response:', response)
    return {"messages": [response]}
builder = StateGraph(AgentState)
#重试策略
builder.add_node("query_database", query_database, retry=RetryPolicy(retry_on=sqlite3.OperationalError))
builder.add_node("model", call_model, retry=RetryPolicy(max_attempts=5))

builder.add_edge(START, "model")
builder.add_edge("model", "query_database")
builder.add_edge("query_database", END)
graph = builder.compile()
# 提示词要完善
result = graph.invoke({"messages": [HumanMessage(
    content="查询Artist表格中前十位艺术家，我的表名叫Artist，表结构信息只有ArtistId与Name两个字段，只返回SQL查询语句，不要返回其他内容，特别是TOP、```sql\n不要出现")]})
print('result：', result)

3、持久化

3.1、图添加持久化

编译图时添加检查点来设置其状态的持久化
关键代码

from langgraph.checkpoint.memory import MemorySaver
memory = MemorySaver()
# 添加检查点实现持久化
graph = builder.compile(checkpointer=memory)
config = {"configurable": {"thread_id": "1"}}
# 待验证持久化
input_message = {"role": "user", "content": "hi! I'm Cat"}
for chunk in graph.stream({"messages": [input_message]}, config, stream_mode="values"):
     # 打印出每部执行的的内容
    chunk["messages"][-1].pretty_print()

3.2、子图持久化

只需要在编译父图时传递一个检查点器。LangGraph 会自动将检查点器传播到子子图中。
关键代码

class SubgraphState(TypedDict):
    foo: str  # note that this key is shared with the parent graph state
    bar: str

def subgraph_node_1(state: SubgraphState):
    return {"bar": "bar"}

def subgraph_node_2(state: SubgraphState):
    # note that this node is using a state key ('bar') that is only available in the subgraph
    # and is sending update on the shared state key ('foo')
    return {"foo": state["foo"] + state["bar"]}
subgraph_builder = StateGraph(SubgraphState)
subgraph_builder.add_node(subgraph_node_1)
subgraph_builder.add_node(subgraph_node_2)
subgraph_builder.add_edge(START, "subgraph_node_1")
subgraph_builder.add_edge("subgraph_node_1", "subgraph_node_2")
subgraph = subgraph_builder.compile()
# 父图
class State(TypedDict):
    foo: str
def node_1(state: State):
    return {"foo": "hi! " + state["foo"]}

builder = StateGraph(State)
builder.add_node("node_1", node_1)
# note that we're adding the compiled subgraph as a node to the parent graph
builder.add_node("node_2", subgraph)
builder.add_edge(START, "node_1")
builder.add_edge("node_1", "node_2")
# 使用内存检查点器（ MemorySaver ）编译该图
checkpointer = MemorySaver()
graph = builder.compile(checkpointer=checkpointer)

# 验证持久性是否生效
config = {"configurable": {"thread_id": "1"}}
for _, chunk in graph.stream({"foo": "foo"}, config, subgraphs=True):
    print(_,chunk)
# 通过使用与调用图相同的配置来查看父图状态。
print(graph.get_state(config).values)   # {'foo': 'hi! foobar'}
print('###')
# 检查父图的状态历史，找到在从node_2（包含子图的节点）返回结果之前的状态快照：
state_with_subgraph = [s for s in graph.get_state_history(config) if s.next == ("node_2",)][0]
print(state_with_subgraph)
print('####')

# 检索子图状态的配置
subgraph_config = state_with_subgraph.tasks[0].state
print(subgraph_config)
print('#####')
print(graph.get_state(subgraph_config).values)

3.3、跨线程持久化

将用户信息（例如姓名或偏好）存储在共享内存中，并在新的对话线程中重复使用这些信息。

# 加载环境变量（通常用于存储API密钥等敏感信息）
from dotenv import load_dotenv
load_dotenv()
# 导入内存存储模块和OpenAI嵌入模型
from langgraph.store.memory import InMemoryStore
from langchain_openai import OpenAIEmbeddings
# 初始化内存存储系统，配置嵌入模型和维度
in_memory_store = InMemoryStore(
    index={
        "embed": OpenAIEmbeddings(model="text-embedding-3-small"),  # 使用OpenAI的小型嵌入模型
        "dims": 1536,  # 对应text-embedding-3-small模型的输出维度
    }
)
# 导入必要模块
import uuid
from langchain_core.runnables import RunnableConfig
from langgraph.graph import StateGraph, MessagesState, START
from langgraph.checkpoint.memory import MemorySaver
from langgraph.store.base import BaseStore
# 初始化OpenAI聊天模型（使用最新的gpt-4o-mini模型）
from langchain_openai import ChatOpenAI
model = ChatOpenAI(model="gpt-4o-mini")

# 定义核心处理函数
def call_model(state: MessagesState, config: RunnableConfig, *, store: BaseStore):
    """对话处理核心逻辑，包含记忆存储和检索功能"""
    # 从配置中获取用户ID，创建专属的命名空间
    user_id = config["configurable"]["user_id"]
    namespace = ("memories", user_id)
    # 在存储中搜索与当前对话相关的记忆
    memories = store.search(namespace, query=str(state["messages"][-1].content))
    # 将记忆数据转换为字符串格式
    info = "\n".join([d.value["data"] for d in memories])
    # 构建系统提示，包含用户记忆信息
    system_msg = f"You are a helpful assistant talking to the user. User info: {info}"
    # 检查是否需要存储新记忆
    last_message = state["messages"][-1]
    print('last_message：',last_message)
    if "remember" in last_message.content.lower():
        # 生成并存储新记忆（示例记忆内容）
        memory = "User name is Cat"
        store.put(namespace, str(uuid.uuid4()), {"data": memory})  # 使用UUID作为唯一键
    # 调用AI模型生成回复（结合系统提示和对话历史）
    response = model.invoke(
        [{"role": "system", "content": system_msg}] + state["messages"]
    )
    return {"messages": response}
# 构建状态图工作流
builder = StateGraph(MessagesState)
builder.add_node("call_model", call_model)  # 添加处理节点
builder.add_edge(START, "call_model")  # 设置起始节点

# 编译完整的工作流图，配置内存检查点和存储
graph = builder.compile(
    checkpointer=MemorySaver(),  # 用于保存对话状态的检查点
    store=in_memory_store  # 使用之前配置的内存存储
)

# 测试场景1：存储记忆
config = {"configurable": {"thread_id": "1", "user_id": "1"}}  # 用户1的对话配置
input_message = {"role": "user", "content": "Hi! Remember: my name is Cat"}
print("第一次对话（存储记忆）:")
for chunk in graph.stream({"messages": [input_message]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()  # 格式化输出模型回复

# 查看存储的记忆
print("\n存储的记忆内容:")
for memory in in_memory_store.search(("memories", "1")):
    print(memory.value)

# 测试场景2：读取记忆
config = {"configurable": {"thread_id": "3", "user_id": "1"}}  # 用户2的对话配置
input_message = {"role": "user", "content": "what is my name?"}
print("\n第二次对话（读取记忆）:")

# 跨线程读取到记忆
for chunk in graph.stream({"messages": [input_message]}, config, stream_mode="values"):
    chunk["messages"][-1].pretty_print()

4、内存管理

1、记录消息 但是仅保留6-10轮次的会话然后清空，或者记录token数量超过某个阈值清空
2、做摘要，5-6轮做一次摘要，但是可能存在记忆污染问题，影响后序工具调用的结果，要做提示词约束迭代

4.1、历史消息管理

"""自定义只记住最后一条消息"""
def filter_messages(messages: list):
    print('messages: ',messages)
    print('-1',messages[-1:])
    return messages[-1:]
def call_model(state: MessagesState):
    messages = filter_messages(state["messages"])
    response = bound_model.invoke(messages)
    return {"messages": response}
# Define a new graph
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    ["action", END],
)
workflow.add_edge("action", "agent")
app = workflow.compile(checkpointer=memory)

4.2、删除消息

1、手动删除

# 当前线程状态
messages = app.get_state(config).values["messages"]
print('当前',messages)
# 手动删除消息
from langchain_core.messages import RemoveMessage
app.update_state(config, {"messages": RemoveMessage(id=messages[0].id)})

messages = app.get_state(config).values["messages"]
print('现在',messages)

2、程序删除消息

'自定义删除方法'
def delete_messages(state):
    messages = state["messages"]
    print('############@',messages)
    if len(messages) > 3:
        return {"messages": [RemoveMessage(id=m.id) for m in messages[:-3]]}
# We need to modify the logic to call delete_messages rather than end right away
def should_continue(state: MessagesState) -> Literal["action", "delete_messages"]:
    last_message = state["messages"][-1]
    if not last_message.tool_calls:
        return "delete_messages"
    return "action"
# Define a new graph
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("action", tool_node)
# This is our new node we're defining
workflow.add_node(delete_messages)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
)
workflow.add_edge("action", "agent")
# This is the new edge we're adding: after we delete messages, we finish
workflow.add_edge("delete_messages", END)
app = workflow.compile(checkpointer=memory)

4.3、对历史消息做摘要

def summarize_conversation(state: State):
    summary = state.get("summary", "")
    if summary:
        summary_message = (
            f"This is summary of the conversation to date: {summary}\n\n"
            "Extend the summary by taking into account the new messages above:"
        )
    else:
        summary_message = "Create a summary of the conversation above:"

    messages = state["messages"] + [HumanMessage(content=summary_message)]
    print('messagesall:',messages)
    response = model.invoke(messages)
    delete_messages = [RemoveMessage(id=m.id) for m in state["messages"][:-2]]
    print('delete_messages',delete_messages)
    print('@@@@@',{"summary": response.content, "messages": delete_messages})
    return {"summary": response.content, "messages": delete_messages}

workflow = StateGraph(State)
workflow.add_node("conversation", call_model)
workflow.add_node(summarize_conversation)
workflow.add_edge(START, "conversation")
workflow.add_conditional_edges(
    "conversation",
    should_continue,
)

5、人工介入

5.1、InjectedState

作用：允许工具直接访问 Agent 状态图（如对话历史、用户上下文），而不暴露给大语言模型（LLM），防止模型误生成或篡改敏感状态。
机制：
工具模式（schema）生成时自动过滤该参数，LLM 调用工具时无需提供；
运行时由 ToolNode 自动从当前状态图中提取并注入对应值。

from typing import Annotated
from langchain_core.tools import tool
from langgraph.prebuilt import InjectedState
# 通过InjectedState自动注入当前状态
@tool
def get_history(query: str, state: Annotated[dict, InjectedState("messages")]) -> str:
    return f"历史消息数：{len(state)}"
tools = [get_history]
tool_node = ToolNode(tools)

5.2、Interrupt

作用：Interrupt 是 LangGraph 内置的一个函数，用于在节点执行过程中主动暂停整个工作流。
用法：
1、暂停执行：立即停止当前图的运行，不继续执行后续节点
2、保存现场：与 Checkpointer 配合，将当前 State 和断点位置持久化
3、等待恢复：无限期等待外部输入，直到收到 Command(resume=...) 指令

from langgraph.types import interrupt
def my_node(state: State):
    # 图会在这里暂停，并向外抛出 "请问您的年龄是？"
    user_input = interrupt("请问您的年龄是？")
    # 当用户输入恢复后，程序从这里继续执行
    return {"age": user_input}
#  手动使用用户输入更新我们的图状态
for event in graph.stream(
    Command(resume="go to my_node!"), thread, stream_mode="updates"
):
    print(event)
    print("\n")
print(graph.get_state(thread).values)

5.3、编辑图状态（update）

# Set up memory
memory = MemorySaver()
# Add
graph = builder.compile(checkpointer=memory, interrupt_before=["step_2"])
# Input
initial_input = {"input": "hello world"}
# Thread
thread = {"configurable": {"thread_id": "1"}}
# Run the graph until the first interruption
for event in graph.stream(initial_input, thread, stream_mode="values"):
    print(event)
print("Current state!")
print(graph.get_state(thread).values)
# 使用update方法来更新图状态
graph.update_state(thread, {"input": "hello universe!"})
print("---\n---\nUpdated state!")
print(graph.get_state(thread).values)
# Continue the graph execution
for event in graph.stream(None, thread, stream_mode="values"):
    print(event)

5.4、ToolNode（工具使用）

@tool
def get_weather(location: str):
    """Call to get the current weather."""
    if location.lower() in ["sf", "san francisco"]:
        return "It's 60 degrees and foggy."
    else:
        return "It's 90 degrees and sunny."
@tool
def get_coolest_cities():
    """Get a list of coolest cities"""
    return "nyc, sf"
tools = [get_weather, get_coolest_cities]
tool_node = ToolNode(tools)
message_with_single_tool_call = AIMessage(
    content="",
    tool_calls=[
        {
            "name": "get_weather",
            "args": {"location": "sf"},
            "id": "tool_call_id",
            "type": "tool_call",
        }
    ],
)
print(tool_node.invoke({"messages": [message_with_single_tool_call]}))

预构建使用

tool_node = ToolNode([get_weather])
from langchain_openai import ChatOpenAI
# 绑定大模型
model_with_tools = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([get_weather])

6、多智能体

6.1、多智能体架构

多智能体架构图

1、Network：每个代理可以与其他所有代理通信。任何代理都可以决定下一步呼叫哪个其他代理。
2、Supervisor：每个代理与一个单一的监督者代理通信。监督者代理决定下一步应该呼叫哪个代理。
3、Supervisor (tool-calling)：这是一种监督者架构的特殊情况。单个代理可以表示为工具。在这种情况下，监督者代理使用一个调用工具的 LLM 来决定要调用哪个代理工具，以及传递给这些代理的参数。
4、Hierarchical：您可以定义一个具有监督者监督的多代理系统。这是监督者架构的泛化，并允许更复杂的控制流程。
5、Custom multi-agent workflow：每个智能体仅与部分智能体通信。流程的部分是确定的，只有某些智能体可以决定下一步调用哪些其他智能体。