LangGraph 实战：StateGraph、手写 ReAct 循环与 Map-Reduce 摘要

from typing import Annotated
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages

class State(TypedDict):
    # Messages 的类型是 "list"。`add_messages` 函数
    # 在注解中定义了如何更新这个状态键
    # （在这种情况下，它会将消息追加到列表中，而不是覆盖它们）
    messages: Annotated[list, add_messages]

# 不使用 Annotated（默认行为：覆盖）
state = {"messages": ["消息1"]}
state["messages"] = ["消息2"]
# 结果：["消息2"]

# 使用 Annotated + add_messages
state = {"messages": ["消息1"]}
# 新消息到来时，add_messages 将其追加到列表
# 结果：["消息1", "消息2"]

class MessagesState(TypedDict):
    messages: Annotated[list, add_messages]

import os
from langgraph.graph import MessagesState, StateGraph, START, END
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(
    model="qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
)

def chatbot_node(state: MessagesState):
    # 取出当前消息，调用 LLM；返回值会自动追加进 messages
    response = llm.invoke(state["messages"])
    return {"messages": [response]}

graph = StateGraph(MessagesState)
graph.add_node("chatbot", chatbot_node)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
app = graph.compile()

result = app.invoke({"messages": [HumanMessage(content="你好！今天能帮我什么？")]})
for message in result["messages"]:
    print(f"{message.__class__.__name__}: {message.content}")

class CustomMessagesState(MessagesState):
    user_id: str
    session_id: int

graph = StateGraph(CustomMessagesState)

from typing import Literal
from langgraph.graph import MessagesState, StateGraph, START, END
from langchain_core.messages import HumanMessage, AIMessage

def route_by_intent(state: MessagesState) -> Literal["help", "chatbot"]:
    last_message = state["messages"][-1].content.lower()
    if "帮助" in last_message:
        return "help"
    return "chatbot"

def help_node(state: MessagesState):
    return {"messages": [AIMessage(content="这里是帮助信息！")]}

def chatbot_node(state: MessagesState):
    return {"messages": [AIMessage(content="来聊聊吧！")]}

graph = StateGraph(MessagesState)
graph.add_node("help", help_node)
graph.add_node("chatbot", chatbot_node)

# 条件边：起点是 START，路由函数的返回值映射到目标节点
graph.add_conditional_edges(
    START,
    route_by_intent,
    {"help": "help", "chatbot": "chatbot"},
)
graph.add_edge("help", END)
graph.add_edge("chatbot", END)

app = graph.compile()
result = app.invoke({"messages": [HumanMessage(content="我需要帮助")]})
print(result["messages"][-1].content)  # 输出：这里是帮助信息！

import os
from typing import List, Dict, Any, Optional
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, ToolMessage

model = ChatOpenAI(
    model="qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
)

@tool
def get_weather(location: str, unit: str = "celsius") -> str:
    """获取指定位置的当前天气。"""
    return f"{location}当前天气晴朗，温度25{unit}"

@tool
def create_user(name: str, email: str, age: Optional[int] = None) -> Dict[str, Any]:
    """在系统中创建新用户。"""
    user = {"id": hash(email) % 10000, "name": name, "email": email}
    if age:
        user["age"] = age
    return user

@tool
def search_database(query: str) -> List[Dict[str, Any]]:
    """搜索数据库获取信息。"""
    if "产品A" in query:
        return [{"id": 1, "name": "产品A", "price": 99.99}]
    return []

tools = [get_weather, create_user, search_database]

# bind_tools：把工具定义绑定到 LLM，LLM 就"知道"这些工具的存在
llm_with_tools = model.bind_tools(tools)

# 工具名 -> Tool 对象，供手动执行时查找
available_tools = {t.name: t for t in tools}

# 步骤 1：初始消息列表
messages = [
    HumanMessage(content="创建一个名为李四的用户，邮箱为lisi@example.com，年龄28岁。然后告诉我北京的天气。"),
]

# 步骤 2：第一次调用 LLM（消息 + 工具定义）
first_response = llm_with_tools.invoke(messages)
messages.append(first_response)   # AIMessage，可能带 tool_calls

