Граф вместо готового цикла | Курс LangGraph урок 1
8
Цель урока: собрать и запустить свой первый StateGraph и сравнить его с агентом из create_agent.
Необходимые знания:
1) урок 7 курса по LangChain: поток, режим updates, форма chunk при version="v2"
2) урок 11 курса по LangChain: create_agent, узлы model и tools, цикл вызова инструментов
3) Python на уровне джуниора: виртуальное окружение, pip, TypedDict, модуль pathlib, генераторные выражения
Зачем собирать граф руками
Агент из create_agent, это готовый цикл: модель запрашивает инструменты, узел инструментов их выполняет, модель получает результаты и решает, продолжать ли. Пока задача укладывается в этот цикл, свой граф не нужен.
Цикла не хватает в трёх случаях.
1) Заранее известный порядок работы. Разбор чужого репозитория идёт этапами: собрать файлы, посчитать строки, разобрать импорты, свести всё в карточку репозитория. Порядок задаёте вы, а не модель, и в примерах 1-4 этапы выполняет обычный код на Python.
2) Возврат назад: проверка провалилась, план надо переписать и пройти участок заново. В готовом цикле вернуться можно только к очередному вызову модели.
3) Остановка в любом месте. HumanInTheLoopMiddleware из урока 17 курса по LangChain останавливает агента только перед вызовом инструмента. В своём графе запуск можно прервать в любом узле, например чтобы человек проверил план, и продолжить с этого узла, хоть на следующий день.
Эти случаи закрывает LangGraph. Он даёт состояние, узлы и переходы между ними, а промпты и устройство агента задаёте вы. Модели и инструменты приходят из LangChain, LangGraph отвечает за порядок работы.
Окружение и стенд для примеров
Для урока создайте отдельную папку, например langgraph_lesson_01, и работайте в ней. К концу раздела в ней будет так:
langgraph_lesson_01/
.venv/
.env
.gitignore
requirements.txt
course_model.py
mini_repo/
app.py
config.py
db.py
utils.py
Каждый пример урока сохраните в этой же папке отдельным файлом, например 01_first_graph.py. Запускайте его из папки при активном окружении командой python 01_first_graph.py. Пути к стенду примеры считают от своего файла, поэтому в Jupyter они не запустятся.
Пакеты langgraph и langchain требуют Python 3.10 или новее, примеры курса запускались на Python 3.12. Проверьте, что версия не ниже 3.10, и создайте в папке урока виртуальное окружение. На macOS и Linux:
python3 --version
python3 -m venv .venv
source .venv/bin/activate
На Windows:
py --version
py -m venv .venv
.venv\Scripts\activate
Положите в папку урока файл requirements.txt с версиями, на которых получен вывод примеров:
langchain[openai]==1.4.2
langchain-core==1.6.3
langchain-openai==1.6.2
langgraph==1.2.11
langgraph-checkpoint==4.2.0
langgraph-checkpoint-sqlite==3.1.1
aiosqlite==0.22.1
python-dotenv==1.2.3
Поставьте набор командой pip install -r requirements.txt. langchain[openai] приносит интеграцию с OpenAI и с любым адресом, совместимым с OpenAI. python-dotenv читает переменные из файла .env. langgraph-checkpoint-sqlite и aiosqlite понадобятся в уроке 5, список один на все уроки курса.
Первые четыре примера работают без модели, пятому нужен ключ провайдера. Создайте файл .env:
# Ключ провайдера. Для OpenAI и для любого адреса, совместимого с OpenAI.
OPENAI_API_KEY=
# Имя модели из списка вашего провайдера.
MODEL_NAME=
# Адрес, совместимый с OpenAI Chat Completions API.
# Оставьте пустым, если работаете с провайдером напрямую.
MODEL_BASE_URL=
MODEL_NAME и MODEL_BASE_URL, это переменные курса, LangChain про них не знает. Их читает модуль course_model.py, он собирает модель для всех примеров курса. Вывод примеров получен на модели deepseek/deepseek-v4-flash, запросы шли на адрес шлюза, совместимого с OpenAI.
Если работаете с другим провайдером напрямую, допишите в requirements.txt дополнение этого провайдера, например строку langchain[anthropic]==1.4.2. Его ключ допишите в .env отдельной строкой, например ANTHROPIC_API_KEY=. MODEL_BASE_URL оставьте пустым, а в MODEL_NAME пишите имя с провайдером: anthropic:claude-sonnet-4-6.
Пример 5 передаёт модели temperature=0. Модели claude-opus-5, claude-sonnet-5, claude-fable-5, claude-opus-4-8 и claude-opus-4-7 на любое значение temperature, кроме значения по умолчанию, отвечают ошибкой 400. С ними уберите temperature=0 из вызова build_model в примере 5.
Чтобы ключ не попал в репозиторий, добавьте файл .gitignore:
.env
.venv/
__pycache__/
*.pyc
Рядом положите модуль course_model.py:
"""Сборка модели для примеров курса по LangGraph.
build_model() собирает модель по переменным MODEL_NAME, MODEL_BASE_URL и
OPENAI_API_KEY из файла .env. У build_model есть необязательный первый аргумент
с именем модели, а рядом лежит gateway_kwargs() для примеров, которые собирают
модель сами.
load_dotenv() ищет файл .env, начиная с папки этого модуля, и поднимается вверх.
"""
import os
from dotenv import load_dotenv
from langchain.chat_models import init_chat_model
load_dotenv()
def gateway_kwargs():
"""Возвращает аргументы доступа к провайдеру: имя, адрес, ключ.
Нужны примерам, которые вызывают init_chat_model сами. У настраиваемой модели
имени модели при создании нет, поэтому build_model ей не подходит.
"""
base_url = os.getenv("MODEL_BASE_URL")
if base_url:
return {
"model_provider": "openai",
"base_url": base_url,
"api_key": os.environ["OPENAI_API_KEY"],
}
return {}
def build_model(model_name=None, **kwargs):
"""Собирает модель курса.
model_name без значения означает модель из переменной MODEL_NAME. Явное имя
нужно примерам, где моделей в приложении больше одной.
Все именованные аргументы уходят в init_chat_model как есть: temperature,
max_tokens, timeout, max_retries, rate_limiter, profile и прочее из раздела
Parameters.
"""
model_name = model_name or os.environ["MODEL_NAME"]
access = gateway_kwargs()
if access:
# Путь для любого адреса, совместимого с OpenAI Chat Completions API.
return init_chat_model(model=model_name, **access, **kwargs)
# Путь напрямую к провайдеру.
return init_chat_model(model_name, **kwargs)
Примеры 1, 3, 4 и 5 работают с одним стендом, выдуманным репозиторием из четырёх файлов. Рядом с примерами создайте папку mini_repo и положите в неё файлы ниже.
Файл app.py:
"""Точка входа выдуманного сервиса заказов."""
from config import SETTINGS
from db import connect
def main():
connection = connect(SETTINGS["dsn"])
return connection.fetch("select count(*) from orders")
if __name__ == "__main__":
print(main())
Файл config.py:
"""Настройки сервиса заказов."""
SETTINGS = {
"dsn": "postgres://localhost/orders",
"retries": 3,
}
Файл db.py:
"""Доступ к базе заказов."""
from config import SETTINGS
class Connection:
def __init__(self, dsn):
self.dsn = dsn
def fetch(self, query):
return [(query, SETTINGS["retries"])]
def connect(dsn):
return Connection(dsn)
Файл utils.py:
"""Утилиты сервиса заказов."""
def slugify(text):
return text.strip().lower().replace(" ", "-")
def chunked(items, size):
for start in range(0, len(items), size):
yield items[start:start + size]
Состояние, узлы, рёбра
Граф состоит из состояния, узлов и рёбер:
1) состояние (state) это общие данные запуска. Схема состояния задаётся один раз для всех узлов и рёбер
2) узлы (nodes) это функции. Узел получает состояние, делает работу и возвращает правку
3) рёбра (edges) это переходы. Ребро задаёт, какой узел выполняется следующим
Внутри узла может быть вызов модели или чтение файла.
Схема состояния пишется на TypedDict. Значения по умолчанию даёт dataclass, проверку данных даёт BaseModel из Pydantic, но он медленнее двух других. Схеме состояния отведён урок 2, здесь хватит словаря с типами.
Узел возвращает словарь только с теми ключами, которые меняет. Ключ, которого нет в правке, сохраняет прежнее значение. Ключ из правки получает новое значение, старое затирается. Сложить старое с новым умеет редьюсер из урока 2.
Кроме словаря узел может вернуть None, это значит "правок нет". Другое значение роняет запуск с InvalidUpdateError и текстом Expected dict. Ключ, которого нет в схеме, отбрасывается без ошибки, поэтому опечатку в имени ключа видно только в потоке updates.
Кроме состояния, узел может принять ещё два аргумента. config даёт настройки запуска, включая thread_id и метки. runtime даёт контекст, хранилище, функцию записи в поток и сведения о текущем запуске. Оба появятся в следующих уроках.
Вход и выход отмечают два особых имени. Ребро из START задаёт узел, с которого начинается запуск. Ребро в END означает, что после этого узла работы нет.
Пример 01_first_graph.py
"""Пример 1 урока 1: первый StateGraph из двух узлов.
Граф обходит стенд mini_repo, который лежит рядом с этим файлом: первый узел
собирает имена файлов, второй считает в них строки. Схема состояния задана TypedDict,
узел возвращает правку состояния словарём.
"""
from pathlib import Path
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
class RepoState(TypedDict):
root: str
files: list[str]
total_lines: int
def collect_files(state: RepoState) -> dict:
"""Собирает имена файлов Python в папке стенда."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"files": names}
def count_lines(state: RepoState) -> dict:
"""Считает строки во всех собранных файлах."""
root = Path(state["root"])
total = 0
for name in state["files"]:
total += len((root / name).read_text(encoding="utf-8").splitlines())
return {"total_lines": total}
builder = StateGraph(RepoState)
builder.add_node("collect_files", collect_files)
builder.add_node("count_lines", count_lines)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "count_lines")
builder.add_edge("count_lines", END)
graph = builder.compile()
result = graph.invoke({"root": str(REPO)})
print("ТИП СОБРАННОГО ГРАФА:", type(graph).__name__)
print("ФАЙЛЫ:", result["files"])
print("СТРОК ВСЕГО:", result["total_lines"])
print("КЛЮЧИ СОСТОЯНИЯ НА ВЫХОДЕ:", sorted(result))
# Вывод:
# ТИП СОБРАННОГО ГРАФА: CompiledStateGraph
# ФАЙЛЫ: ['app.py', 'config.py', 'db.py', 'utils.py']
# СТРОК ВСЕГО: 44
# КЛЮЧИ СОСТОЯНИЯ НА ВЫХОДЕ: ['files', 'root', 'total_lines']
Первый аргумент add_node задаёт имя узла, и рёбра ссылаются на это имя. Если передать одну функцию, add_node(collect_files), узел получит имя функции.
Узел получает состояние целиком, а возвращает часть. count_lines читает ключ files, которого на входе в граф не было: его положил предыдущий узел.
На вход invoke хватает одного ключа. Остальные появляются по ходу запуска, а на выходе приходит состояние целиком.
Цепочку из примера 1 можно собрать короче. add_sequence([collect_files, count_lines]) добавляет оба узла и ребро между ними, ребро из START к collect_files пишется отдельно. Ребро в END не нужно: после count_lines выполнять нечего, и запуск кончается.
Компиляция: что она проверяет
У StateGraph нет ни invoke, ни stream: эти методы есть у объекта, который возвращает .compile().
Компиляция проверяет структуру графа и принимает настройки запуска: чекпойнтер, кеш, хранилище, точки останова до и после узла. Настройки разобраны в следующих уроках, проверку структуры показывает пример 2.
Пример 02_compile.py
"""Пример 2 урока 1: что compile проверяет и чего не проверяет.
Четыре сборки одного графа: запуск без компиляции, граф без входа, ребро в
несуществующий узел и узел, к которому не ведёт ни одного ребра. Первые три
падают на месте, четвёртая компилируется и запускается.
"""
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
class State(TypedDict):
steps: list[str]
def step_a(state: State) -> dict:
return {"steps": state["steps"] + ["a"]}
def step_b(state: State) -> dict:
return {"steps": state["steps"] + ["b"]}
def fresh() -> StateGraph:
"""Возвращает сборщик с двумя узлами и без единого ребра."""
builder = StateGraph(State)
builder.add_node("step_a", step_a)
builder.add_node("step_b", step_b)
return builder
# 1. Запуск без компиляции.
builder = fresh()
builder.add_edge(START, "step_a")
builder.add_edge("step_a", END)
try:
builder.invoke({"steps": []})
except AttributeError as error:
print("1. без компиляции:", error)
# 2. Ни одного ребра из START.
builder = fresh()
builder.add_edge("step_a", END)
try:
builder.compile()
except ValueError as error:
print("2. без входа:", error)
# 3. Ребро ведёт в узел, которого в графе нет.
builder = fresh()
builder.add_edge(START, "step_a")
builder.add_edge("step_a", "step_c")
try:
builder.compile()
except ValueError as error:
print("3. чужое имя:", error)
# 4. Узел step_b добавлен, рёбер к нему нет.
builder = fresh()
builder.add_edge(START, "step_a")
builder.add_edge("step_a", END)
graph = builder.compile()
print("4. узел без рёбер, результат:", graph.invoke({"steps": []}))
print(" узлы собранного графа:", sorted(graph.get_graph().nodes))
# Вывод:
# 1. без компиляции: 'StateGraph' object has no attribute 'invoke'
# 2. без входа: Graph must have an entrypoint: add at least one edge from START to another node
# 3. чужое имя: Found edge ending at unknown node `step_c`
# 4. узел без рёбер, результат: {'steps': ['a']}
# узлы собранного графа: ['__end__', '__start__', 'step_a', 'step_b']
Первые три сборки падают раньше, чем выполнится хоть один узел: первая на invoke, вторая и третья на compile().
В чужом коде вместо рёбер из START и в END часто стоят set_entry_point и set_finish_point. Это те же два ребра: set_entry_point(key) равносилен add_edge(START, key), set_finish_point(key) равносилен add_edge(key, END). В примерах курса стоит запись с add_edge.
Четвёртая сборка компилируется без ошибки: узел step_b в графе есть, но ни одно ребро к нему не ведёт, и он не выполняется. Страница Graph API в документации обещает при компиляции проверку на узлы без связей, в langgraph 1.2.11 такой проверки нет.
Если узел не выполнился, первым делом проверьте, ведёт ли к нему ребро. На схеме графа из draw_mermaid(), её печатает пример 4, у такого узла нет входящей стрелки, а в потоке updates от него нет ни одного chunk.
Шаг выполнения: почему параллельные узлы, это один шаг
Граф работает передачей сообщений. Здесь сообщение это сигнал движка от узла к узлу, с сообщениями модели оно не связано. Узел закончил работу и отправил сообщение по исходящим рёбрам. Узел, получивший сообщение, становится активным и выполняется. Запуск кончается, когда активных узлов не осталось и сообщений в пути нет.
Этот порядок взят из системы Pregel от Google, оттуда же название шага. Шаг выполнения (super-step), это одна итерация: выполняются все активные узлы, и только потом их правки применяются к состоянию. Параллельные узлы попадают в один шаг, последовательные в разные.
Пример 3 печатает для каждого снимка состояния заполненные ключи, кроме входного root.
Пример 03_super_step.py
"""Пример 3 урока 1: один шаг графа против двух.
Четыре узла и две схемы связи. В последовательной схеме счёт строк и разбор
импортов стоят друг за другом, в параллельной, рядом. В режиме values поток
отдаёт снимок состояния на входе и после каждого шага, поэтому снимков на один
больше, чем шагов.
"""
from pathlib import Path
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
class RepoState(TypedDict):
root: str
files: list[str]
total_lines: int
imports: list[str]
card: str
def collect_files(state: RepoState) -> dict:
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"files": names}
def count_lines(state: RepoState) -> dict:
root = Path(state["root"])
total = 0
for name in state["files"]:
total += len((root / name).read_text(encoding="utf-8").splitlines())
return {"total_lines": total}
def collect_imports(state: RepoState) -> dict:
root = Path(state["root"])
found = set()
for name in state["files"]:
for line in (root / name).read_text(encoding="utf-8").splitlines():
if line.startswith("from ") or line.startswith("import "):
found.add(line.split()[1])
return {"imports": sorted(found)}
def make_card(state: RepoState) -> dict:
card = (
f"файлов {len(state['files'])}, "
f"строк {state['total_lines']}, "
f"импортов {len(state['imports'])}"
)
return {"card": card}
def sequential() -> StateGraph:
"""Четыре узла цепочкой: каждый ждёт предыдущего."""
builder = StateGraph(RepoState)
builder.add_node("collect_files", collect_files)
builder.add_node("count_lines", count_lines)
builder.add_node("collect_imports", collect_imports)
builder.add_node("make_card", make_card)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "count_lines")
builder.add_edge("count_lines", "collect_imports")
builder.add_edge("collect_imports", "make_card")
builder.add_edge("make_card", END)
return builder
def parallel() -> StateGraph:
"""Те же узлы, но счёт строк и разбор импортов стоят рядом."""
builder = StateGraph(RepoState)
builder.add_node("collect_files", collect_files)
builder.add_node("count_lines", count_lines)
builder.add_node("collect_imports", collect_imports)
builder.add_node("make_card", make_card)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "count_lines")
builder.add_edge("collect_files", "collect_imports")
builder.add_edge("count_lines", "make_card")
builder.add_edge("collect_imports", "make_card")
builder.add_edge("make_card", END)
return builder
def run(title: str, builder: StateGraph) -> None:
graph = builder.compile()
print(title)
number = 0
for chunk in graph.stream({"root": str(REPO)}, stream_mode="values", version="v2"):
number += 1
filled = sorted(key for key in chunk["data"] if key != "root")
print(f" снимок {number}: заполнены {filled}")
print(f" снимков всего: {number}")
run("ПОСЛЕДОВАТЕЛЬНАЯ СХЕМА", sequential())
print()
run("ПАРАЛЛЕЛЬНАЯ СХЕМА", parallel())
# Вывод:
# ПОСЛЕДОВАТЕЛЬНАЯ СХЕМА
# снимок 1: заполнены []
# снимок 2: заполнены ['files']
# снимок 3: заполнены ['files', 'total_lines']
# снимок 4: заполнены ['files', 'imports', 'total_lines']
# снимок 5: заполнены ['card', 'files', 'imports', 'total_lines']
# снимков всего: 5
#
# ПАРАЛЛЕЛЬНАЯ СХЕМА
# снимок 1: заполнены []
# снимок 2: заполнены ['files']
# снимок 3: заполнены ['files', 'imports', 'total_lines']
# снимок 4: заполнены ['card', 'files', 'imports', 'total_lines']
# снимков всего: 4
В обеих схемах четыре узла, но последовательная проходит их за четыре шага, а параллельная за три. В параллельной схеме из collect_files выходят два ребра, и оба узла назначения выполняются в следующем шаге вместе. Так можно, потому что счёт строк и разбор импортов не зависят друг от друга.
Первый снимок в обоих запусках, это вход до работы узлов, дальше по снимку на шаг.
Если при компиляции передан чекпойнтер, на границе шага он сохраняет checkpoint, состояние запуска на этот момент. В примерах этого урока чекпойнтера нет, и ничего не сохраняется. От границы шага работают повтор узла (урок 4), продолжение с того же места и возврат назад (урок 5), пауза для человека (урок 6).
Правки узлов одного шага применяются вместе. Пока узлы пишут в разные ключи, ошибки нет. Если два узла пишут в один ключ, запуск падает с InvalidUpdateError, это исправляет редьюсер из урока 2.
Как смотреть на запуск: updates, values и схема графа
invoke возвращает только итог. В графе из четырёх узлов нужно знать, какой узел отработал, в каком порядке и что прислал.
Для этого у собранного графа есть метод stream. Он принимает режим потока или список режимов, а version="v2" приводит все chunks к одной форме: словарь с ключами type, ns и data. Без него форма chunk зависит от числа режимов.
Режимы потока:
| Режим | Что приходит | Нужен чекпойнтер |
|---|---|---|
values |
состояние целиком после каждого шага | нет |
updates |
правки состояния после каждого шага, по одной на узел | нет |
messages |
пары: chunk ответа модели и его метаданные | нет |
custom |
данные, которые узел записал функцией из get_stream_writer |
нет |
checkpoints |
сохранённые checkpoints в той же форме, что у get_state() |
да, без него событий нет |
tasks |
старт и финиш задач с результатами и ошибками | нет |
debug |
события checkpoints и tasks с номером шага и временем |
без него приходят только события задач |
messages и custom разобраны в уроке 7 курса по LangChain, последние три понадобятся для отладки в уроке 8 этого курса. Документация пишет, что режиму tasks нужен чекпойнтер. В langgraph 1.2.11 события задач приходят и без него. Здесь работают values и updates.
Пример 04_watch_run.py
"""Пример 4 урока 1: два режима потока и схема графа.
Тот же параллельный граф, что в примере 3, запущен потоком сразу в двух режимах.
Режим updates отдаёт правку и имя узла, режим values отдаёт состояние целиком.
Дальше пример печатает схему графа на языке mermaid и сохраняет её картинкой.
Картинку рисует внешний сервис, поэтому вызов обёрнут в try.
"""
from pathlib import Path
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
PICTURE = Path(__file__).parent / "graph.png"
class RepoState(TypedDict):
root: str
files: list[str]
total_lines: int
imports: list[str]
card: str
def collect_files(state: RepoState) -> dict:
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"files": names}
def count_lines(state: RepoState) -> dict:
root = Path(state["root"])
total = 0
for name in state["files"]:
total += len((root / name).read_text(encoding="utf-8").splitlines())
return {"total_lines": total}
def collect_imports(state: RepoState) -> dict:
root = Path(state["root"])
found = set()
for name in state["files"]:
for line in (root / name).read_text(encoding="utf-8").splitlines():
if line.startswith("from ") or line.startswith("import "):
found.add(line.split()[1])
return {"imports": sorted(found)}
def make_card(state: RepoState) -> dict:
card = (
f"файлов {len(state['files'])}, "
f"строк {state['total_lines']}, "
f"импортов {len(state['imports'])}"
)
return {"card": card}
builder = StateGraph(RepoState)
builder.add_node("collect_files", collect_files)
builder.add_node("count_lines", count_lines)
builder.add_node("collect_imports", collect_imports)
builder.add_node("make_card", make_card)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "count_lines")
builder.add_edge("collect_files", "collect_imports")
builder.add_edge("count_lines", "make_card")
builder.add_edge("collect_imports", "make_card")
builder.add_edge("make_card", END)
graph = builder.compile()
for chunk in graph.stream(
{"root": str(REPO)},
stream_mode=["updates", "values"],
version="v2",
):
if chunk["type"] == "updates":
for node, update in chunk["data"].items():
print(f"updates: узел {node} прислал ключи {sorted(update)}")
elif chunk["type"] == "values":
print(f"values: состояние целиком, ключей {len(chunk['data'])}")
print()
print(graph.get_graph().draw_mermaid())
try:
graph.get_graph().draw_mermaid_png(output_file_path=str(PICTURE))
print("картинка сохранена:", PICTURE.name)
except Exception as error:
print("картинку получить не вышло:", type(error).__name__, error)
# Вывод:
# values: состояние целиком, ключей 1
# updates: узел collect_files прислал ключи ['files']
# values: состояние целиком, ключей 2
# updates: узел collect_imports прислал ключи ['imports']
# updates: узел count_lines прислал ключи ['total_lines']
# values: состояние целиком, ключей 4
# updates: узел make_card прислал ключи ['card']
# values: состояние целиком, ключей 5
#
# ---
# config:
# flowchart:
# curve: linear
# ---
# graph TD;
# __start__([<p>__start__</p>]):::first
# collect_files(collect_files)
# count_lines(count_lines)
# collect_imports(collect_imports)
# make_card(make_card)
# __end__([<p>__end__</p>]):::last
# __start__ --> collect_files;
# collect_files --> collect_imports;
# collect_files --> count_lines;
# collect_imports --> make_card;
# count_lines --> make_card;
# make_card --> __end__;
# classDef default fill:#f2f0ff,line-height:1.2
# ... (подрезано: ещё две строки classDef)
#
# картинка сохранена: graph.png
Параллельные узлы дают два отдельных chunk updates, а chunk values между ними нет: снимок состояния приходит один на весь шаг.
Порядок двух chunks updates внутри шага не закреплён, опираться на него в коде нельзя.
Метод draw_mermaid() возвращает схему графа текстом на языке mermaid, draw_mermaid_png() делает из неё картинку. По умолчанию картинку рисует внешний сервис mermaid.ink, без сети останется только текст схемы.
Ещё один способ смотреть на запуск, это stream_events(version="v3") из урока 8 курса по LangChain: он раскладывает запуск на проекции, и chunks не приходится разбирать руками. Версия v3 в коде пакета помечена как экспериментальная.
Граница: Graph API против Functional API
У LangGraph два интерфейса поверх одного движка.
Graph API, это StateGraph, узлы, рёбра и общее состояние из этого урока. Его берут для явного состояния, ветвления с несколькими точками решения, параллельных путей со сборкой результатов и наглядной схемы.
Functional API, это декораторы @entrypoint и @task поверх обычного кода. Порядок задают if и циклы Python, состояние остаётся внутри функции, узлы и рёбра объявлять не нужно. Его берут, чтобы подключить LangGraph к готовому коду почти без переписывания.
Оба интерфейса дают сохранение состояния, поток, паузу для человека и память, и их можно смешивать в одном приложении.
Курс идёт по Graph API: дальше нужны явное состояние и явные переходы, на них стоят все уроки от ветвления до подграфов.
create_agent тоже граф
create_agent из LangChain собирает граф LangGraph. Всё, что урок 7 скажет о сохранении состояния подграфов, действует и на агента, встроенного в граф.
Пример 5 собирает модель функцией build_model из модуля course_model.py, листинг модуля стоит в разделе "Окружение и стенд для примеров".
Пример 05_agent_is_graph.py
"""Пример 5 урока 1: тот же разбор стенда, но агентом.
Агент собран функцией create_agent с двумя инструментами для того же стенда
mini_repo. Печатаются класс агента, имена узлов его графа и правки узлов при
запуске в режиме updates. Порядок вызовов инструментов выбирает модель,
температура нулевая, чтобы порядок меньше менялся от запуска к запуску.
"""
from pathlib import Path
from langchain.agents import create_agent
from course_model import build_model
REPO = Path(__file__).parent / "mini_repo"
def list_files() -> str:
"""Возвращает имена файлов Python в репозитории."""
names = sorted(path.name for path in REPO.glob("*.py"))
return ", ".join(names)
def file_lines(name: str) -> str:
"""Возвращает число строк в файле репозитория по его имени."""
path = REPO / name
if not path.exists():
return f"Файла {name} в репозитории нет."
return f"{name}: {len(path.read_text(encoding='utf-8').splitlines())} строк"
agent = create_agent(
model=build_model(temperature=0, max_tokens=512),
tools=[list_files, file_lines],
system_prompt=(
"Вы разбираете чужой репозиторий. Отвечайте по-русски, коротко. "
"Числа берите только из инструментов, своих догадок не добавляйте."
),
)
print("КЛАСС АГЕНТА:", type(agent).__name__)
print("УЗЛЫ ГРАФА АГЕНТА:", sorted(agent.get_graph().nodes))
print()
update_no = 0
for chunk in agent.stream(
{"messages": [{"role": "user", "content": "Сколько строк в файле db.py и какие ещё файлы есть?"}]},
stream_mode="updates",
version="v2",
):
if chunk["type"] != "updates":
continue
for node, update in chunk["data"].items():
update_no += 1
messages = update.get("messages", []) if isinstance(update, dict) else []
for message in messages:
calls = getattr(message, "tool_calls", None)
if calls:
print(f"правка {update_no}: узел {node}, запрошены {[call['name'] for call in calls]}")
else:
text = message.text.replace("\n", " ")
print(f"правка {update_no}: узел {node}, {type(message).__name__}: {text[:70]}")
print()
print("ПРАВОК ОТ УЗЛОВ:", update_no)
# Вывод:
# КЛАСС АГЕНТА: CompiledStateGraph
# УЗЛЫ ГРАФА АГЕНТА: ['__end__', '__start__', 'model', 'tools']
#
# правка 1: узел model, запрошены ['file_lines', 'list_files']
# правка 2: узел tools, ToolMessage: db.py: 15 строк
# правка 3: узел tools, ToolMessage: app.py, config.py, db.py, utils.py
# правка 4: узел model, AIMessage: Файл **db.py** содержит **15 строк**. В репозитории также есть файлы:
#
# ПРАВОК ОТ УЗЛОВ: 4
Класс тот же, что в примере 1, с теми же методами invoke, stream, get_graph. Разница в том, кто задаёт узлы и порядок.
В своём графе узлы ваши, число шагов задано схемой. У агента узлы model и tools, а число шагов зависит от ответов модели. Два инструмента, запрошенные одним сообщением, выполняются в одном шаге, запрошенные по одному, в разных.
Агента берут там, где порядок действий заранее неизвестен и его выбирает модель. Свой граф берут там, где порядок известен вам, а модель нужна в отдельных узлах. Агента можно встроить в граф узлом, это урок 7 про подграфы.
Практическое задание
Соберите в файле repo_report.py граф, который строит отчёт по тому же стенду mini_repo.
Требования:
1) в состоянии пять ключей: путь к папке, список файлов, имя файла с наибольшим числом строк, список файлов без импортов, текст отчёта
2) четыре узла: сбор файлов, поиск файла с наибольшим числом строк, поиск файлов без импортов, сборка отчёта
3) два средних узла стоят параллельно и пишут в разные ключи
4) запуск идёт потоком в режиме values, для каждого снимка печатаются заполненные ключи
5) в конце печатается схема графа из draw_mermaid()
Как проверить результат:
1) четыре снимка: вход, сбор файлов, параллельная пара, отчёт
2) в схеме графа из узла сбора файлов выходят две стрелки
3) если поменять параллельную пару на цепочку, снимков станет пять, а отчёт не изменится
4) если оба средних узла пишут в один ключ из схемы, запуск падает с InvalidUpdateError
Подсказка: файл без импортов ищется тем же разбором строк, что в примере 3, только с проверкой на пустой результат.
Итоги урока
Граф состоит из состояния, узлов и рёбер. Состояние хранит общие для всех узлов данные, узел это функция от состояния с правкой на выходе. Ребро задаёт переход, START и END отмечают вход и выход.
StateGraph держит описание, исполняемый объект даёт .compile(). Компиляция ловит неизвестный источник ребра, отсутствие входа и ребро в несуществующий узел. Узел без входящих рёбер она пропускает, его видно по схеме и по потоку updates.
Запуск идёт шагами: параллельные узлы попадают в один шаг, последовательные в разные, а с чекпойнтером на границе шага пишется checkpoint. Поток updates показывает, какой узел что изменил, values показывает состояние после каждого шага.
Агент из create_agent, это такой же собранный граф с узлами model и tools, порядок шагов в нём выбирает модель.
Без редьюсера правка затирает значение ключа, поэтому два узла одного шага не могут писать в один ключ.
В уроке 2 разберу состояние графа: три способа задать схему и редьюсер, который сводит текущее значение ключа с правкой узла. Ещё в нём add_messages для списка сообщений, отдельные схемы входа и выхода и каналы, на которых построено состояние.
Код урока
Примеры этого урока лежат в репозитории курса, папка lesson_01. Закреплённые версии, на которых получен вывод в тексте, лежат в requirements.txt в корне репозитория.
Предыдущий урок: Подтверждение человеком и guardrails, курс по LangChain
Подписывайтесь на мой Telegram канал
Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия
Авторизуйтесь, чтобы оставить комментарий.
Нет комментариев.
Тут может быть ваша реклама
Пишите info@aisferaic.ru