Структурные ограничения для ИИ-агентов: внутри конвейера ResolveFlow
Объясняет, как агент на основе LangGraph обеспечивает разделение процессов рассуждения и выполнения с помощью проверок на уровне кода вместо инструкций из промптов, включая ошибку поиска данных, возникшую на этом этапе.
Большинство демонстраций ИИ с агентным подходом следуют одной и той же основной схеме: модель принимает решение о действии и немедленно его выполняет. К промпту прикрепляется инструмент, модель его запускает, и инструмент работает без дополнительной проверки. В кратком демонстрационном ролике это может выглядеть убедительно, но именно такая конфигурация вызывает беспокойство у тех, кто задумывается о возможности использования автономных систем для взаимодействия с чем-либо значимым — поскольку зачастую единственной защитой от неверного диагноза или вредоносных изменений в рабочей системе является предупреждающее сообщение внутри промпта.
Проект, описанный здесь, был создан с целью избежания зависимости от одного-единственного предупреждающего указания.
ResolveFlow принимает ссылку на задачу в GitHub, собирает подтверждающие доказательства, относит задачу к определенной категории, а затем в зависимости от этой категории выполняет одно из трех действий: применяет фиксированное, неоспоримое решение, начинает расследование с использованием ИИ на основе собранных доказательств или направляет задачу непосредственно к человеку-рецензенту. Вместо однократного цикла, при котором модель запрашивает выполнение операции инструментом, система построена как автомат состояний LangGraph, основанный на одной центральной идее:
Мышление и выполнение действий разделены из-за структуры самой системы, а не из-за какого-либо установленного правила.
Это означает, что речь не просто о том, чтобы указать модели сначала свериться с кем-то. Вместо этого второй, независимый вызов LLM анализирует и оценивает диагноз первой модели ещё до того, как его увидит человек. А единственная функция во всем кодовом базе, которой разрешено записывать данные обратно в GitHub, проверяет наличие явного флага одобрения в рамках собственной логики — не потому, что граф вызовов предполагает такой маршрут, а потому что сама функция откажется выполняться без этого флага, даже если в будущем изменения кода создадут прямой путь, исключающий обычные этапы.
Остальная часть этого руководства посвящена тому, как на самом деле собирается система с использованием реальной реализации, а также ошибке, возникшей во время разработки. Эта ошибка служит полезным уроком: модель, указывающая на настоящий исходный документ, — это не то же самое, что модель, указывающая на источник, действительно релевантный поставленному вопросу.
Структура конвейера
Шесть этапов выполняются последовательно, и только одному из них разрешено изменять что-либо вне самого конвейера:
- fetch_evidence — фактические вызовы REST API GitHub для получения текста тикета, его комментариев и результатов проверок CI.
- normalize_evidence — полученный в результате этих вызовов необработанный JSON анализируется и преобразуется в объект типа IssueEvidence с указанием типов данных.
- classify — простой этап, основанный на правилах, при котором не происходит никаких вызовов LLM.
В следующих разделах рассматриваются наиболее важные аспекты.
Классификация: намеренно не является вызовом LLM
Наличие готового к использованию LLM побуждает принимать каждое решение с его участием, даже те, которые не требуют подобного рода анализа. Этап классификации определяет, в какой степени любая проблема может перейти к дорогостоящему и высокорисковому способу решения — диагностике на основе модели, запросам к источникам данных и последующим записям. Благодаря этой роли фильтра он должен быть недорогим, быстрым и полностью предсказуемым сам по себе:
def classify(state: GraphState) -> dict:
if state["evidence"].has_failing_ci:
return {"classification": "deterministic"}
elif state["evidence"].is_information_sparse:
return {"classification": "ai_investigation"}
else:
return {"classification": "human_review"}
Этот этап даёт три возможных результата, каждый из которых требует разного уровня доверия на последующих этапах обработки:
- детерминистичный — сбой проверки CI является однозначным, механическим сигналом. Нет необходимости в дополнительных исследованиях; проблема просто маркируется и передаётся администратору.
Стоит отметить, что в качестве альтернативы используется human_review, а не ai_investigation. Каждый раз, когда система не может понять ситуацию, она не пытается придумать умный ответ.
Диагностика: сначала поиск информации, затем анализ структуры, а не свободный текст
Как только проблема направляется в ветку ai_investigation, шаг generate_diagnosis ищет в индексе Pinecone примерно 2,750 фрагментов, взятых из реальных закрытых проблем из четырех разных репозиториев (facebook/react, langchain-ai/langchain, microsoft/terminal и vercel/next.js). Затем эти полученные фрагменты отправляются в LLM в качестве дополнительного контекста для обработки запроса:
def generate_diagnosis(state: GraphState) -> dict:
evidence = state["evidence"]
query = f"{evidence.title}\n\n{evidence.body}"
snippets = retrieve_evidence(query, k=3)
snippet_block = "\n\n".join(
f"[{s['id']}] (relevance: {s['score']:.2f}) {s['text']}" for s in snippets
)
prompt = (
f"Issue: {evidence.title}\n{evidence.body}\n\n"
f"Comments:\n{chr(10).join(evidence.comments) or '(none)'}\n\n"
f"Evidence snippets (cite by id in square brackets):\n{snippet_block}"
)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(Diagnosis)
diagnosis = structured_llm.invoke([("system", _SYSTEM_PROMPT), ("human", prompt)])
return {
"diagnosis": diagnosis,
"retrieved_ids": [s["id"] for s in snippets],
"retrieved_scores": {s["id"]: s["score"] for s in snippets},
}
Несколько деталей этого шага диагностики заслуживают более внимательного рассмотрения.
Во-первых, формат вывода не определяется задним числом — он гарантируется заранее. Диагностика описывается с помощью модели Pydantic:
class Diagnosis(BaseModel):
root_cause: str
severity: Literal["low", "medium", "high"]
missing_info: list[str] = Field(default_factory=list)
recommended_next_steps: list[str]
citations: list[str] = Field(
default_factory=list,
description="IDs of retrieved evidence/doc snippets that support each claim above",
)
При вызове .with_structured_output(Diagnosis) модель вынуждена формировать свой ответ именно в этой структуре. После этого не применяется никакой регулярный выражение для поиска корневой причины среди текста — когда вызов успешен, возвращается уже типизированный объект, а не текст, который нужно интерпретировать.
Во-вторых, цитаты не связаны с тоном или уверенностью модели — они должны быть реальными идентификаторами. В инструкциях системы четко указано, что любая цитата должна соответствовать одному из предоставленных ей идентификаторов фрагментов; ничего вымышленного, а также ничего, что не подтверждается содержанием фрагмента. Казалось бы, этого ограничения должно быть достаточно, но этого недостаточно — именно поэтому существует следующий этап обработки.
Независимый анализ: модель объясняет, код принимает решение
Это, пожалуй, самый важный архитектурный выбор во всей системе.
Узел independent_review инициирует второй, полностью отдельный вызов функции ChatOpenAI с собственным запросом и без общего контекста с тем вызовом, который дал диагноз. Его задача — оценить этот диагноз.
Однако вот ключевой момент: решение о подтверждении или передаче на более высокий уровень никогда не оставляется на усмотрение модели. Оно основывается на трех логических значениях, вычисленных с помощью обычного Python, а вывод ЯИ-модели сводится к человекочитаемым комментариям, от которых на самом деле не зависит логика утверждения.
def independent_review(state: GraphState) -> dict:
evidence = state["evidence"]
diagnosis = state["diagnosis"]
retrieved_ids = set(state.get("retrieved_ids", []))
retrieved_scores = state.get("retrieved_scores", {})
groundedness_ok = bool(diagnosis.citations) and all(
citation_id in retrieved_ids
and retrieved_scores.get(citation_id, 0.0) >= MIN_RELEVANCE_SCORE
for citation_id in diagnosis.citations
)
risk_ok = diagnosis.severity in _ALLOWED_SEVERITIES # {"low", "medium"}
permission_ok = True # comment/label are the only writes available today
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
reasoning = llm.invoke(
[("system", _SYSTEM_PROMPT), ("human", prompt)]
).content # human-readable critique — not what the gate checks
outcome = "approve" if (groundedness_ok and risk_ok and permission_ok) else "escalate_to_human"
return {"review_result": ReviewResult(
outcome=outcome,
groundedness_ok=groundedness_ok,
risk_ok=risk_ok,
permission_ok=permission_ok,
reasoning=reasoning,
)}
Это общий принцип, которому должна следовать любая надежная версия концепции «ЯИ-модель в роли судьи»: модель может объяснить свои действия, но решение принимает код.
Если вы попросите одну модель оценить работу другой модели, а затем просто поверите любому вердикту, указанному в её ответе, вы фактически создадите систему, надежность которой ограничивается тем самым фактором, который вы пытаетесь проверить.
В такой схеме вывод ЯИМ представляет собой полезное описание для человека-читателя, но тот механизм, который действительно имеет значение, нельзя склонить к негативному решению с помощью убедительных формулировок, поскольку он не анализирует то, что написала модель, для принятия решения.
Ещё одна деталь, на которую стоит обратить внимание: уровень высокой серьезности никогда не встречается в списке _ALLOWED_SEVERITIES. Любой диагноз с маркировкой высокой серьезности автоматически передается человеку, независимо от того, насколько достоверны приведенные им цитаты. Быть правильным и быть безопасным для автоматического одобрения — это совершенно разные вещи.
Ошибка: «обоснованный» не то же самое, что «релевантный»
Именно здесь на практике всё стало по-настоящему сложным.
Изначальная проверка groundedness_ok просто убеждалась в том, что идентификатор цитаты совпадает с чем-то в полученном наборе — реальным идентификатором, а не вымышленным. На бумаге это кажется разумной проверкой. Однако этого оказалось недостаточно.
Во время тестирования на небольшом корпусе из 40 проблем через всю систему была пропущена настоящая проблема React с пустым телом сообщения (facebook/react#36932, «experimental_taintUniqueValue бросает ошибку RangeError при обработке крупных двоичных значений»). В результате поиска были получены три фрагмента; все они были действительными и правильно идентифицированными — но ни один из них не имел отношения к этой конкретной ошибке. Несмотря на такие слабые данные, модель всё равно сформулировала уверенный и кажущийся точным диагноз, который оказался совершенно неверным: она утверждала наличие «проблемы совместимости расширения React DevTools». Все цитаты успешно прошли тест на обоснованность. Однако сам диагноз оставался бесполезным.
Решением стало прекращение использования фразы «был получен» в качестве замены для выражения «имеет отношение». Файл tools/retrieval.py был обновлен так, что теперь каждый фрагмент вместе со своим текстом возвращает также показатель косинусного сходства:
def retrieve_evidence(query: str, k: int = 3) -> list[dict]:
results = vector_store.similarity_search_with_score(query, k)
return [
{"id": doc.metadata["id"], "text": doc.page_content,
"source": doc.metadata["source"], "score": score}
for doc, score in results
]
Затем я проверил, каковы на самом деле показатели сходства по сравнению с актуальным индексом, вместо того чтобы просто предполагать значение порога. Запрос, полностью соответствующий реальному контенту в корпусе, получал оценку от 0,53 до 0,63. Запрос по теме, полностью отсутствующей в четырех обработанных репозиториях, давал оценку от 0,21 до 0,22. Исходя из этой разницы, в файле independent_review.py теперь установлен строгий минимум: каждый приведенный фрагмент должен превышать значение MIN_RELEVANCE_SCORE = 0,35. Этот порог намеренно выбран ближе к верхней границе разницы, чем к её средней точке, поскольку ошибка в повышении статуса хорошего диагноза стоит гораздо меньше, чем пропуск плохого диагноза как одобренного.
Помимо исправления логики оценки, необходимо было расширить сам корпус данных для поиска. Изначально он включал 40 статей из одного репозитория, что давало примерно 130 фрагментов — настолько маленьких, что у случайной реальной статьи часто не находилось подходящего по тематике фрагмента для извлечения. Позже объем был увеличен до примерно 600 статей из четырех реальных репозиториев, в результате чего получилось почти 2 750 фрагментов.
После создания более крупного корпуса данных та же проблема taintUniqueValue теперь отсекает два подлинных отчета с почти идентичным содержимым, позволяя выявить точную и конкретную причину: функция String.fromCharCode.apply превышает лимит количества аргументов, установленный движком JavaScript при обработке крупных буферов. Следует отметить, что механизм independent_review все равно передает этот случай на рассмотрение человека, поскольку его степень серьезности классифицируется как высокая, а находки с высокой степенью серьезности по замыслу всегда передаются на дальнейшую проверку, независимо от того, насколько уверенным или точным является диагноз. Точность не освобождает от прохождения проверки на наличие рисков.
Более общий урок заключается в том, что наличие цитаты, указывающей на реальный ID фрагмента данных, является необходимым, но не достаточным условием. Утверждение «модель что-то цитировала» — это не то же самое, что утверждение «модель цитировала что-то верное и релевантное для этой ошибки»; система, которая проверяет только первое утверждение, кажется тщательной, но на самом деле просто подтверждает бессмысленную информацию.
Шаг утверждения представляет собой реальную паузу, а не лишь визуальное состояние загрузки
Каждый описанный до сих пор этап лишь предлагает определенное действие. Ничего не записывается обратно в GitHub до определенного момента в работе pipeline. Все предыдущие этапы — классификация, диагностика, проверка — также лишь предлагают действия; ничего не записывается в GitHub до этого единственного шага.
Узел await_approval формирует точный комментарий (и, при необходимости, точную метку), который будет опубликован, используя идентичную логику независимо от того, была ли проблема направлена через детерминистическую ветвь или поступила из одобренной диагностики ai_investigation. Затем он вызывает метод interrupt() в LangGraph:
def await_approval(state: GraphState) -> dict:
proposed_action = _build_proposed_action(state)
approved = interrupt({
"classification": state["classification"],
"proposed_action": proposed_action,
})
return {"proposed_action": proposed_action, "approved": bool(approved)}
Вызов interrupt() приводит не только к отображению индикатора «ожидание одобрения», пока процесс находится в состоянии паузы, но и фактически останавливает выполнение графа в процессе его работы. Чтобы возобновить его работу позже, необходимо сделать совершенно отдельный запрос для поиска той же приостановленной нити и продолжения выполнения с того места, где оно было прервано, с использованием вызова вида result = await compiled_graph.ainvoke(Command(resume=True), config) с тем же thread_id, который использовался при первоначальном запуске. Для того чтобы это вообще сработало, состояние графа должно сохраняться между двумя независимыми HTTP-запросами, что исключает возможность использования обычной памяти внутри процесса для хранения данных.
Только после того, как это краткое описание готово, выполняется узел execute. Здесь стоит отметить два момента. Во-первых, проверка разрешений происходит внутри собственного кода узла, а не выражается исключительно через ребро графа — поэтому даже если в будущем при рефакторинге случайно появится прямое ускоряющее ребро к узлу execute, эта проверка всё равно его обнаружит. Во-вторых, узел execute отправляет значение state["proposed_action"] точно в том виде, в котором оно было одобрено; он никогда не генерирует комментарий заново. То, что утвердил человек, именно то, что публикуется.
def execute(state: GraphState) -> dict:
if not state.get("approved"):
raise PermissionError("execute() called without explicit approval")
evidence = state["evidence"]
action = state["proposed_action"]
token = state.get("github_token")
result = {
"comment": post_comment(evidence.repo, evidence.issue_number,
action["comment"], token=token)
}
if action.get("label"):
try:
result["label"] = add_label(evidence.repo, evidence.issue_number,
action["label"], token=token)
except requests.HTTPError as exc:
result["label_error"] = str(exc)
return {"execution_result": result}
Почему хранилище для приостановленных запусков имеет большее значение, чем кажется
Изначальная реализация опиралась на SqliteSaver, который сохраняет состояние в локальный файл на диске самого бэкенда. Такая конфигурация работает без проблем на компьютере разработчика.
В реальных условиях это приводит к проблемам, которые легко упустить из виду: на хостинге бесплатного тарифа, таком как Render, дисковое хранилище не сохраняется между перезагрузками. Процесс останавливается после периода бездействия, а при следующем запросе запускается заново в совершенно новом контейнере.
Вот пример того, как это происходит на практике. Запуск достигает точки await_approval и останавливается в ожидании действий человека. Ещё до того, как кто-либо нажмет «одобрить», бесплатная инстанция переходит в режим бездействия и останавливается. Следующий запрос запускает новый контейнер с полностью пустой базой данных. Когда выполняется команда Command(resume=...), уже нечего возобновлять — история остановленного потока исчезает без каких-либо ошибок или предупреждений.
Решением является AsyncPostgresSaver, работающий с реальной базой данных Postgres (в данной конфигурации — Neon), которая существует независимо от того, в каком контейнере выполняется приложение:
async with (
AsyncPostgresSaver.from_conn_string(DATABASE_URL, serde=get_serde()) as saver,
AsyncConnectionPool(DATABASE_URL, open=False,
check=AsyncConnectionPool.check_connection) as pool),
):
Даже если контейнер уничтожается и создается заново, все приостановленные потоки сохраняются, пока DATABASE_URL продолжает указывать на ту же базу данных. Механизмы приостановки и возобновления работы с помощью interrupt() совершенно не меняются — изменяется лишь место, где хранится состояние приостановки.
Стоит отметить отдельно: операции записи, выполняемые после одобрения, происходят от имени того, кто их одобрил, а не с использованием каких-либо общих учетных данных для развертывания. В рабочем приложении может войти любой пользователь GitHub, и каждая операция чтения или записи для данной сессии использует собственный токен OAuth этого пользователя. Запрос вида post_comment(evidence.repo, evidence.issue_number, action["comment"], token=token) использует токен одобрившего, а не токен того, кто случайно развернул приложение.
Это важно не только с точки зрения безопасности аутентификации. Это превращает утверждение «это опубликовал человек, который одобрил» в что-то, что сам GitHub может подтвердить, проверив автора комментария, вместо того чтобы полагаться на утверждения, исходящие от собственного интерфейса приложения.
Это также позволяет существующей системе разрешений GitHub обеспечивать эффективный контроль без дополнительного кода: пользователь, просто просматривающий репозиторий, который ему не принадлежит, может оставить комментарий, но для добавления метки требуется либо право на обработку запросов, либо право на запись в этом репозитории. execute.py грамотно справляется с этим — неудачная попытка добавить метку считается частичным успехом, вместо того чтобы приводить к сбою всего запроса.
Запросы к LLM и модели эмбеддингов по-прежнему выполняются с использованием собственных API-ключей владельца развертывания, независимо от того, кто их инициирует; именно поэтому существует ежедневный лимит на количество запросов в зависимости от пользователя, чтобы сдерживать расходы.
Существует важное различие, которое стоит четко обозначить: оценка регрессии и оценка функциональных возможностей проверяют совершенно разные аспекты, и ошибкой является оценивание их одинаково — в эту ошибку легко впасть. Порядок обработки маршрутов внутри функции classify(), логика контроля внутри функции independent_review() и проверка прав доступа внутри функции execute() каждая имеют ровно один правильный способ работы, и единственным приемлемым уровнем прохождения таких тестов является 100 процентов, точка. Безопасностный барьер, который будет обойден хотя бы один раз при любом количестве запусков, считается критической неисправностью — это не показатель, который можно скорректировать с помощью среднего значения.
Этот стандарт совершенно отличается от оценки того, дает ли функция generate_diagnosis качественные диагнозы. Такая оценка осуществляется с помощью модели, действительно улучшается со временем и никогда не могла бы реально достичь 100 процентов. Считать оба вида проверок эквивалентными — это распространенная ошибка: защитный механизм, который «в основном» работает, на самом деле не выполняет свою функции.
Что на самом деле сейчас происходит
Лучше преуменьшать ситуацию, чем преувеличивать её, поэтому вот таково реальное положение дел, а не какая-то презентация:
Сбор доказательств осуществляется с помощью реальных запросов GitHub REST, преобразуемых в типизированные объекты.
Классификация происходит на основе правил и является детерминированной, без участия нейросетей.
Диагностика получается в результате реального запроса к OpenAI, который генерирует структурированный вывод на основе фактического поиска в Pinecone среди примерно 2 750 фрагментов данных, которые загружаются заново еженедельно.
Независимая проверка выполняется с помощью отдельного запроса к OpenAI, но фактический механизм контроля — это код, а не собственное мнение модели.
Этап одобрения человеком представляет собой реальную паузу через interrupt(), возобновление работы с помощью Command(resume=...) и проверку каждого байта от начала до конца.
Фронтенд и бэкенд развернуты на Vercel и Render соответственно, работают полностью асинхронно и сохраняют состояние в Postgres.
GitHub OAuth позволяет любому посетителю войти с использованием собственного аккаунта; операции записи выполняются от имени этого пользователя с ежедневным лимитом.
Набор тестов включает регрессионные тесты на проверку соблюдения правил безопасности, которые r
Эти два недостатка не скрываются — они указаны в плане разработки, потому что действительно являются самыми важными задачами на следующем этапе, а не потому что их проигнорировали.
Уроки, которые стоит применить в собственном проекте
Разработка агента, предназначенного для действий в реальных системах, выявляет ряд принципов, которые выходят за рамки этого конкретного инструмента для решения проблем в GitHub:
Сохраняйте логику обработки и её выполнение в отдельных кодовых путях, а не просто в разных запросах. Инструкция вроде «всегда спрашивайте перед действием» по-прежнему остаётся лишь правилом поведения, и такие правила могут нарушаться именно в тот момент, когда они особенно нужны. Функция, которая категорически отказывается выполняться без установленного проверенного флага, представляет собой реальный барьер, а не просто предложение.
Когда вы говорите, что этап проверки «независим», убедитесь, что это действительно так: отдельный вызов, отсутствие общего следа выполнения, а вердикт формируется кодом, а не текстом, сгенерированным самой моделью. Модель может объяснить свои рассуждения, но она не должна сама оценивать их.
Не позволяйте фразе «модель указала на что-то реальное» заменять выражение «модель указала на что-то релевантное». Перед тем как установить порог сходства, действительно измерьте качество поиска по реальным запросам.
Если для вашей разработки важна пауза с участием человека, явно протестируйте, что происходит при возобновлении процесса до того, как человек отреагирует. Состояние в памяти и временные диски могут казаться нормальными в локальных тестах, но затем привести к сбоям именно там, где это самое опасное.
Наконец, четко указывайте, что действительно завершено, а что всё ещё является временной заменой. Таблица статусов, признающая свои недостатки, вызывает больше доверия, чем README-файл, который тихо намекает на то, что всё сделано.
Связанные материалы
- Agentic AI Explained: From Language Models to Autonomous Agents — структурированный обзор того, как LLM превращаются в агентные системы с помощью инструментов, памяти, планирования, архитектур многих агентов и интеграции MCP.