# 步骤 3：检查并处理 tool_calls
if first_response.tool_calls:
    for tool_call in first_response.tool_calls:
        # tool_call 是 {"name": ..., "args": ..., "id": ...}
        # 注意：args 在 LangChain 里已经是解析好的 dict
        function_name = tool_call["name"]
        function_args = tool_call["args"]

        if function_name in available_tools:
            try:
                result = available_tools[function_name].invoke(function_args)
            except Exception as e:
                result = f"Error executing function {function_name}: {e}"
            content = str(result)
        else:
            content = f"Error: Unknown function {function_name}"

        # 结果包成 ToolMessage，tool_call_id 关联原始请求
        messages.append(ToolMessage(
            tool_call_id=tool_call["id"],   # 必须提供原始调用的 ID
            content=content,
            name=function_name,
        ))

    # 步骤 4：带着工具结果二次调用 LLM
    final_response = llm_with_tools.invoke(messages)
    print(final_response.content)
else:
    # 第一次响应就没有 tool_calls，说明 LLM 直接回答了
    print(first_response.content)

pip install langchain langgraph langchain_openai langchain_core

import os
if not os.getenv("DASHSCOPE_API_KEY"):
    raise ValueError("DASHSCOPE_API_KEY environment variable not set. Please set it before running.")

from langchain_core.tools import tool

@tool
def get_weather_updates(query: str) -> str:
    """
    查询城市当前天气 (Query current weather for a city)
    Use this tool to find out the current weather for a given city.
    """
    print(f"--- Tool 'get_weather_updates' called with query: {query} ---")
    query_lower = query.lower()
    if "上海" in query_lower or 'shanghai' in query_lower:
        return "now is 30 celsius, foggy"
    elif "北京" in query_lower or 'beijing' in query_lower:
        return "now is 20 celsius, sunny"
    else:
        return f"Weather information for {query} not available with this tool. Only Shanghai and Beijing are supported."

tools = [get_weather_updates]

from langchain_openai import ChatOpenAI

model = ChatOpenAI(
    model="qwen-max",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    temperature=0
)

model_with_tools = model.bind_tools(tools)

from langchain_core.messages import HumanMessage, AIMessage, ToolMessage
from langgraph.graph import StateGraph, MessagesState
from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools)  # 预构建的工具执行节点

def call_model(state: MessagesState):
    messages = state["messages"]
    print(f"--- Calling model with {len(messages)} messages. Last message type: {type(messages[-1])} ---")
    response = model_with_tools.invoke(messages)
    return {"messages": [response]}

from typing import Literal
from langgraph.graph import START, END
from langgraph.checkpoint.memory import MemorySaver

def should_continue(state: MessagesState) -> Literal["tools", END]:
    messages = state["messages"]
    last_message = messages[-1]
    # 最后一条消息是否是带工具调用请求的 AIMessage
    if hasattr(last_message, 'tool_calls') and last_message.tool_calls and len(last_message.tool_calls) > 0:
        print(f"--- LLM decided to use tools: {last_message.tool_calls} ---")
        return "tools"   # 有工具调用，去 "tools" 节点
    print("--- LLM decided NOT to use tools. Ending. ---")
    return END            # 没有，结束流程

workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)   # Agent 节点
workflow.add_node("tools", tool_node)   # 工具节点
workflow.add_edge(START, "agent")       # 入口

# 条件边：从 "agent" 出发，由 should_continue 的返回值决定去向
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",  # 返回 "tools" -> 去 "tools" 节点
        END: END           # 返回 END -> 结束
    }
)
workflow.add_edge("tools", "agent")      # 工具执行完，回到 "agent"

checkpointer = MemorySaver()            # 内存检查点，保存对话状态
app = workflow.compile(checkpointer=checkpointer)

# 用字符串区分不同的对话线程
thread_id = "chat_thread_42"
config = {"configurable": {"thread_id": thread_id}}

print("\nInvoking for Shanghai...")
shanghai_input = {"messages": [HumanMessage(content="what's the weather in Shanghai?")]}
final_state_shanghai = app.invoke(shanghai_input, config=config)

