LangGraph 状态编排

from typing import TypedDict
from langgraph.graph import StateGraph, START, END

# 1. 定义状态：全图共享的数据契约
class State(TypedDict):
    topic: str          # 输入：选题
    outline: str        # 中间产物：大纲
    article: str        # 输出：成稿

# 2. 节点：纯函数，返回"要更新的字段"
def make_outline(state: State) -> dict:
    return {"outline": f"《{state['topic']}》三段式大纲"}

def write_article(state: State) -> dict:
    return {"article": f"正文按大纲展开：{state['outline']}"}

# 3. 组装图
builder = StateGraph(State)
builder.add_node("outline", make_outline)
builder.add_node("write", write_article)
builder.add_edge(START, "outline")       # 入口
builder.add_edge("outline", "write")     # 固定边
builder.add_edge("write", END)           # 出口

# 4. 编译：得到可调用的 Runnable
graph = builder.compile()

# 5. 调用
result = graph.invoke({"topic": "向量数据库选型"})
print(result["article"])

from langgraph.graph import StateGraph, MessagesState, START
from langgraph.graph.message import add_messages
from typing import Annotated, TypedDict

# 手写等价形式，理解 MessagesState 到底做了什么
class MyMessagesState(TypedDict):
    messages: Annotated[list, add_messages]   # add_messages = 追加式 Reducer

# 大多数时候直接用内置的
from langgraph.graph import StateGraph, MessagesState

builder = StateGraph(MessagesState)
builder.add_node("chat", lambda state: {"messages": [...]})
builder.add_edge(START, "chat")
graph = builder.compile()

import operator
from typing import Annotated, TypedDict
from langgraph.graph.message import add_messages

def merge_scores(left: float, right: float) -> float:
    """并行节点各产出置信度，合并时取平均。"""
    return round((left + right) / 2, 4)

class ResearchState(TypedDict):
    topic: str
    # 无 Annotated = 默认覆盖（确认只有一个写者时使用）
    final_answer: str
    # 列表追加：两个来源节点各返回若干条，自动拼接不丢
    notes: Annotated[list[str], operator.add]
    # 自定义合并：并行评分取平均
    score: Annotated[float, merge_scores]
    # 消息历史：追加 + 同 id 替换
    messages: Annotated[list, add_messages]

from typing import Literal

def route_by_quality(state: State) -> Literal["revise", "publish"]:
    if state["score"] < 0.8:
        return "revise"
    return "publish"

builder.add_conditional_edges("review", route_by_quality, ["revise", "publish"])

from typing import Literal
from langgraph.types import Command

def review(state: State) -> Command[Literal["revise", "publish"]]:
    score = evaluate(state["draft"])
    if score < 0.8:
        # 去向 + 一次状态更新，二合一
        return Command(goto="revise", update={"score": score, "rounds": state["rounds"] + 1})
    return Command(goto="publish", update={"score": score})

from langgraph.checkpoint.memory import MemorySaver
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.checkpoint.postgres import PostgresSaver

# 开发期：内存存档，进程退出即失
graph = builder.compile(checkpointer=MemorySaver())

# 单机持久化：SQLite 文件，重启不丢
with SqliteSaver.from_conn_string("checkpoints.db") as saver:
    graph = builder.compile(checkpointer=saver)

# 生产：Postgres（异步用 AsyncPostgresSaver，用法一致）
with PostgresSaver.from_conn_string("postgresql://user:pwd@host/db") as saver:
    saver.setup()                       # 首次运行建表，之后可省
    graph = builder.compile(checkpointer=saver)

config = {"configurable": {"thread_id": "conv-9812"}}

# 第一轮：告诉它名字
graph.invoke({"messages": [{"role": "user", "content": "我叫林可"}]}, config)

# 第二轮：只传新消息，历史已被自动恢复
graph.invoke({"messages": [{"role": "user", "content": "我叫什么？"}]}, config)
# 模型能正确回答“林可”

# 检查某个线程的当前状态
snap = graph.get_state(config)
print(snap.values)    # 当前完整状态（含全部消息）
print(snap.next)      # 下一步待执行的节点；非空 = 图正停在中断处

from langgraph.types import interrupt, Command

def human_review(state: State) -> dict:
    # 中断点：图在此暂停，payload 会原样交给调用方展示给人类
    decision = interrupt({
        "question": "是否执行转账？",
        "amount": state["amount"],
        "payee": state["payee"],
    })
    # 只有 resume 之后才会走到这里，decision 就是人类的答复
    if decision == "approve":
        return {"status": "已执行转账"}
    return {"status": "已拒绝"}

config = {"configurable": {"thread_id": "tx-1001"}}

# 第一次调用：停在 human_review，返回值带 __interrupt__ 信息
result = graph.invoke({"amount": 500, "payee": "张三"}, config)
print(result["__interrupt__"])     # 人类看到题目，去审批……

# 审批通过后恢复执行（可以发生在另一个进程）
graph.invoke(Command(resume="approve"), config)

