LangGraph 入门#

1
# pip install langgraph
2
from typing import TypedDict, Annotated
3
from langgraph.graph.message import add_messages
4

5
class AgentState(TypedDict):
6
    """Agent 的全局状态"""
7
    messages: Annotated[list, add_messages]  # 对话消息列表（自动追加）
8
    next_action: str                          # 下一步要执行的动作

1
# pip install langgraph langchain-openai
2
from langchain_openai import ChatOpenAI
3

4
llm = ChatOpenAI(model="gpt-4o", temperature=0)
5

6
def chatbot_node(state: AgentState) -> dict:
7
    """聊天机器人节点：调用 LLM 生成回复"""
8
    response = llm.invoke(state["messages"])
9
    return {"messages": [response]}  # 返回部分更新，Reducer 会自动追加

1
# pip install langgraph
2
from langgraph.graph import StateGraph, START, END
3

4
graph = StateGraph(AgentState)
5
graph.add_node("chatbot", chatbot_node)
6

7
# 固定边
8
graph.add_edge(START, "chatbot")
9
graph.add_edge("chatbot", END)
10

11
# 条件边示例
12
def should_continue(state: AgentState) -> str:
13
    last_message = state["messages"][-1]
14
    if last_message.tool_calls:
15
        return "tools"
16
    return "end"
17

18
graph.add_conditional_edges(
19
    "chatbot",              # 从哪个节点出发
20
    should_continue,        # 路由函数
21
    {"tools": "tool_node", "end": END},  # 路由映射表
22
)

1
# pip install langgraph langchain-openai langchain-core
2
from typing import TypedDict, Annotated
3
from langgraph.graph import StateGraph, START, END
4
from langgraph.graph.message import add_messages
5
from langchain_openai import ChatOpenAI
6
from langchain_core.messages import HumanMessage
7

8
# ---- 1. 定义 State ----
9
class ChatState(TypedDict):
10
    messages: Annotated[list, add_messages]
11

12
# ---- 2. 定义节点函数 ----
13
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
14

15
def chatbot(state: ChatState) -> dict:
16
    response = llm.invoke(state["messages"])
17
    return {"messages": [response]}
18

19
# ---- 3. 构建图 ----
20
graph_builder = StateGraph(ChatState)
21
graph_builder.add_node("chatbot", chatbot)
22
graph_builder.add_edge(START, "chatbot")
23
graph_builder.add_edge("chatbot", END)
24

25
# ---- 4. 编译图 ----
26
graph = graph_builder.compile()
27

28
# ---- 5. 运行 ----
29
result = graph.invoke({
30
    "messages": [HumanMessage(content="你好！请用一句话介绍 LangGraph。")]
31
})
32
print(result["messages"][-1].content)

1
# pip install langgraph langchain-openai langchain-core
2
from langchain_core.messages import HumanMessage
3

4
for event in graph.stream(
5
    {"messages": [HumanMessage(content="请解释什么是状态机。")]},
6
    stream_mode="updates",
7
):
8
    for node_name, state_update in event.items():
9
        print(f"--- 节点: {node_name} ---")
10
        if "messages" in state_update:
11
            print(state_update["messages"][-1].content)

1
# pip install langgraph langchain-openai langchain-core
2
from typing import TypedDict, Annotated
3
from langgraph.graph import StateGraph, START, END
4
from langgraph.graph.message import add_messages
5
from langgraph.prebuilt import ToolNode
6
from langchain_openai import ChatOpenAI
7
from langchain_core.tools import tool
8
from langchain_core.messages import HumanMessage
9

10
# ---- 1. 定义工具 ----
11
@tool
12
def search_weather(city: str) -> str:
13
    """查询指定城市的当前天气信息。"""
14
    weather_db = {
15
        "北京": "晴天，25°C，湿度 30%",
16
        "上海": "多云，22°C，湿度 65%",
17
        "深圳": "雷阵雨，28°C，湿度 85%",
18
    }