if final_state_shanghai["messages"] and isinstance(final_state_shanghai["messages"][-1], AIMessage):
    print(f"Final response for Shanghai: {final_state_shanghai['messages'][-1].content}")

print("\nInvoking for Beijing (same thread)...")
beijing_input_message = HumanMessage(content="what's the weather in Beijing?")
# 第二次调用：同一个 thread_id，新消息自动追加到已有历史之后
final_state_beijing = app.invoke({"messages": [beijing_input_message]}, config=config)

if final_state_beijing["messages"] and isinstance(final_state_beijing["messages"][-1], AIMessage):
    print(f"Final response for Beijing: {final_state_beijing['messages'][-1].content}")

import os
from typing import TypedDict, List

from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langgraph.graph import StateGraph, START, END

llm = ChatOpenAI(
    model="qwen-plus",
    base_url="https://dashscope.aliyuncs.com/compatible-mode/v1",
    api_key=os.getenv("DASHSCOPE_API_KEY"),
    temperature=0,
)

text_splitter = RecursiveCharacterTextSplitter(chunk_size=2000, chunk_overlap=200)

# 1. 图的状态：整个流程要跟踪的全部数据
class GraphState(TypedDict):
    text: str              # 原始长文本
    chunks: List[str]      # 切分后的文本块列表
    summaries: List[str]   # 每个文本块的摘要列表
    final_summary: str     # 最终的摘要

# 2. 节点：切分
def chunk_text_node(state: GraphState):
    print("--- 正在切分文本 ---")
    chunks = text_splitter.split_text(state['text'])
    return {"chunks": chunks}

# 3. 节点：Map，对每个块独立摘要
def summarize_map_node(state: GraphState):
    print("--- 正在对每个块进行摘要 (Map) ---")
    map_prompt = ChatPromptTemplate.from_template(
        "简要总结以下文本内容：\n\n{chunk}"
    )
    summarize_chain = map_prompt | llm

    # batch 并行处理所有块，max_concurrency 控制并发
    summaries = summarize_chain.batch(
        [{"chunk": chunk} for chunk in state['chunks']],
        config={"max_concurrency": 5}
    )
    # batch 返回 AIMessage 列表，提取 .content
    cleaned_summaries = [s.content for s in summaries]
    return {"summaries": cleaned_summaries}

# 4. 节点：Reduce，合并所有小摘要
def summarize_reduce_node(state: GraphState):
    print("--- 正在合并所有摘要 (Reduce) ---")
    summaries_joined = "\n\n".join(state['summaries'])

    reduce_prompt = ChatPromptTemplate.from_template(
        "你收到了关于一个长文档的多个摘要。请将它们整合成一个连贯、流畅、全面的最终摘要。\n\n"
        "以下是各个部分的摘要：\n{summaries_text}"
    )
    reduce_chain = reduce_prompt | llm
    final_summary = reduce_chain.invoke({"summaries_text": summaries_joined})
    return {"final_summary": final_summary.content}

# 5. 连图
workflow = StateGraph(GraphState)
workflow.add_node("chunker", chunk_text_node)
workflow.add_node("mapper", summarize_map_node)
workflow.add_node("reducer", summarize_reduce_node)

workflow.add_edge(START, "chunker")   # 入口
workflow.add_edge("chunker", "mapper")
workflow.add_edge("mapper", "reducer")
workflow.add_edge("reducer", END)
app = workflow.compile()

# 6. 跑一个示例：把一段文本重复 10 次模拟长文本
long_text = """
人工智能（AI）正在以前所未有的速度改变世界。从自动驾驶汽车到医疗诊断，AI的应用无处不在。
其核心技术包括机器学习、深度学习和自然语言处理。机器学习使计算机能够从数据中学习规律，而无需进行显式编程。
深度学习是机器学习的一个分支，它利用深度神经网络模型，在图像识别、语音识别等领域取得了巨大成功。
自然语言处理则致力于让计算机能够理解和生成人类语言，Siri和ChatGPT就是最好的例子。
然而，AI的发展也带来了挑战，如数据隐私、算法偏见和就业冲击。解决这些问题需要技术、法律和伦理的共同努力。
""" * 10