history = list(graph.get_state_history(config))
for snap in history:
    ckpt = snap.config["configurable"]["checkpoint_id"]
    print(ckpt, snap.next, len(snap.values.get("messages", [])))

# 选一个历史快照，不传输入（None = 从该点继续执行）
to_replay = history[3]
graph.invoke(None, to_replay.config)
# 该线程被“回退”到那个时刻，之后的执行从这里重新展开

old_ckpt = history[3].config["configurable"]["checkpoint_id"]
fork_config = {"configurable": {
    "thread_id": "fork-branch-a",     # 新线程：不影响原对话
    "checkpoint_id": old_ckpt,          # 但从旧存档点出发
}}
graph.invoke({"messages": [{"role": "user", "content": "换个角度重写"}]}, fork_config)
# 原线程 history[3] 之后的历史保持不变，两条时间线互不干扰

from langgraph.types import Send

def fan_out(state: State) -> list[Send]:
    # topics 有几个就分发几份，并行度运行时才确定
    return [Send("search", {"topic": t}) for t in state["topics"]]

def search(payload: dict) -> dict:
    # 注意：Send 的载荷直接成为节点输入，不经过全局 State
    return {"notes": [f"{payload['topic']} 的检索结果"]}   # notes 声明 operator.add

builder.add_conditional_edges("plan", fan_out, ["search"])
builder.add_edge("search", "aggregate")   # 全部 search 实例结束后汇入 aggregate

research = research_builder.compile()      # 一张完整的研究子图
writer = writer_builder.compile()          # 一张完整的写作子图

parent = StateGraph(ParentState)
parent.add_node("research", research)      # 图直接当节点
parent.add_node("writer", writer)
parent.add_edge(START, "research")
parent.add_edge("research", "writer")
parent.add_edge("writer", END)

# 父图 State 必须包含子图需要的同名字段（如 notes、messages）；
# 子图对这些字段的更新经 Reducer 合并回父图状态

from langgraph.func import entrypoint, task

@task
def search(q: str) -> str:
    return retriever.invoke(q)

@task
def write(notes: list[str]) -> str:
    return llm.invoke("根据以下笔记写报告：\n" + "\n".join(notes)).content

@entrypoint(checkpointer=saver)      # 存档依然可用，interrupt 也支持
def report(inputs: dict) -> dict:
    found = [search(q).result() for q in inputs["questions"]]
    if not any(found):              # 普通 Python 条件，不用建边
        found = [search("默认主题").result()]
    return {"report": write(found).result()}

# 节点级：updates，看每一步谁改了什么
for chunk in graph.stream(inputs, config, stream_mode="updates"):
    for node, update in chunk.items():
        print(f"[{node}] {update}")

# token 级：messages（消息块与元数据成对产出）
for msg_chunk, meta in graph.stream(inputs, config, stream_mode="messages"):
    print(msg_chunk.content, end="", flush=True)

# 一次订阅多种模式：收到 (mode, chunk) 二元组
for mode, chunk in graph.stream(inputs, config,
                                stream_mode=["updates", "messages"]):
    if mode == "messages":
        ...   # 渲染 token
    else:
        ...   # 更新步骤面板

from langgraph.config import get_stream_writer

def research(state: State) -> dict:
    writer = get_stream_writer()
    writer({"stage": "searching", "progress": 0.3})
    notes = do_search(state["topic"])
    writer({"stage": "summarizing", "progress": 0.8})
    summary = do_summarize(notes)
    writer({"stage": "done", "progress": 1.0})
    return {"notes": notes, "summary": summary}

# 调用侧：stream_mode="custom" 接收这些事件
for event in graph.stream(inputs, config, stream_mode="custom"):
    print(event)    # {'stage': 'searching', 'progress': 0.3} ...

from typing import Literal
from langchain.chat_models import init_chat_model
from langchain_core.tools import tool
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.prebuilt import ToolNode

@tool
def add(a: int, b: int) -> int:
    """两数相加。"""
    return a + b

@tool
def multiply(a: int, b: int) -> int:
    """两数相乘。"""
    return a * b

tools = [add, multiply]
llm = init_chat_model("openai:gpt-4o-mini").bind_tools(tools)

def call_model(state: MessagesState) -> dict:
    # 模型节点只做一件事：带着全部历史调用模型
    return {"messages": [llm.invoke(state["messages"])]}

def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
    last = state["messages"][-1]
    if last.tool_calls:        # 模型生成了工具调用请求
        return "tools"
    return END                 # 纯文本回答，循环结束

builder = StateGraph(MessagesState)
builder.add_node("agent", call_model)
builder.add_node("tools", ToolNode(tools))   # 预置节点：执行工具并回填结果
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue, ["tools", END])
builder.add_edge("tools", "agent")           # 工具结果送回模型，形成循环

graph = builder.compile(checkpointer=MemorySaver())

config = {"configurable": {"thread_id": "demo-1"}}
result = graph.invoke(
    {"messages": [{"role": "user", "content": "先算 23 乘 7，再加 100"}]},
    config,
)
for msg in result["messages"]:
    print(msg.type, ":", msg.content)