langgraph agents langchain state orchestration python

Компоненты LangGraph: состояние, узлы, рёбра и вызовы

Словарь LangGraph перед его созданием: введённое состояние, маленькие узлы, условные рёбра, вызов инструментов и сообщения в качестве состояния, показано на примере простого агента.

Издатель: REACTAPP.TOPДобавлено на сайт 26 сент. 2026 г.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 или редюсеров) хранит сообщения, флаги и промежуточные результаты. Выбирайте редюсеры осознанно: используйте функцию append для списков сообщений и replace для скалярных значений.

Узлы

Узлы — это функции: они считывают состояние, выполняют работу (с использованием LLM, инструментов или чистого Python) и возвращают частичное обновление состояния. Старайтесь делать их простыми и поддающимися тестированию.

Ребра

Статические рёбра фиксируют порядок; условные рёбра создают ветвления в зависимости от состояния. Графы становятся полезными, когда появляются ветвления и циклы.

Построение графа

Регистрируются узлы, прокладываются рёбра, выполняется компиляция с необязательным контрольным указателем для обеспечения надёжности.

Запуск графа

Запускается с начальным состоянием и настройками (идентификатор потока при сохранении состояния). Для улучшения пользовательского интерфейса передаются события в реальном времени.

Цепочки против графов

Линейные цепочки позволяют выполнять операции за один проход. Графы поддерживают возможность паузы/продолжения, ветвления и общее именованное состояние.

Сообщения и модели чата

Списки сообщений представляют собой стандартное состояние диалога. Модели чата привязываются к узлам, а не скрываются в логике рёбер для обработки ввода-вывода.

Инструменты

Инструменты предоставляются узлам, вызывающим модели; выполняется парсинг запросов к инструментам; наблюдения записываются обратно в состояние.

Использование сообщений в качестве состояния

Когда сообщения являются состоянием, редьюсеры должны безопасно добавлять новые данные, причем обрезка/краткое изложение становится явным элементом структуры — а не случайностью.

Заключение

В первой части рассматривается словарный запас: состояние, узлы, рёбра, процессы компиляции и вызова. В последующих частях на этой же основе добавляются информация о долговечных потоках, прерываниях типа HITL и паттернах с участием нескольких агентов.

Называйте узлы в соответствии с бизнес-шагами (classify, retrieve, draft), а не node1. Читаемые графы — это графы, с которыми можно работать.

Добавьте тест на единицу, в котором будет вызываться скомпилированный граф с заданном начальным состоянием, и проверяться, какие узлы были обработаны. Одного только метода compile() недостаточно для подтверждения полноты реализации бизнес-логики.

Добавьте тест на единицу, который запускает скомпилированную структуру с заданным состоянием и проверяет, какие узлы были выполнены. Одного только метода compile() недостаточно для подтверждения полноты охвата бизнес-логики.

Называйте узлы в соответствии с бизнес-шагами (classify, retrieve, draft), а не node1. Читаемые структуры — это структуры, с которыми можно работать.

Называйте узлы в соответствии с бизнес-шагами (classify, retrieve, draft), а не node1. Читаемые графы — это работоспособные графы.

Добавьте тест на единицу, который запускает скомпилированный граф с заданном состоянием и проверяет, какие узлы были выполнены. Одного только метода compile() недостаточно для подтверждения полноты реализации бизнес-логики.

Добавьте тест единицы, который запускает скомпилированный граф с заданном состоянием фикстчера и проверяет, какие узлы были выполнены. Одного только метода compile() недостаточно для подтверждения полноты реализации бизнес-логики.

Для тестов маршрутизации предпочитайте детерминистичные узлы: используйте фиктивный LLM, возвращающий фиксированные вызовы инструментов, чтобы CI не требовал сетевого доступа.

Заранее задокументируйте правило авторизации thread_id: наличие чекпоинта не означает право на его чтение.

Даже в прототипах части 0 держите инструменты с побочными эффектами за ребрами одобрения — привычки формируются быстро.

Заранее задокументируйте правило авторизации thread_id: нахождение чекпоинта не означает разрешения на его чтение.

Даже в прототипах части 0 держите инструменты с побочными эффектами за границей механизмов утверждения — привычки формируются быстро.

Для тестов маршрутизации предпочитайте детерминистичные узлы: вставьте фиктивный LLM, который возвращает фиксированные вызовы инструментов, чтобы CI не нуждался в сетевом доступе.

← Все статьи Прислать бриф