В этой статье представлен экспресс-гид по использованию графовых структур для автоматизации знаний, встроенных в промпты, чтобы создать эффективного чат-бот-агента для бизнеса или личных нужд.
Сколько бы я ни обращался к ChatGPT, мне не удаётся получить качественные ответы — нужных формулировок всё нет.
Порой кажется, что время утекает впустую, и я уверен, что вы знакомы с этим ощущением. Проблема не в топорном промпте — вероятнее всего, запрашиваемая информация недостаточно упорядочена.
Лично я не раз сталкивался с тем, что задавал ChatGPT вопрос, а в ответ получал что-то совершенно иное.
Если вы уже работали с ИИ-агентами, то, скорее всего, слышали о «цепочке рассуждений» (chain of thought). Это как бы вы просите большую языковую модель: «Подождите, дайте мне проанализировать шаг за шагом». Такой подход позволяет LLM демонстрировать пошаговый ход мыслей при решении задач.
Однако он полностью опирается на уже заложенные в модель данные и нередко не справляется с узкоспециализированными темами.
Что насчёт RAG (Retrieval-Augmented Generation)? Да, она обогащает ответы за счёт извлечения информации из внешних хранилищ. Но простая навалом поданная фактура не всегда помогает модели рассуждать глубже.
Учёные обнаружили, что графы знаний — наиболее удобная форма представления взаимосвязей.
Методы от «цепочки знаний» (chain-of-knowledge, CoK) до «рассуждений на графах» (reasoning on graphs, RoG) извлекают из графа оптимальные маршруты или планы для управления моделью.
Но все они базируются на статичной и полной БД, тогда как в реальном мире экспертные знания часто неполны и порой ошибочны.
Ещё хуже, что такие методы передают информацию лишь в одну сторону: граф отдаёт, а модель не возвращает обратно ни новых выводов, ни корректив.
В итоге искажения в графе накапливаются и закрепляются.
Учёные из Университета Джонса Хопкинса предложили систему EGO-Prompt, которая рассматривает экспертные знания как динамичный объект, а не застывший эталон. EGO-Prompt начинает с экспертной причинно-следственной модели, а затем непрерывно обновляет и промпты, и сам граф на основе реальных случаев.
Давайте взглянем на работающего чат-бота, чтобы проиллюстрировать идею.
На вход подаётся заранее подготовленный набор данных, разделённый на обучающую, валидационную и тестовую части. Я всегда ищу самый простой и одновременно эффективный алгоритм для интеграции в свой проект.
При первом прогоне агент формирует системный промпт и семантический причинно-следственный граф (SCG), который показывает, как факторы взаимодействуют друг с другом. Этот черновой граф служит отправной точкой и не обязательно должен быть идеальным.
При появлении нового кейса (например, отчёта о диабете) система не смешивает всё в одну кучу.
Вместо этого задействуется двухступенчатый процесс. Первая модель выступает аналитиком: она сравнивает входные данные с SCG и извлекает несколько наиболее релевантных причинно-следственных цепочек.
Делегируйте рутину вместе с BotHub! Доступ без VPN, поддерживается оплата картами РФ. По ссылке — 100 000 бесплатных токенов на старте.
Если в кейсе не упоминается гипогликемия, аналитик исключит все цепочки, связанные с этим феноменом. Затем в дело вступает вторая модель — принимающая решение: она объединяет исходный кейс с подобранными рекомендациями и выносит обоснованный вердикт.
После прогноза система сверяет результат с эталоном. При расхождении активируется механизм «текстового градиента»: ментор, более мощная модель, создаёт отчёт с указанием ошибок в промптах или в SCG.
На основе обратной связи агент автоматически правит граф: добавляет, удаляет узлы или корректирует описания и системные промпты. Цикл повторяется, и SCG становится всё точнее и полнее.
В чём уникальность EGO-Prompt?
SCG представляет собой ориентированный ациклический граф (DAG). Узлы — семантические элементы задачи (например, «диабет», «низкий уровень глюкозы»), а рёбра — текстовые описания воздействий и влияний.
Важно, что начальный SCG может быть неполным и неточным — алгоритм постоянно исправит и доработает его, без необходимости жёстких каузальных допущений.
Это снижает нагрузку на экспертов и позволяет запускаться на «черновых» знаниях, постепенно достигая высокой точности.
Как это работает?
Эта система напоминает эволюцию «студента», «учебника» и «репетитора» в едином процессе:
-
LLM (студент): учится рассуждать по подсказкам и причинно-следственным диаграммам.
-
SCG (учебник): допускает ошибки и обновляется в реальном времени.
-
Текстовый градиент (конспект репетитора): постоянный движитель улучшений.
Когда студент ошибается, он обращается к репетитору (GPT-4o), который не выдаёт готовый ответ, а составляет «конспект» с анализом ошибки.
«Конспекты» разделяются на две части: обратная связь студенту и правки в SCG — добавление, удаление или корректировка связей.
Начинаем кодить
Сначала импортируем зависимости, задаём пустой ключ OpenAI (его нужно заполнить), импортируем утилиты и промпты, инициализируем параметры для эксперимента с набором «pandemic», где тестируется «gpt-4o-mini», а оценка проводится «gpt-4o».
import os
os.environ['OPENAI_API_KEY'] = ""
from utils import *
from prompts import Prompts, TASK_LABELS, TAGS
dataset_name = "pandemic"
test_model = "experimental:gpt-4o-mini"
eval_model = "gpt-4o"
iteration = 1
date = "0919"
total_steps = 5
epoch = 1
batch_size = 3Затем импортируем промпты и метки для выбранного набора, инициализируем движки LLM и настраиваем бэкенд для градиентного расчёта.
cm_labels = TASK_LABELS[dataset_name]
tags = TAGS[dataset_name]
CAUSAL_SYSTEM = Prompts[dataset_name]['CAUSAL_SYSTEM']
CAUSAL_SYSTEM_CONSTRAINT = Prompts[dataset_name]['CAUSAL_SYSTEM_CONSTRAINT']
SYSTEM = Prompts[dataset_name]['SYSTEM']
llm_api_eval = tg.get_engine(engine_name=eval_model)
llm_api_test = tg.get_engine(engine_name=test_model, cache=False)
tg.set_backward_engine(llm_api_eval, override=True)Загружаем train/val/test, оборачиваем промпты в теги и выводим размеры наборов.
train_set, val_set, test_ori, eval_fn = load_task(
dataset_name, evaluation_api=llm_api_eval, prompt_col="organized_prompt"
)
train_loader = tg.tasks.DataLoader(train_set, batch_size=batch_size, shuffle=True)
col = "organized_prompt" if dataset_name == 'swiss' else "prompt"
for ds in (train_set, val_set, test_ori):
ds.data[col] = ds.data[col].apply(lambda x: f"{tags[0]}{x}{tags[1]}")
print("Train/Val/Test lengths:", len(train_set), len(val_set), len(test_ori))Создаём копию тестового набора, вызываем init() для подготовки промптов, моделей и оптимизаторов, и инициализируем словарь для результатов.
from copy import deepcopy
test_set = deepcopy(test_ori)
system_prompt, causal_prompt, model, causal_model, optimizer, optimizer_causal = init(
SYSTEM, CAUSAL_SYSTEM, llm_api_test, llm_api_eval, CAUSAL_SYSTEM_CONSTRAINT
)
results = {"test_f1": [], "validation_f1": [], "prompt": [], "system_prompt": [], "causal_prompt": []}Функция run_one_worker() копирует датасеты, отслеживает лучшие F1, добавляет таймстемпы и запускает цикл итераций, оптимизирующий промпты и модели.
import time, copy, numpy as np
from concurrent.futures import ThreadPoolExecutor, as_completed
NUM_WORKERS = 1
def run_one_worker(worker_id: int):
local_test = copy.deepcopy(test_set)
local_ori = copy.deepcopy(test_ori)
local_val = copy.deepcopy(val_set)
local_loader = train_loader
best_val, best_test = -np.inf, -np.inf
all_val_f1s, all_test_f1s = [], []
local_test.data[col] = local_ori.data[col].apply(
lambda x: f", {x}"
)
for cur_iter in range(iteration):
print(f"[W{worker_id}] Iter {cur_iter+1}/{iteration} start")
output_json = f"res/{date}_{dataset_name}_{test_model.split(':')[-1]}_w{worker_id}_it{cur_iter+1}.json"
initialize_json_file(output_json)
system_prompt, causal_prompt, model, causal_model, optimizer, optimizer_causal = init(
SYSTEM, CAUSAL_SYSTEM, llm_api_test, llm_api_eval, CAUSAL_SYSTEM_CONSTRAINT
)
results, _, _ = init_eval(
local_val, local_test, eval_fn, model, causal_model,
system_prompt, causal_prompt, cm_labels, iters=ITERS
)
results = run_training(
local_loader, local_val, local_test, eval_fn,
model, causal_model, system_prompt, causal_prompt,
optimizer, optimizer_causal, results, cm_labels,
output_json=output_json, epoch=epoch, steps=total_steps, iters=ITERS
)
all_val_f1s.append(results['validation_f1'])
all_test_f1s.append(results['test_f1'])
cur_val = results['validation_f1'][-1]
cur_test = results['test_f1'][-1]
if cur_val > best_val:
best_val, best_test = cur_val, cur_test
print(f"[W{worker_id}] Best val={best_val:.4f}, test@best={best_test:.4f}")
return {'best_test_f1': best_test, 'val_f1s': all_val_f1s, 'test_f1s': all_test_f1s, 'worker_id': worker_id}Наконец, запускаем три параллельных потока, которые перебирают промпты и сохраняют лучшие F1 каждого.
NUM_WORKERS = 3
ITERS = 1
total_steps = 5
batch_size = 3
EGO_res = []
with ThreadPoolExecutor(max_workers=NUM_WORKERS) as executor:
futures = [executor.submit(run_one_worker, i) for i in range(NUM_WORKERS)]
for fut in as_completed(futures):
res = fut.result()
EGO_res.append(res['best_test_f1'])
print(f"[Main] Worker {res['worker_id']} done. Best test_f1={res['best_test_f1']:.4f}")
print("EGO_res (best F1 per worker):", EGO_res)Заключение
Раньше ИИ учили по жёстким правилам, словно зубрилу: много фактов и жёсткий регламент. В результате мы получали блестящего повторителя, но робота, не способного справиться с новым заданием.
EGO-Prompt переворачивает эту парадигму: мы предоставляем ИИ черновой граф знаний и разрешаем ошибаться, обучая его анализировать и исправлять собственные промахи.
А теперь ваша очередь
Как вы считаете, станет ли такой подход новым стандартом для создания рассуждающих ИИ, или останется экспериментом в лабораториях?
Готовы ли мы доверить ИИ, способному самостоятельно учиться и развиваться, решение действительно важных задач?
Делитесь мнениями в комментариях, спасибо за внимание!
