Блоки будови LangGraph: стани, вузли, ребра та виклики

Коли моделі навчилися використовувати інструменти та здійснювати багатокрокове міркування, спеціальні цикли стали важкими для тестування та продовження роботи. 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 або reducers) зберігає повідомлення, флаги та проміжні результати. Ретельно обирайте reducers — використовуйте append для списків повідомлень та replace для скалярних значень.

Вузли

Вузли — це функції: вони читають стан, виконують роботу (LLM, інструменти, чистий Python) та повертають часткове оновлення. Робіть їх простими та піддаємими тестуванню.

Краї

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

Створення графів

Реєструйте вузли, під’єднуйте ребра, компілюйте з необов’язковим контрольним показником для забезпечення надійності.

Виклик графа

Виконуйте виклик із початковим станом та налаштуваннями (ідентифікатор потоку під час створення контрольної точки). Транслюйте події для покращення користувацького інтерфейсу.

Ланцюги проти графів

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

Повідомлення та моделі чату

Списки повідомлень є звичайним форматом зберігання інформації про розмову. Вбудовуйте моделі чату всередину вузлів, а не ховайте операції вводу-виводу у логіку ребер.

Інструменти

Робіть інструменти доступними для вузлів, які викликають моделі; аналізуйте виклики інструментів; записуйте результати спостережень назад у стан.

Використання повідомлень як стану

Коли повідомлення станом, редюсери повинні додавати їх безпечно, а обрізка/узагальнення стає явним елементом — а не випадковістю.

Закінчення

Частина 0 присвятована лексикології: стан, вузли, ребра, компіляція, виклик. У наступних частинах на цій основі додаються стійкі потоки, переривання HITL та патерни багатьох агентів.

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

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

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

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

Називайте вузли за бізнес-кроками (classify, retrieve, draft), а не node1. Читабельні структури є також операційними.

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

Називайте вузли за бізнес-кроками (classify, retrieve, draft), а не node1. Читабельні структури є також операційними.

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

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

Додайте тест на одиницю, який запускає скомпільований граф у стані фікстури та перевіряє, які вузли були виконані. Сама функція compile() не доводить повноти виконання бізнес-логіки.

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

Додайте тест на одиницю, який запускає скомпільований граф у стані фікстури та перевіряє, які вузли були виконані. Сама функція compile() не доводить повноти виконання бізнес-логіки.

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

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

Називайте вузли за бізнес-кроками (classify, retrieve, draft), а не node1. Читабельні структури є також операційними.

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

Називайте вузли за бізнес-кроками (classify, retrieve, draft), а не node1. Читабельні структури є також операційними.

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

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

Додайте тест на одиницю, який запускає скомпільований граф у стані фікстури та перевіряє, які вузли були виконані. Сама функція compile() не доводить повноти виконання бізнес-логіки.

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

Додайте тест на одиницю, який запускає скомпільований граф у стані фікстури та перевіряє, які вузли були виконані. Сама функція compile() не доводить повноти виконання бізнес-логіки.

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

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

Називайте вузли за бізнес-кроками (classify, retrieve, draft), а не node1. Читабельні структури є операційними структурами.

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

Називайте вузли за бізнес-кроками (classify, retrieve, draft), а не node1. Читабельні структури є операційними структурами.

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

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

Додайте тест на одиницю, який запускає скомпільований граф у стані фікстури та перевіряє, які вузли були виконані. Сама функція compile() не доводить повноти виконання бізнес-логіки.

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

Додайте тест на одиницю, який запускає скомпільований граф у стані фікстури та перевіряє, які вузли були виконані. Сама функція compile() не доводить повноти виконання бізнес-логіки.

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

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

Називайте вузли за бізнес-кроками (classify, retrieve, draft), а не node1. Читабельні структури є також операційними.

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

Називайте вузли за бізнес-кроками (classify, retrieve, draft), а не node1. Читабельні структури є також операційними.

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

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

Додайте тест на одиницю, який запускає скомпільований граф із певним станом та перевіряє, які вузли були виконані. Сама функція compile() не доводить повноти виконання бізнес-логіки.

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

Якомога раніше задокументуйте правило автентифікації thread_id: наявність чекпоїнта не означає дозволу на його читання.

Навіть у прототипах частини 0 тримайте інструменти з побічними ефектами за межами країв схвалення — звички формуються швидко.

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

Якомога раніше задокументуйте правило автентифікації thread_id: знаходження контрольної точки не означає дозволу на її читання.

Зберігайте інструменти з побічними ефектами за межами процедур схвалення навіть у прототипах частини 0 — звички формуються швидко.

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

Якомога раніше задокументуйте правило автентифікації thread_id: знаходження контрольної точки не означає дозволу на її читання.

Зберігайте інструменти з побічними ефектами за межами процедур схвалення навіть у прототипах частини 0 — звички формуються швидко.

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