langgraph agents langchain state orchestration python

LangGraph构建模块：状态、节点、边以及调用功能

构建之前的LangGraph词汇表：输入状态、小型节点、条件边、工具调用以及以状态形式存在的消息，通过一个简单的智能体进行演示。

发布者：REACTAPP.TOP加入本站 2026年9月26日阅读约 9 分钟2150 词

来源: The Building Blocks of LangGraph (Part 0)

随着模型具备了工具使用能力和多步骤推理能力，临时循环变得难以测试和恢复。LangGraph将控制流明确化：状态、节点、边以及调用关系——这样智能体就变成了可以查看的图结构，而非杂乱无章的提示词组合。

最简单的图结构

最基础的智能体流程为：开始 → 推理/执行节点 → 结束，各步骤之间通过共享状态进行传递。

构建第一个简单智能体

Receive Order

Prepare Food

Deliver Food

Receive Order
      ↓
Prepare Food
      ↓
Deliver Food

Pizza
Address: 123 Main Street
Customer: John

Receive Order
      ↓
Prepare Food
      ↓
Deliver Food

from typing_extensions import TypedDict

class State(TypedDict):
    graph_state: strNodes

def node_1(state):
  print("---Node 1---")
  return {"graph_state": state['graph_state'] + "I am"}

def node_2(state):
  print("---Node 2---")
  return {"graph_state": state['graph_state'] + "happy!"}

def node_3(state):
  print("---Node 3---")
  return {"graph_state": state['graph_state'] + "sad!"}

import random
from typing import Literal

def decide_mood(state) -> Literal["node_2", "node_3"]:

  # Often, we will use state to decide on the next node to visit
  user_input = state['graph_state']


  # Here, let's just do a 50 / 50 split between nodes 2, 3
  if random.random() < 0.5

      # 50% of the time, we return Node 2
      return "node_2"

  # 50% of the time, we return Node 3
  return "node_3"

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

#Build Graph
builder = StateGraph(state)
builder.add_node("node_1", node_1)
builder.add_node("node_2", node_2)
builder.add_node("node_3", node_3)

#Logic
builder.add_edge(START, "node_1")
builder.add_conditional_edges("node_1", decide_mood)
builder.add_edge("node_2", END)
builder.add_edge("node_3", END)

#Add
graph = builder.compile()

#View
display(Image(graph.get_graph().draw_mermaid_png()))

graph.invoke({"graph_state" : "Hi, this is Lance."})

###OUTPUT
---Node 1---
---Node 3---
{'graph_state': 'Hi, this is Lance. I am sad!'}

{'graph_state': 'Hi, this is Lance. I am sad!'}

%%capture --no-stderr
%pip install --quiet -U langchain_openai langchain_core langgraph

from pprint import pprint
from langchain_core.messages import AIMessage, HumanMessage

messages = [AIMessage(content=f"So you said you were researching ocean mammals?", name="Model")]
messages.append(HumanMessage(content=f"Yes, that's right.",name="Lance"))
messages.append(AIMessage(content=f"Great, what would you like to learn about.", name="Model"))
messages.append(HumanMessage(content=f"I want to learn about the best place to see Orcas in the US.", name="Lance"))

for m in messages:
    m.pretty_print()

#OUTPUT
================================== Ai Message ==================================
Name: Model

So you said you were researching ocean mammals?
================================ Human Message =================================
Name: Lance

Yes, that's right.
================================== Ai Message ==================================
Name: Model

Great, what would you like to learn about.
================================ Human Message =================================
Name: Lance

I want to learn about the best place to see Orcas in the US.

from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o")
result = llm.invoke(messages)
type(result)
result

#OUTPUT

langchain_core.messages.ai.AIMessage
result
AIMessage(content='One of the best places to see orcas in the United States is the Pacific Northwest, particularly around the San Juan Islands in Washington State. Here are some details:\n\n1. **San Juan Islands, Washington**: These islands are a renowned spot for whale watching, with orcas frequently spotted between late spring and early fall. The waters around the San Juan Islands are home to both resident and transient orca pods, making it an excellent location for sightings.\n\n2. **Puget Sound, Washington**: This area, including places like Seattle and the surrounding waters, offers additional opportunities to see orcas, particularly the Southern Resident killer whale population.\n\n3. **Olympic National Park, Washington**: The coastal areas of the park provide a stunning backdrop for spotting orcas, especially during their migration periods.\n\nWhen planning a trip for whale watching, consider peak seasons for orca activity and book tours with reputable operators who adhere to responsible wildlife viewing practices. Additionally, land-based spots like Lime Kiln Point State Park, also known as “Whale Watch Park,” on San Juan Island, offer great opportunities for orca watching from shore.', additional_kwargs={'refusal': None}, response_metadata={'token_usage': {'completion_tokens': 228, 'prompt_tokens': 67, 'total_tokens': 295, 'completion_tokens_details': {'accepted_prediction_tokens': 0, 'audio_tokens': 0, 'reasoning_tokens': 0, 'rejected_prediction_tokens': 0}, 'prompt_tokens_details': {'audio_tokens': 0, 'cached_tokens': 0}}, 'model_name': 'gpt-4o-2024-08-06', 'system_fingerprint': 'fp_50cad350e4', 'finish_reason': 'stop', 'logprobs': None}, id='run-57ed2891-c426-4452-b44b-15d0a5c3f225-0', usage_metadata={'input_tokens': 67, 'output_tokens': 228, 'total_tokens': 295, 'input_token_details': {'audio': 0, 'cache_read': 0}, 'output_token_details': {'audio': 0, 'reasoning': 0}})