19
    return weather_db.get(city, f"抱歉，未找到 {city} 的天气数据。")
20

21
@tool
22
def calculate(expression: str) -> str:
23
    """计算数学表达式，返回结果。"""
24
    try:
25
        return f"计算结果: {eval(expression)}"
26
    except Exception as e:
27
        return f"计算错误: {e}"
28

29
tools = [search_weather, calculate]
30

31
# ---- 2. 定义 State ----
32
class AgentState(TypedDict):
33
    messages: Annotated[list, add_messages]
34

35
# ---- 3. 定义节点 ----
36
llm = ChatOpenAI(model="gpt-4o", temperature=0)
37
llm_with_tools = llm.bind_tools(tools)
38

39
def llm_node(state: AgentState) -> dict:
40
    """LLM 节点：生成回复或工具调用请求"""
41
    response = llm_with_tools.invoke(state["messages"])
42
    return {"messages": [response]}
43

44
tool_node = ToolNode(tools)
45

46
# ---- 4. 定义路由函数 ----
47
def should_continue(state: AgentState) -> str:
48
    last_message = state["messages"][-1]
49
    if last_message.tool_calls:
50
        return "tools"
51
    return "end"
52

53
# ---- 5. 构建图 ----
54
graph_builder = StateGraph(AgentState)
55
graph_builder.add_node("llm", llm_node)
56
graph_builder.add_node("tools", tool_node)
57

58
graph_builder.add_edge(START, "llm")
59
graph_builder.add_conditional_edges(
60
    "llm", should_continue, {"tools": "tools", "end": END},
61
)
62
graph_builder.add_edge("tools", "llm")  # 工具 → 回到 LLM（形成循环）
63

64
# ---- 6. 编译并运行 ----
65
graph = graph_builder.compile()
66

67
result = graph.invoke({
68
    "messages": [HumanMessage(
69
        content="北京今天天气怎么样？另外帮我算一下 1024 * 768 等于多少。"
70
    )]
71
})
72
print(result["messages"][-1].content)

1
# pip install langgraph langchain-openai langchain-core
2
from typing import TypedDict, Annotated
3
from langgraph.graph import StateGraph, START, END
4
from langgraph.graph.message import add_messages
5
from langgraph.prebuilt import ToolNode
6
from langgraph.checkpoint.memory import MemorySaver
7
from langchain_openai import ChatOpenAI
8
from langchain_core.tools import tool
9
from langchain_core.messages import HumanMessage
10

11
@tool
12
def send_email(to: str, subject: str, body: str) -> str:
13
    """发送电子邮件。敏感操作，需要人工确认。"""
14
    return f"邮件已成功发送至 {to}，主题：{subject}"
15

16
tools = [send_email]
17

18
class State(TypedDict):
19
    messages: Annotated[list, add_messages]
20

21
llm = ChatOpenAI(model="gpt-4o", temperature=0)
22
llm_with_tools = llm.bind_tools(tools)
23

24
def assistant(state: State) -> dict:
25
    return {"messages": [llm_with_tools.invoke(state["messages"])]}
26

27
def route(state: State) -> str:
28
    if state["messages"][-1].tool_calls:
29
        return "tools"
30
    return "end"
31

32
# 构建图（注意 interrupt_before）
33
graph_builder = StateGraph(State)
34
graph_builder.add_node("assistant", assistant)
35
graph_builder.add_node("tools", ToolNode(tools))
36
graph_builder.add_edge(START, "assistant")
37
graph_builder.add_conditional_edges("assistant", route, {"tools": "tools", "end": END})
38
graph_builder.add_edge("tools", "assistant")
39

40
memory = MemorySaver()
41
graph = graph_builder.compile(
42
    checkpointer=memory,
43
    interrupt_before=["tools"],  # 在执行工具节点之前暂停！
44
)
45