final_state = app.invoke({"text": long_text})
print(final_state['final_summary'])

from langgraph.checkpoint.memory import MemorySaver

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

config = {"configurable": {"thread_id": "1"}}
result = app.invoke({"messages": [HumanMessage(content="你好！")]}, config)

# 后续调用恢复状态：只传新消息，历史自动接上
result = app.invoke({"messages": [HumanMessage(content="继续聊")]}, config)

from datetime import timedelta
from langgraph.types import TimeoutPolicy

# 简单封顶：秒数或 timedelta
builder.add_node("call_model", call_model, timeout=60)
builder.add_node("call_model", call_model, timeout=timedelta(minutes=2))

# 分别限制总时长和空闲时长
builder.add_node(
    "call_model",
    call_model,
    timeout=TimeoutPolicy(run_timeout=120, idle_timeout=30),
)

from langgraph.errors import NodeError
from langgraph.types import Command, RetryPolicy

def charge_payment(state: State) -> State:
    raise RuntimeError("payment gateway timeout")

def payment_error_handler(state: State, error: NodeError) -> Command:
    return Command(
        update={"status": f"compensated: {error.error}"},
        goto="finalize",
    )

graph = (
    StateGraph(State)
    .add_node(
        "charge_payment",
        charge_payment,
        retry_policy=RetryPolicy(max_attempts=3, retry_on=ConnectionError),
        error_handler=payment_error_handler,
    )
    .add_node("finalize", finalize)
    .add_edge(START, "charge_payment")
    .compile()
)

from langgraph.types import RetryPolicy, TimeoutPolicy

builder = (
    StateGraph(State)
    .set_node_defaults(
        error_handler=default_error_handler,
        timeout=TimeoutPolicy(run_timeout=30),
    )
    .add_node("step_a", step_a)                                       # 用默认处理器
    .add_node("step_b", step_b, error_handler=custom_error_handler)   # 用自己的
)

import signal
from langgraph.runtime import RunControl
from langgraph.errors import GraphDrained

control = RunControl()
signal.signal(signal.SIGTERM, lambda *_: control.request_drain("sigterm"))

try:
    result = graph.invoke(inputs, config, control=control)
except GraphDrained as e:
    log.info("graph drained: %s", e.reason)
    # 下次启动时用同一个 config 继续

from langgraph.runtime import Runtime

async def my_node(state: State, runtime: Runtime) -> State:
    if runtime.drain_requested:
        return {"status": "skipped", "reason": runtime.drain_reason}
    return {"status": await do_work()}

stream = graph.stream_events(input_data, version="v3")

for message in stream.messages:      # token 级模型输出
    print(str(message.text), end="", flush=True)

final_state = stream.output          # 最终状态

from langgraph.types import GraphOutput

result: GraphOutput = graph.invoke(input_data, config, version="v2")
print(result.value)       # 状态值
print(result.interrupts)  # 中断信息

from langchain_community.llms import Tongyi
from langchain_community.graphs import NetworkXEntityGraph
from langchain_experimental.graph_transformers.llm import LLMGraphTransformer
from langchain_core.documents import Document

llm_for_graph = Tongyi(model_name="qwen-plus")

document_content = """
张三是北京大学的教授，他研究人工智能领域。
李四是张三的学生，他就读于清华大学，学习计算机科学。
王五是百度的工程师，百度是一家位于北京的科技公司。
"""
docs = [Document(page_content=document_content)]

llm_transformer = LLMGraphTransformer(llm=llm_for_graph)
graph_documents = llm_transformer.convert_to_graph_documents(docs)

final_graph = NetworkXEntityGraph()
final_graph.add_graph_documents(graph_documents)
for triple in final_graph.get_triples():
    print(triple)

def get_relevant_triples(graph, entity_name):
    """检索与某个实体相关的三元组"""
    lines = []
    for s, p, o in graph.get_triples():
        if entity_name in str(s) or entity_name in str(o):
            lines.append(f"({s}, {p}, {o})")
    return "\n".join(lines)

context = get_relevant_triples(final_graph, "张三")
# 再把 context 和问题拼进 prompt，交给 LLM 生成答案