StateGraph(State)
    ├── .add_node(name, func)     # 添加节点（Agent 步骤）
    ├── .add_edge(source, target)  # 添加边（固定流转）
    ├── .add_conditional_edges()   # 添加条件边（动态路由）
    └── .compile()                # 编译成可执行的图

from typing import TypedDict
from langgraph.graph import StateSchema

class AgentState(TypedDict):
    messages: list[str]           # 对话历史
    current_step: str             # 当前步骤
    context: dict                 # 额外上下文
    checkpoint: str               # 持久化检查点

from langgraph.graph import StateGraph

def should_continue(state: AgentState) -> str:
    """路由节点：根据状态决定下一步"""
    if len(state["messages"]) > 10:
        return "end"
    return "continue"

# 定义节点
graph = StateGraph(AgentState)
graph.add_node("chat", chat_node)           # 聊天节点
graph.add_node("search", search_node)        # 搜索节点
graph.add_node("router", should_continue)    # 路由节点

# 固定边：必然发生的流转
graph.add_edge("chat", "search")             # chat 完成后必然到 search

# 条件边：根据状态动态决定
graph.add_conditional_edges(
    "router",
    lambda state: "chat" if state["needs_human"] else "search"
)

from langgraph.checkpoint.memory import MemorySaver

# 内存 Checkpointer（开发用）
checkpointer = MemorySaver()

# 持久化 Checkpointer（生产用）
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver.from_conn_string("postgresql://...")

# 编译时绑定 Checkpointer
app = graph.compile(checkpointer=checkpointer)

# 从检查点恢复并继续执行
config = {"configurable": {"thread_id": "user_123"}}
app.invoke(input, config=config)

from langgraph.graph import StateGraph
from langgraph.checkpoint.postgres import PostgresSaver

# 1. 定义 Agent
class AgentState(TypedDict):
    messages: list
    step: int

def chat_node(state):
    response = llm.invoke(state["messages"])
    return {"messages": [response], "step": state["step"] + 1}

# 2. 构建图
graph = StateGraph(AgentState)
graph.add_node("chat", chat_node)
graph.set_entry_point("chat")
graph.add_edge("chat", END)

# 3. 配置持久化
checkpointer = PostgresSaver.from_conn_string("postgresql://user:pass@host/db")
app = graph.compile(checkpointer=checkpointer)

# 4. 持久的用户会话
config = {"configurable": {"thread_id": "user_123"}}
for message in conversation:
    app.invoke({"messages": [message]}, config=config)

# 场景：执行到一半网络断开
try:
    result = app.invoke(input, config=config)
except APITimeoutError:
    pass  # 记录异常但不做处理

# 后续：用户重新连接
config = {"configurable": {"thread_id": "user_123"}}
# LangGraph 自动从上一个检查点恢复
result = app.invoke(None, config=config)  # input=None 表示继续

from langgraph.errors import NodeInterrupt

def sensitive_node(state):
    """发送邮件节点"""
    email_content = draft_email(state)

    # 中断执行，等待人工确认
    raise NodeInterrupt(
        f"即将发送邮件给 {state['recipient']}，内容：\n{email_content}"
    )

# 在 Agent 执行中
config = {"configurable": {"thread_id": "user_123"}}

# 执行到 sensitive_node 时会抛出 NodeInterrupt
try:
    app.invoke(input, config=config)
except NodeInterrupt as e:
    print(f"需要人工确认：{e.message}")

    # 人工审核后，可以：
    # 1. 批准继续（修改状态后继续）
    # 2. 修改内容后继续
    # 3. 中止执行

# 人工审核时，可以修改 Agent 状态
from langgraph.constants import Edit

# 假设审核时发现邮件内容有误，修正后继续
app.update_state(
    config,
    {"email_content": "修正后的邮件内容"}
)

# 批准后继续执行
result = app.invoke(None, config=config)

from langgraph.graph import add_messages

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

# messages 自动去重和追加
def chat_node(state: AgentState):
    new_messages = [HumanMessage(content="Hello")]
    # add_messages 策略自动处理
    return {"messages": new_messages}

from langchain.memory import VectorStoreRetrieverMemory

# 连接向量数据库作为长期记忆
memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5})
)

def recall_memory(state: AgentState) -> dict:
    """从长期记忆检索相关内容"""
    query = state["messages"][-1].content
    relevant = memory.load_memory_variables({"input": query})
    return {"context": relevant["history"]}

# 在图中添加记忆节点
graph.add_node("recall", recall_memory)
graph.add_edge("recall", "chat")

pip install -U langgraph

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

# 1. 定义状态
class AgentState(TypedDict):
    input: str
    agent_outcome: str
    intermediate_steps: list

# 2. 定义节点
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4")
tools = [search, calculate]  # 你的工具

def agent(state):
    messages = [{"role": "user", "content": state["input"]}]
    response = llm.bind_tools(tools).invoke(messages)
    return {"agent_outcome": response}

def should_continue(state):
    if hasattr(state["agent_outcome"], "tool_calls"):
        return "continue"
    return "end"

# 3. 构建图
graph = StateGraph(AgentState)
graph.add_node("agent", agent)
graph.add_node("action", tool_node)  # 工具执行节点
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue)
graph.add_edge("action", "agent")  # 工具执行完返回 agent
graph.add_edge("agent", END)

# 4. 编译并运行
app = graph.compile()
result = app.invoke({"input": "北京今天天气怎么样？"})

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"

# 启用后，所有执行都会被追踪
app.invoke(input, config=config)  # LangSmith 自动记录完整轨迹

# 部署到 LangGraph Cloud
langgraph deploy --name my-agent

# 本地开发服务器
langgraph dev