Циклы и остановка | Курс LangGraph урок 4
5
Цель урока: закольцевать участок графа и остановить его по признаку, до того как сработает предел шагов.
Необходимые знания:
1) урок 1 этого курса: StateGraph, узлы и рёбра, компиляция, шаг выполнения, схема графа из get_graph().draw_mermaid()
2) урок 2 этого курса: схема состояния на TypedDict, редьюсер, Annotated с operator.add, аннотация аргумента state, которая ограничивает поля на входе узла
3) урок 3 этого курса: add_conditional_edges, функция маршрутизации, возврат END из неё
4) Python на уровне джуниора: Literal, pathlib, работа с файлами, tempfile
Чем закончился урок 3
Граф из урока 3 умеет выбирать путь и раскладывать работу по веткам. При этом все его стрелки ведут вперёд, к END. Узел scan_file в этом графе выполнялся по разу на каждый файл, отдельной копией из Send. Все копии выполнялись в одном шаге, поэтому ни одна не получала результата другой.
Обходу большого репозитория нужно другое: брать файлы по одному, чтобы следующий проход получал результат предыдущего и граф мог остановиться посреди очереди. Второй случай, это проверка гипотезы о коде: один узел пишет заметку в отчёт, другой проверяет гипотезу. Если гипотеза не подтвердилась, граф возвращается к первому узлу, и тот пишет заметку заново.
Оба случая, это циклы, и оба упираются в вопрос, когда остановиться. Цикл без ответа на него доходит до предела шагов и роняет запуск.
Примеры работают на стенде из урока 1: положите папку mini_repo с файлами app.py, config.py, db.py и utils.py рядом с примерами, листинги файлов есть в уроке 1. Модель и ключ API в этом уроке не нужны.
Цикл: ребро назад и условный выход
Отдельного средства для цикла в Graph API нет, цикл собирается из знакомых частей:
1) обычное ребро назад: add_edge("scan_file", "pick_file"), то есть стрелка из узла в узел, который уже выполнялся
2) условное ребро с выходом на END: функция маршрутизации возвращает END, когда работа закончилась
Развилка ставится в начале каждого прохода цикла. Узел pick_file берёт следующий файл, функция маршрутизации проверяет его имя: пустое, значит END, иначе разбор. Так тело цикла не выполняется лишний раз.
При обратном порядке, сначала разбор, потом проверка, при пустой очереди разбор всё равно выполнится один раз. Узел scan_file из примера 1 при пустом имени собрал бы путь к папке стенда, и чтение упало бы с ошибкой. Узел с проверкой пустого имени, как в примере 2, не упадёт, но добавит в находки лишнюю строку.
Пример 01_loop.py
"""Пример 1 урока 4: цикл из ребра назад и условного выхода на END.
Граф обходит стенд mini_repo по одному файлу за проход. Узел pick_file берёт
следующее имя из очереди, узел scan_file считает в нём строки, ребро от
scan_file ведёт назад к pick_file. Выход на END стоит в функции маршрутизации
после pick_file, то есть проверяется до разбора.
"""
import operator
from pathlib import Path
from typing import Annotated, Literal
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
class WalkState(TypedDict):
root: str
queue: list[str]
current: str
findings: Annotated[list[str], operator.add]
def collect_files(state: WalkState) -> dict:
"""Складывает имена файлов стенда в очередь обхода."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"queue": names, "current": ""}
def pick_file(state: WalkState) -> dict:
"""Берёт следующее имя из очереди и убирает его оттуда."""
queue = list(state["queue"])
if not queue:
return {"current": ""}
return {"current": queue.pop(0), "queue": queue}
def scan_file(state: WalkState) -> dict:
"""Считает строки в выбранном файле."""
path = Path(state["root"]) / state["current"]
lines = len(path.read_text(encoding="utf-8").splitlines())
return {"findings": [f"{state['current']}: строк {lines}"]}
def route_next(state: WalkState) -> Literal["scan_file", "__end__"]:
"""Условие выхода: пустое имя означает, что очередь закончилась."""
if not state["current"]:
return END
return "scan_file"
builder = StateGraph(WalkState)
builder.add_node("collect_files", collect_files)
builder.add_node("pick_file", pick_file)
builder.add_node("scan_file", scan_file)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "pick_file")
builder.add_conditional_edges("pick_file", route_next)
builder.add_edge("scan_file", "pick_file")
graph = builder.compile()
# 1. Каждый шаг запуска отдельной строкой.
print("1. обход по шагам:")
for event in graph.stream(
{"root": str(REPO), "queue": [], "current": "", "findings": []},
stream_mode="updates",
):
print(" ", event)
# 2. Итог обхода.
result = graph.invoke(
{"root": str(REPO), "queue": [], "current": "", "findings": []}
)
print("2. находок:", len(result["findings"]))
for line in result["findings"]:
print(" ", line)
# 3. Ребро назад на схеме графа.
print("3. схема графа:")
print(graph.get_graph().draw_mermaid())
# Вывод:
# 1. обход по шагам:
# {'collect_files': {'queue': ['app.py', 'config.py', 'db.py', 'utils.py'], 'current': ''}}
# {'pick_file': {'current': 'app.py', 'queue': ['config.py', 'db.py', 'utils.py']}}
# {'scan_file': {'findings': ['app.py: строк 13']}}
# {'pick_file': {'current': 'config.py', 'queue': ['db.py', 'utils.py']}}
# {'scan_file': {'findings': ['config.py: строк 6']}}
# {'pick_file': {'current': 'db.py', 'queue': ['utils.py']}}
# {'scan_file': {'findings': ['db.py: строк 15']}}
# {'pick_file': {'current': 'utils.py', 'queue': []}}
# {'scan_file': {'findings': ['utils.py: строк 10']}}
# {'pick_file': {'current': ''}}
# 2. находок: 4
# app.py: строк 13
# config.py: строк 6
# db.py: строк 15
# utils.py: строк 10
# 3. схема графа:
# ---
# config:
# flowchart:
# curve: linear
# ---
# graph TD;
# __start__([<p>__start__</p>]):::first
# collect_files(collect_files)
# pick_file(pick_file)
# scan_file(scan_file)
# __end__([<p>__end__</p>]):::last
# __start__ --> collect_files;
# collect_files --> pick_file;
# pick_file -.-> __end__;
# pick_file -.-> scan_file;
# scan_file --> pick_file;
# classDef default fill:#f2f0ff,line-height:1.2
# ... (ещё две строки classDef)
На один файл уходит два шага выполнения. Для четырёх файлов стенда это шаг на сбор очереди, пять выполнений pick_file и четыре scan_file, всего десять шагов. На последнем выполнении pick_file возвращает пустое имя, и запуск заканчивается.
На схеме графа цикл виден стрелкой scan_file --> pick_file. Стрелки от pick_file пунктирные, их поставило условное ребро по аннотации Literal["scan_file", "__end__"]. Как аннотация работает на схеме и при запуске, разобрано в уроке 3.
Ключ findings с редьюсером operator.add копит находки. Ключ queue редьюсера не имеет, и каждый проход перезаписывает его списком на одно имя короче. Очередь пустеет, и цикл заканчивается.
Предел числа шагов: страховка от зацикливания
Число шагов одного вызова invoke или stream ограничено ключом recursion_limit. Когда запуск доходит до предела, LangGraph бросает GraphRecursionError из langgraph.errors.
Ключ ставится в конфигурации верхнего уровня, рядом с configurable: graph.invoke(inputs, {"recursion_limit": 11}). Внутри configurable ключ не действует, и ошибки при этом нет: запуск идёт с пределом по умолчанию.
Значение по умолчанию большое. Документация называет 1000 шагов начиная с версии 1.0.6, а langgraph 1.2.16 даёт 10007. Это число напечатает пример 3 в разделе про RemainingSteps.
Пример 02_recursion_limit.py
"""Пример 2 урока 4: предел числа шагов и GraphRecursionError.
Тот же обход стенда, что в примере 1, запускается с разными значениями
recursion_limit. Рядом стоит второй граф, у которого условие выхода забыли:
он повторяет цикл, пока предел не остановит его ошибкой. Счётчик вызовов узла
разбора показывает, сколько раз узел успел выполниться до остановки.
"""
import operator
from pathlib import Path
from typing import Annotated, Literal
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
SCANS: list[str] = []
class WalkState(TypedDict):
root: str
queue: list[str]
current: str
findings: Annotated[list[str], operator.add]
def collect_files(state: WalkState) -> dict:
"""Складывает имена файлов стенда в очередь обхода."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"queue": names, "current": ""}
def pick_file(state: WalkState) -> dict:
"""Берёт следующее имя из очереди и убирает его оттуда."""
queue = list(state["queue"])
if not queue:
return {"current": ""}
return {"current": queue.pop(0), "queue": queue}
def scan_file(state: WalkState) -> dict:
"""Считает строки в выбранном файле и отмечается в счётчике вызовов."""
SCANS.append(state["current"])
if not state["current"]:
return {"findings": ["разбирать нечего"]}
path = Path(state["root"]) / state["current"]
lines = len(path.read_text(encoding="utf-8").splitlines())
return {"findings": [f"{state['current']}: строк {lines}"]}
def route_next(state: WalkState) -> Literal["scan_file", "__end__"]:
"""Условие выхода на месте: пустое имя заканчивает обход."""
if not state["current"]:
return END
return "scan_file"
def route_forgotten(state: WalkState) -> Literal["scan_file"]:
"""Условие выхода забыто: маршрут всегда ведёт обратно в разбор."""
return "scan_file"
def build(router) -> StateGraph:
"""Собирает один и тот же цикл с разной функцией маршрутизации."""
builder = StateGraph(WalkState)
builder.add_node("collect_files", collect_files)
builder.add_node("pick_file", pick_file)
builder.add_node("scan_file", scan_file)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "pick_file")
builder.add_conditional_edges("pick_file", router)
builder.add_edge("scan_file", "pick_file")
return builder.compile()
def start() -> dict:
"""Входное состояние обхода."""
return {"root": str(REPO), "queue": [], "current": "", "findings": []}
graph = build(route_next)
broken = build(route_forgotten)
# 1. Обход четырёх файлов занимает десять шагов, и предела 10 ему не хватает.
for limit in (10, 11):
SCANS.clear()
try:
result = graph.invoke(start(), {"recursion_limit": limit})
print(f"1. предел {limit}: находок {len(result['findings'])}")
except GraphRecursionError as error:
print(f"1. предел {limit}: {type(error).__name__}")
print(" разборов до остановки:", len(SCANS))
print(" текст ошибки:", str(error).splitlines()[0])
# 2. Цикл без выхода: работа закончилась, а граф идёт дальше.
SCANS.clear()
try:
broken.invoke(start(), {"recursion_limit": 12})
except GraphRecursionError as error:
print("2. цикл без выхода:", type(error).__name__)
print(" разборов до остановки:", len(SCANS))
print(" что разбирал:", SCANS)
# 3. Предел повышен, но цикл без выхода упирается в него снова.
SCANS.clear()
try:
broken.invoke(start(), {"recursion_limit": 30})
except GraphRecursionError:
print("3. предел 30, разборов до остановки:", len(SCANS))
# Вывод:
# 1. предел 10: GraphRecursionError
# разборов до остановки: 4
# текст ошибки: Recursion limit of 10 reached without hitting a stop condition. You can increase the limit by setting the `recursion_limit` config key.
# 1. предел 11: находок 4
# 2. цикл без выхода: GraphRecursionError
# разборов до остановки: 5
# что разбирал: ['app.py', 'config.py', 'db.py', 'utils.py', '']
# 3. предел 30, разборов до остановки: 14
Предел проверяется в начале каждого шага, до поиска задач на этот шаг. После десятого шага задач уже нет, но до поиска дело не доходит: на пределе 10 запуск падает с ошибкой. Запуску из N шагов нужен предел N + 1, впритык подбирать нельзя.
invoke бросает исключение и не возвращает то, что успело собраться: четыре разбора в пункте 1 выполнились, и их результат пропал.
Метод stream отдаёт chunks шагов до исключения, и их можно сложить в свой список. В режиме updates это правки узлов, в режиме values состояние после каждого законченного шага, и последний chunk показывает, на чём запуск остановился.
Для цикла без выхода больший предел только откладывает ошибку. В пункте 2 очередь закончилась, а маршрут всё равно ведёт в разбор, и узел работает с пустым именем. Предел 30 в пункте 3 дал ту же ошибку, только разборов стало 14 вместо 5, из них пустых 10.
recursion_limit это предохранитель на случай, когда условие выхода написано неверно.
Счётчик шага и RemainingSteps
Во время запуска узлу доступны номер шага и остаток шагов до предела.
Номер текущего шага лежит в config["metadata"]["langgraph_step"]. Объект config узел получает параметром config: RunnableConfig, как scan_file в примере 3. LangGraph подставляет объект по имени параметра: узел с cfg: RunnableConfig вместо config падает с TypeError о пропущенном аргументе. Рядом в метаданных лежат имя узла langgraph_node и другие служебные ключи запуска.
Остаток шагов даёт поле схемы состояния с типом RemainingSteps из langgraph.managed. Значение подставляет LangGraph перед каждым шагом, узлы в это поле не пишут.
Значение из входного состояния поле не берёт: положите туда своё число, и узел всё равно получит остаток от предела. В итоговом состоянии этого ключа нет.
Пример 03_remaining_steps.py
"""Пример 3 урока 4: счётчик шага и RemainingSteps.
Тот же обход стенда. Узел разбора печатает номер текущего шага из
config["metadata"]["langgraph_step"] и остаток шагов из состояния. Второй граф
использует остаток как условие мягкого выхода: обход сворачивается сам и
возвращает то, что успел собрать, вместо GraphRecursionError.
"""
import operator
from pathlib import Path
from typing import Annotated, Literal
from langchain_core.runnables import RunnableConfig
from langgraph.errors import GraphRecursionError
from langgraph.graph import END, START, StateGraph
from langgraph.managed import RemainingSteps
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
class WalkState(TypedDict):
root: str
queue: list[str]
current: str
findings: Annotated[list[str], operator.add]
remaining_steps: RemainingSteps
def collect_files(state: WalkState) -> dict:
"""Складывает имена файлов стенда в очередь обхода."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"queue": names, "current": ""}
def pick_file(state: WalkState) -> dict:
"""Берёт следующее имя из очереди и убирает его оттуда."""
queue = list(state["queue"])
if not queue:
return {"current": ""}
return {"current": queue.pop(0), "queue": queue}
def scan_file(state: WalkState, config: RunnableConfig) -> dict:
"""Считает строки и печатает номер шага и остаток шагов."""
step = config["metadata"]["langgraph_step"]
remaining = state["remaining_steps"]
print(f" шаг {step}, остаток {remaining}, сумма {step + remaining}")
path = Path(state["root"]) / state["current"]
lines = len(path.read_text(encoding="utf-8").splitlines())
return {"findings": [f"{state['current']}: строк {lines}"]}
def route_next(state: WalkState) -> Literal["scan_file", "__end__"]:
"""Выход только по пустой очереди."""
if not state["current"]:
return END
return "scan_file"
def route_soft(state: WalkState) -> Literal["scan_file", "__end__"]:
"""Тот же выход плюс мягкая остановка за два шага до предела."""
if state["remaining_steps"] <= 2:
return END
if not state["current"]:
return END
return "scan_file"
def build(router) -> StateGraph:
"""Собирает один и тот же цикл с разной функцией маршрутизации."""
builder = StateGraph(WalkState)
builder.add_node("collect_files", collect_files)
builder.add_node("pick_file", pick_file)
builder.add_node("scan_file", scan_file)
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "pick_file")
builder.add_conditional_edges("pick_file", router)
builder.add_edge("scan_file", "pick_file")
return builder.compile()
def start() -> dict:
"""Входное состояние обхода."""
return {"root": str(REPO), "queue": [], "current": "", "findings": []}
graph = build(route_next)
soft = build(route_soft)
# 1. Предел задан, и остаток считается от него.
print("1. предел 11:")
result = graph.invoke(start(), {"recursion_limit": 11})
print(" ключи итога:", sorted(result))
# 2. Предел не задан, работает значение по умолчанию.
print("2. без предела:")
graph.invoke(start())
# 3. Жёсткая остановка: работа не доделана, ошибка вместо результата.
print("3. предел 7, жёсткое условие:")
try:
graph.invoke(start(), {"recursion_limit": 7})
except GraphRecursionError as error:
print(" ", type(error).__name__)
# 4. Мягкая остановка на том же пределе: результат неполный, но он есть.
print("4. предел 7, мягкое условие:")
partial = soft.invoke(start(), {"recursion_limit": 7})
print(" находок:", len(partial["findings"]), "осталось в очереди:", partial["queue"])
# Вывод:
# 1. предел 11:
# шаг 3, остаток 8, сумма 11
# шаг 5, остаток 6, сумма 11
# шаг 7, остаток 4, сумма 11
# шаг 9, остаток 2, сумма 11
# ключи итога: ['current', 'findings', 'queue', 'root']
# 2. без предела:
# шаг 3, остаток 10004, сумма 10007
# шаг 5, остаток 10002, сумма 10007
# шаг 7, остаток 10000, сумма 10007
# шаг 9, остаток 9998, сумма 10007
# 3. предел 7, жёсткое условие:
# шаг 3, остаток 4, сумма 7
# шаг 5, остаток 2, сумма 7
# шаг 7, остаток 0, сумма 7
# GraphRecursionError
# 4. предел 7, мягкое условие:
# шаг 3, остаток 4, сумма 7
# шаг 5, остаток 2, сумма 7
# находок: 2 осталось в очереди: ['utils.py']
Номер шага и остаток в сумме дают предел: RemainingSteps это тот же счётчик, отсчитанный от предела. Равенство верно для запуска с нуля, как во всех примерах урока. При продолжении сохранённого запуска (урок 5) номер шага идёт от прежнего счёта, а остаток от предела, отсчитанного заново, и сумма выходит больше предела.
Выход по остатку шагов, мягкая остановка, меняет результат. Запуск в пункте 4 закончился без ошибки и вернул две находки из четырёх. В очереди остался utils.py, а db.py уже снят с неё и стоит в current неразобранным: чтобы продолжить обход, непройденную часть надо собрать из обоих ключей. Запуск в пункте 3 на том же пределе закончился исключением и не вернул ничего.
Порог в функции маршрутизации зависит от того, сколько шагов стоит решение идти дальше: разбор, новое выполнение pick_file и последний, пустой шаг. Поэтому выход нужен, когда остатка меньше трёх, и порог 2 для цикла из двух узлов минимальный, запаса в нём нет. Цикл из большего числа узлов требует порога выше.
Условие выхода по делу (очередь пуста, гипотеза сошлась) и условие по остатку шагов стоят в функции маршрутизации рядом. Оба ведут на END, и порядок их проверки не важен. recursion_limit остаётся предохранителем, до которого дело не доходит.
Узел при возврате выполняется с начала
Узел, в который граф вернулся по ребру назад, выполняется с первой своей строки: каждое выполнение узла, это новый вызов его функции. Так же заново начинается узел при продолжении сохранённого запуска: checkpoint пишется на границе шага, внутри функции узла его нет.
Вместе с узлом повторяются его побочные действия: запись в файл, вставка строки в базу, отправка письма, создание тикета в трекере, коммит. LangGraph не отслеживает, что из этого уже сделано, и откатить сделанное не может.
Узел выполняется повторно в таких случаях:
1) цикл: граф вернулся по ребру назад, этот урок
2) продолжение сохранённого запуска, урок 5
3) возобновление после паузы, урок 6
4) повтор после падения узла, урок 8
Пример 04_repeat.py
"""Пример 4 урока 4: узел при возврате выполняется с начала.
Цикл проверки гипотезы: узел write_note записывает заметку в отчёт, узел check
проверяет гипотезу, и сходится она только после третьей заметки. Первый вариант
узла write_note дописывает строку в конец файла, второй пишет её по ключу. Тело узла
выполняется три раза в обоих случаях, а вот отчёт получается разный.
"""
import operator
import tempfile
from pathlib import Path
from typing import Annotated, Literal
from langgraph.graph import END, START, StateGraph
from typing_extensions import TypedDict
CALLS: list[int] = []
class CheckState(TypedDict):
target: str
report: str
attempt: int
passed: bool
findings: Annotated[list[str], operator.add]
def write_note_append(state: CheckState) -> dict:
"""Дописывает заметку в конец отчёта."""
CALLS.append(state["attempt"])
line = f"- {state['target']}: гипотеза про импорты\n"
with open(state["report"], "a", encoding="utf-8") as report:
report.write(line)
return {"attempt": state["attempt"] + 1}
def write_note_upsert(state: CheckState) -> dict:
"""Пишет ту же заметку по ключу: читает отчёт, убирает старую строку, ставит новую."""
CALLS.append(state["attempt"])
key = f"- {state['target']}:"
path = Path(state["report"])
kept = [
line
for line in path.read_text(encoding="utf-8").splitlines()
if not line.startswith(key)
]
kept.append(f"{key} гипотеза про импорты, попытка {state['attempt'] + 1}")
path.write_text("\n".join(kept) + "\n", encoding="utf-8")
return {"attempt": state["attempt"] + 1}
def check(state: CheckState) -> dict:
"""Проверка гипотезы: сходится после третьей заметки."""
passed = state["attempt"] >= 4
verdict = "сошлось" if passed else "не сошлось"
return {"passed": passed, "findings": [f"попытка {state['attempt']}: {verdict}"]}
def route_check(state: CheckState) -> Literal["write_note", "__end__"]:
"""Проверка провалена, значит заметка пишется заново."""
if state["passed"]:
return END
return "write_note"
def build(writer) -> StateGraph:
"""Собирает цикл с одним из двух вариантов заметки."""
builder = StateGraph(CheckState)
builder.add_node("write_note", writer)
builder.add_node("check", check)
builder.add_edge(START, "write_note")
builder.add_edge("write_note", "check")
builder.add_conditional_edges("check", route_check)
return builder.compile()
def run(writer, folder: Path, name: str) -> None:
"""Запускает цикл и печатает отчёт вместе с числом вызовов узла."""
CALLS.clear()
report = folder / name
report.write_text("# Отчёт обхода\n", encoding="utf-8")
graph = build(writer)
result = graph.invoke(
{
"target": "db.py",
"report": str(report),
"attempt": 1,
"passed": False,
"findings": [],
}
)
print(" тело узла выполнено раз:", len(CALLS), "с попытками", CALLS)
print(" проверки по попыткам:", result["findings"])
print(" отчёт:")
for line in report.read_text(encoding="utf-8").splitlines():
print(" ", line)
with tempfile.TemporaryDirectory() as tmp:
folder = Path(tmp)
print("1. заметка дописывается в конец:")
run(write_note_append, folder, "append.md")
print("2. заметка пишется по ключу:")
run(write_note_upsert, folder, "upsert.md")
# Вывод:
# 1. заметка дописывается в конец:
# тело узла выполнено раз: 3 с попытками [1, 2, 3]
# проверки по попыткам: ['попытка 2: не сошлось', 'попытка 3: не сошлось', 'попытка 4: сошлось']
# отчёт:
# # Отчёт обхода
# - db.py: гипотеза про импорты
# - db.py: гипотеза про импорты
# - db.py: гипотеза про импорты
# 2. заметка пишется по ключу:
# тело узла выполнено раз: 3 с попытками [1, 2, 3]
# проверки по попыткам: ['попытка 2: не сошлось', 'попытка 3: не сошлось', 'попытка 4: сошлось']
# отчёт:
# # Отчёт обхода
# - db.py: гипотеза про импорты, попытка 4
В обоих вариантах тело узла выполнилось три раза. Разница в отчёте: дописывание в конец дало три одинаковые строки, запись по ключу одну.
Второй вариант идемпотентный: повторный вызов не добавляет лишнего, строка в отчёте одна, сколько бы раз узел ни выполнился. Повтор делают безопасным запись по ключу (upsert), чтение перед записью и ключ идемпотентности, метка действия, по которой сервис на другой стороне узнаёт повтор. В примере работают первые два: узел читает отчёт, убирает свою прежнюю строку и ставит новую.
Для каждого действия узла за пределами состояния проверьте, что будет при двух выполнениях подряд. Счётчик в базе, увеличенный на единицу, повтора не переживёт, перезапись файла целиком переживёт. С письмом то же, что со счётчиком, пока сервис отправки не получает ключ идемпотентности.
Кеш узла: когда повтор не нужен
Разбор того же файла при повторном запуске даёт тот же ответ, и выполнять узел заново незачем.
Чтобы пропустить такой повтор, у узла есть кеш: cache_policy=CachePolicy(ttl=60) в add_node и cache=InMemoryCache() в compile. Кроме кеша в памяти процесса есть SqliteCache из langgraph.cache.sqlite, он приходит с пакетом langgraph-checkpoint-sqlite из списка урока 1.
Ключ кеша по умолчанию считается от входа узла: функция key_func из CachePolicy превращает вход в байты модулем pickle. Изменился вход, значит изменился ключ, и кеш не срабатывает.
При каждом проходе цикла состояние меняется: находки копятся, очередь укорачивается. Узел, который принимает состояние целиком, получает и находки, накопленные с начала запуска. Поэтому одна лишняя находка во входном состоянии меняет ключ кеша этого узла для всех файлов, и ни одна запись кеша от прошлого запуска не срабатывает.
Пример 05_cache.py
"""Пример 5 урока 4: кеш узла и его ключ.
Узел разбора помечен CachePolicy, граф скомпилирован с InMemoryCache. Первый
вариант узла принимает узкую схему входа (корень и текущий файл), второй читает
всё состояние целиком. Три запуска подряд показывают, когда кеш срабатывает, а
когда ключ расходится и узел выполняется заново.
"""
import operator
from pathlib import Path
from typing import Annotated, Literal
from langgraph.cache.memory import InMemoryCache
from langgraph.graph import END, START, StateGraph
from langgraph.types import CachePolicy
from typing_extensions import TypedDict
REPO = Path(__file__).parent / "mini_repo"
CALLS: list[str] = []
class WalkState(TypedDict):
root: str
queue: list[str]
current: str
findings: Annotated[list[str], operator.add]
class ScanIn(TypedDict):
root: str
current: str
def collect_files(state: WalkState) -> dict:
"""Складывает имена файлов стенда в очередь обхода."""
names = sorted(path.name for path in Path(state["root"]).glob("*.py"))
return {"queue": names, "current": ""}
def pick_file(state: WalkState) -> dict:
"""Берёт следующее имя из очереди и убирает его оттуда."""
queue = list(state["queue"])
if not queue:
return {"current": ""}
return {"current": queue.pop(0), "queue": queue}
def scan_narrow(state: ScanIn) -> dict:
"""Считает строки. Из состояния получает только корень и текущий файл."""
CALLS.append(state["current"])
path = Path(state["root"]) / state["current"]
lines = len(path.read_text(encoding="utf-8").splitlines())
return {"findings": [f"{state['current']}: строк {lines}"]}
def scan_wide(state: WalkState) -> dict:
"""Тот же разбор, но на входе всё состояние, включая накопленные находки."""
CALLS.append(state["current"])
path = Path(state["root"]) / state["current"]
lines = len(path.read_text(encoding="utf-8").splitlines())
return {"findings": [f"{state['current']}: строк {lines}"]}
def route_next(state: WalkState) -> Literal["scan_file", "__end__"]:
"""Выход по пустой очереди."""
if not state["current"]:
return END
return "scan_file"
def build(scanner) -> StateGraph:
"""Собирает обход с кешируемым узлом разбора и своим кешем в памяти."""
builder = StateGraph(WalkState)
builder.add_node("collect_files", collect_files)
builder.add_node("pick_file", pick_file)
builder.add_node("scan_file", scanner, cache_policy=CachePolicy(ttl=60))
builder.add_edge(START, "collect_files")
builder.add_edge("collect_files", "pick_file")
builder.add_conditional_edges("pick_file", route_next)
builder.add_edge("scan_file", "pick_file")
return builder.compile(cache=InMemoryCache())
def run(graph, note: str) -> None:
"""Запускает обход и считает, сколько разборов пришло из кеша."""
CALLS.clear()
cached = 0
start = {"root": str(REPO), "queue": [], "current": "", "findings": []}
if note:
start["findings"] = [note]
for event in graph.stream(start, stream_mode="updates"):
if event.get("__metadata__", {}).get("cached"):
cached += 1
print(f" из кеша {cached}, выполнений тела узла {len(CALLS)}")
PASSES = (
("первый запуск", ""),
("тот же вход", ""),
("лишняя строка на входе", "начато вручную"),
)
narrow = build(scan_narrow)
wide = build(scan_wide)
print("1. узкая схема входа:")
for label, note in PASSES:
print(f" {label}:")
run(narrow, note)
print("2. на входе всё состояние:")
for label, note in PASSES:
print(f" {label}:")
run(wide, note)
# 3. Пометка о кеше приходит только в режиме updates.
print("3. где видна пометка:")
clean = {"root": str(REPO), "queue": [], "current": "", "findings": []}
events = list(narrow.stream(clean, stream_mode="updates"))
print(" updates, третье событие:", events[2])
print(" invoke, ключи итога:", sorted(narrow.invoke(clean)))
# Вывод:
# 1. узкая схема входа:
# первый запуск:
# из кеша 0, выполнений тела узла 4
# тот же вход:
# из кеша 4, выполнений тела узла 0
# лишняя строка на входе:
# из кеша 4, выполнений тела узла 0
# 2. на входе всё состояние:
# первый запуск:
# из кеша 0, выполнений тела узла 4
# тот же вход:
# из кеша 4, выполнений тела узла 0
# лишняя строка на входе:
# из кеша 0, выполнений тела узла 4
# 3. где видна пометка:
# updates, третье событие: {'scan_file': {'findings': ['app.py: строк 13']}, '__metadata__': {'cached': True}}
# invoke, ключи итога: ['current', 'findings', 'queue', 'root']
Разницу между пунктами 1 и 2 даёт аннотация аргумента: scan_narrow принимает ScanIn из двух полей, scan_wide всё состояние. Лишняя строка во входных находках разбора не касается, но широкому узлу она меняет ключ и заставляет считать заново.
Поэтому узлу, который вы собираетесь кешировать, дайте узкую схему входа. В уроке 2 аннотация аргумента state задавала, какие поля состояния получает узел. Здесь от неё зависит ещё и ключ кеша.
Пометка о попадании в кеш приходит только в режиме updates, ключом __metadata__ рядом с именем узла. В итоге invoke её нет.
ttl задаёт срок хранения записи в секундах, без него запись не истекает. Кеш в памяти процесса исчезает вместе с процессом.
Кеш не заменяет идемпотентность. При совпавшем ключе тело узла пропускается, и запись в файл тоже. При другом входе узел выполняется вместе со своими побочными действиями.
Практическое задание
Соберите обход репозитория с выходом по остатку шагов и записью отчёта по ключу.
Требования:
1) схема состояния на TypedDict: корень, очередь имён, текущий файл, находки с редьюсером, поле RemainingSteps, путь к отчёту
2) цикл из двух узлов: выбор следующего файла и его разбор, ребро назад из разбора в выбор
3) функция маршрутизации проверяет два условия: очередь пуста, значит END, остаток шагов меньше трёх, значит тоже END
4) узел разбора пишет строку в файл отчёта по ключу, чтобы повторный проход по тому же файлу не плодил копий
5) узел разбора принимает узкую схему входа и помечен CachePolicy с ttl, граф скомпилирован с InMemoryCache
Как проверить результат:
1) запуск по стенду mini_repo с пределом по умолчанию даёт четыре находки и четыре строки в отчёте
2) запуск с recursion_limit в 6 заканчивается без ошибки, отдаёт неполный список находок и непустую очередь
3) второй запуск того же графа по тому же стенду даёт пометку cached в режиме updates на каждом разборе
4) без CachePolicy запуск по очереди, где один файл стоит дважды, оставляет в отчёте одну строку на файл
5) без проверки остатка шагов пункт 2 падает с GraphRecursionError
Подсказка: заранее заполненную очередь можно передать во входном состоянии, отдельный узел сбора для задания не обязателен.
Итоги урока
Цикл это обычное ребро назад и условное ребро с выходом на END. Развилку ставьте в начале прохода, чтобы тело цикла не выполнялось при пустой очереди.
recursion_limit задаётся ключом верхнего уровня в конфигурации запуска. Когда запуск доходит до предела, он падает с GraphRecursionError, и invoke не возвращает собранное до ошибки. Запуску из десяти шагов нужен предел одиннадцать.
RemainingSteps в схеме состояния даёт остаток шагов до предела. Выход по нему возвращает неполный результат без ошибки.
Узел при повторе выполняется с начала вместе с побочными действиями, поэтому делайте их идемпотентными: запись по ключу, перезапись файла целиком, ключ идемпотентности для внешнего сервиса.
Кеш узла (CachePolicy и InMemoryCache) по умолчанию считает ключ от входа узла и при попадании пропускает тело вместе с побочными действиями. При другом входе тело выполняется заново, поэтому кеш не заменяет идемпотентность.
В уроке 5 разберу checkpointer: thread_id, историю запуска, продолжение через сутки и возврат к прошлому шагу.
Код урока
Примеры этого урока лежат в репозитории курса, папка lesson_04. Закреплённые версии, на которых получен вывод в тексте, лежат в requirements.txt в корне репозитория.
Предыдущий урок: Ветвление: условные рёбра, Command и Send
Подписывайтесь на мой Telegram канал
Если вам нужен ментор и вы хотите научиться разрабатывать AI агентов, пишите, обсудим условия
Авторизуйтесь, чтобы оставить комментарий.
Нет комментариев.
Тут может быть ваша реклама
Пишите info@aisferaic.ru
Похожие туториалы