46
# ---- 第一次运行：Agent 在工具节点前暂停 ----
47
config = {"configurable": {"thread_id": "email-thread-001"}}
48
result = graph.invoke(
49
    {"messages": [HumanMessage(content="帮我发邮件给 alice@example.com，主题是会议通知")]},
50
    config=config,
51
)
52

53
print("=== Agent 想要执行以下操作 ===")
54
for tc in result["messages"][-1].tool_calls:
55
    print(f"工具: {tc['name']}, 参数: {tc['args']}")
56

57
# ---- 人工确认后继续执行 ----
58
final_result = graph.invoke(None, config=config)  # None = 从暂停点继续
59
print(final_result["messages"][-1].content)

1
# pip install langgraph langchain-openai langchain-core aiosqlite
2
from typing import TypedDict, Annotated
3
from langgraph.graph import StateGraph, START, END
4
from langgraph.graph.message import add_messages
5
from langgraph.checkpoint.sqlite import SqliteSaver
6
from langchain_openai import ChatOpenAI
7
from langchain_core.messages import HumanMessage
8

9
class ChatState(TypedDict):
10
    messages: Annotated[list, add_messages]
11

12
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)
13

14
def chatbot(state: ChatState) -> dict:
15
    return {"messages": [llm.invoke(state["messages"])]}
16

17
graph_builder = StateGraph(ChatState)
18
graph_builder.add_node("chatbot", chatbot)
19
graph_builder.add_edge(START, "chatbot")
20
graph_builder.add_edge("chatbot", END)
21

22
with SqliteSaver.from_conn_string("checkpoint.db") as checkpointer:
23
    graph = graph_builder.compile(checkpointer=checkpointer)
24
    config = {"configurable": {"thread_id": "user-alice-001"}}
25

26
    # 第一轮对话
27
    result1 = graph.invoke(
28
        {"messages": [HumanMessage(content="我叫 Alice，我是一名数据工程师。")]},
29
        config=config,
30
    )
31
    print("第一轮:", result1["messages"][-1].content)
32

33
    # 第二轮对话（同一个 thread_id，自动携带历史）
34
    result2 = graph.invoke(
35
        {"messages": [HumanMessage(content="你还记得我的名字和职业吗？")]},
36
        config=config,
37
    )
38
    print("第二轮:", result2["messages"][-1].content)
39
    # 模型会回答："你叫 Alice，你是一名数据工程师。"
40

41
    # 不同的 thread_id，互不影响
42
    config_bob = {"configurable": {"thread_id": "user-bob-002"}}
43
    result3 = graph.invoke(
44
        {"messages": [HumanMessage(content="你知道我叫什么吗？")]},
45
        config=config_bob,
46
    )
47
    print("Bob 的对话:", result3["messages"][-1].content)
48
    # 模型不知道 Bob 的信息，因为这是一个全新的 thread

1
# pip install langgraph
2
for state_snapshot in graph.get_state_history(config):
3
    print(f"时间: {state_snapshot.created_at}")
4
    print(f"消息数: {len(state_snapshot.values.get('messages', []))}")
5
    print(f"下一步: {state_snapshot.next}")
6
    print("---")

1
# pip install langchain langchain-openai langchain-core
2
from langchain_openai import ChatOpenAI
3
from langchain_core.tools import tool
4
from langchain.agents import create_tool_calling_agent, AgentExecutor
5
from langchain_core.prompts import ChatPromptTemplate
6

7
@tool
8
def get_weather(city: str) -> str:
9
    """查询天气"""
10
    return f"{city}: 晴天，25°C"
11

12
llm = ChatOpenAI(model="gpt-4o", temperature=0)
13
prompt = ChatPromptTemplate.from_messages([
14
    ("system", "你是一个天气助手。"),
15
    ("human", "{input}"),
16
    ("placeholder", "{agent_scratchpad}"),
17
])
18