def multiply(a:int, b:int)-> int:
  return a*b

llm_with_tools = llm.bind_tools([multiply])

tool_call = llm_with_tools.invoke([HumanMessage(content=f"What is 2 multiplied by 3
", name="Lance")])

tool_call.tool_calls

#OUTPUT
[{'name': 'multiply',
  'args': {'a': 2, 'b': 3},
  'id': 'call_lBBBNo5oYpHGRqwxNaNRbsiT',
  'type': 'tool_call'}]

from typing_extentions import TypedDict
from langchain_core.messges import AnyMessage

class MessagesState(TypedDict):
  messages : list[AnyMessage]

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

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

from langgraph.graph import MessagesState

class MessagesState(MessagesState):
    # Add any keys needed beyond messages, which is pre-built
    pass

# Initial state
initial_messages = [AIMessage(content="Hello! How can I assist you?", name="Model"),
                    HumanMessage(content="I'm looking for information on marine biology.", name="Lance")
                   ]

# New message to add
new_message = AIMessage(content="Sure, I can help with that. What specifically are you interested in?", name="Model")

# Test
add_messages(initial_messages , new_message)

from IPython.display import Image, display
from langgraph.graph import StateGraph, START, END

# Node
def tool_calling_llm(state: MessagesState):
    return {"messages": [llm_with_tools.invoke(state["messages"])]}

# Build graph
builder = StateGraph(MessagesState)
builder.add_node("tool_calling_llm", tool_calling_llm)
builder.add_edge(START, "tool_calling_llm")
builder.add_edge("tool_calling_llm", END)
graph = builder.compile()

# View
display(Image(graph.get_graph().draw_mermaid_png()))

messages = graph.invoke({"messages": HumanMessage(content="Hello!")})
for m in messages['messages']:
    m.pretty_print()

#OUTPUT
================================ Human Message =================================

Hello!
================================== Ai Message ==================================

Hi there! How can I assist you today?

messages = graph.invoke({"messages": HumanMessage(content="Multiply 2 and 3")})
for m in messages['messages']:
    m.pretty_print()

OUTPUT

================================ Human Message =================================

Multiply 2 and 3!
================================== Ai Message ==================================
Tool Calls:
  multiply (call_Er4gChFoSGzU7lsuaGzfSGTQ)
 Call ID: call_Er4gChFoSGzU7lsuaGzfSGTQ
  Args:
    a: 2
    b: 3

状态

结构化状态（通常为TypedDict或reducer函数）用于存储消息、标志及中间结果。选择reducer时需谨慎——处理消息列表用append，处理标量值用replace。

节点

节点即函数：读取状态，执行操作（如LLM调用、工具使用或纯Python代码），并返回部分更新结果。应保持节点简洁且易于测试。

边

静态边用于确定顺序；条件边则根据状态进行分支。当出现分支和循环时，图结构才能真正发挥作用。

图构建

注册节点，连接边，并可选地使用检查点机制以确保数据持久性。

图调用

通过输入状态和配置（检查点时需提供线程ID）来调用图结构，同时输出事件以提升用户体验。

链与图

线性链仍适合单次处理流程，而图结构则支持暂停/继续、分支操作以及共享的命名状态。

消息与聊天模型

消息列表是常见的对话状态表示方式，应将聊天模型绑定在节点内部，而非将I/O操作隐藏在边逻辑中。

工具

为模型调用节点提供工具接口，解析工具调用请求，并将观察结果写回状态中。

将消息用作状态

当消息本身就是状态时， reducer必须以安全的方式追加数据，此时截断/摘要处理会成为一种有意的节点设计——而非偶然现象。

总结

第一部分介绍了基本概念：状态、节点、边、编译与调用。后续部分将在相同基础上进一步探讨持久线程、HITL中断以及多智能体模式。

应为节点命名业务步骤的名称（如classify、retrieve、draft），而非使用node1这类名称。易于理解的图结构才是可操作的图结构。

应添加单元测试，使用固定状态调用已编译的图结构，并验证哪些节点被执行。仅通过compile()函数无法证明业务逻辑的完整性。

应为节点命名业务步骤的名称（如classify、retrieve、draft），而非使用node1这类名称。易于理解的图结构才是可操作的图结构。