19
agent = create_tool_calling_agent(llm, [get_weather], prompt)
20
executor = AgentExecutor(agent=agent, tools=[get_weather], verbose=True)
21
result = executor.invoke({"input": "北京天气如何？"})
22
print(result["output"])

1
# pip install langgraph langchain-openai langchain-core
2
from typing import TypedDict, Annotated
3
from langgraph.graph import StateGraph, START, END
4
from langgraph.graph.message import add_messages
5
from langgraph.prebuilt import ToolNode
6
from langgraph.checkpoint.memory import MemorySaver
7
from langchain_openai import ChatOpenAI
8
from langchain_core.tools import tool
9
from langchain_core.messages import HumanMessage, SystemMessage
10

11
@tool
12
def get_weather(city: str) -> str:
13
    """查询天气"""
14
    return f"{city}: 晴天，25°C"
15

16
class State(TypedDict):
17
    messages: Annotated[list, add_messages]
18

19
llm = ChatOpenAI(model="gpt-4o", temperature=0)
20
llm_with_tools = llm.bind_tools([get_weather])
21

22
def assistant(state: State) -> dict:
23
    messages = [SystemMessage(content="你是一个天气助手。")] + state["messages"]
24
    return {"messages": [llm_with_tools.invoke(messages)]}
25

26
def route(state: State) -> str:
27
    if state["messages"][-1].tool_calls:
28
        return "tools"
29
    return "end"
30

31
builder = StateGraph(State)
32
builder.add_node("assistant", assistant)
33
builder.add_node("tools", ToolNode([get_weather]))
34
builder.add_edge(START, "assistant")
35
builder.add_conditional_edges("assistant", route, {"tools": "tools", "end": END})
36
builder.add_edge("tools", "assistant")
37

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

40
config = {"configurable": {"thread_id": "weather-001"}}
41
result = graph.invoke(
42
    {"messages": [HumanMessage(content="北京天气如何？")]}, config=config,
43
)
44
print(result["messages"][-1].content)

1
你的 Agent 需要...
2
├── 只是简单的工具调用？ → AgentExecutor 足够
3
├── 条件分支？ → 迁移到 LangGraph
4
├── 人工审批 / 确认？ → 迁移到 LangGraph
5
├── 断点恢复 / 状态持久化？ → 迁移到 LangGraph
6
├── 多 Agent 协作？ → 迁移到 LangGraph
7
└── 生产部署？ → 迁移到 LangGraph

1
# pip install langgraph langchain-openai langchain-core
2
from langgraph.prebuilt import create_react_agent
3
from langgraph.checkpoint.memory import MemorySaver
4
from langchain_openai import ChatOpenAI
5
from langchain_core.tools import tool
6
from langchain_core.messages import HumanMessage
7

8
@tool
9
def get_weather(city: str) -> str:
10
    """查询天气"""
11
    return f"{city}: 晴天，25°C"
12

13
# 一行代码创建 Agent
14
agent = create_react_agent(
15
    ChatOpenAI(model="gpt-4o", temperature=0),
16
    tools=[get_weather],
17
    checkpointer=MemorySaver(),
18
)
19

20
config = {"configurable": {"thread_id": "react-001"}}
21
result = agent.invoke(
22
    {"messages": [HumanMessage(content="上海天气如何？")]}, config=config,
23
)
24
print(result["messages"][-1].content)

1
LangGraph 核心公式：
2

3
StateGraph(State)           # 创建图
4
  .add_node(name, func)     # 添加节点
5
  .add_edge(A, B)           # 添加固定边
6
  .add_conditional_edges(   # 添加条件边
7
      source,               #   从哪个节点出发
8
      router_func,          #   路由函数
9
      path_map              #   路由映射表
10
  )
11
  .compile(                 # 编译
12
      checkpointer=...,     #   持久化后端
13
      interrupt_before=..., #   暂停点
14
  )
15
  .invoke(input, config)    # 运行