В этой статье я поделюсь сверхбыстрым руководством, которое покажет, как использовать графы для автоматизации знаний, заложенных в промпты, чтобы создать мощного чат-бота-агента для вашего бизнеса или личного пользования.
Сколько бы раз я ни спрашивал ChatGPT, я не могу получить хороший ответ - нужные мне предложения так и не появляются.
Кажется, что время просто уходит в никуда, и я знаю, что вы сталкивались с такой же проблемой. Дело не в том, что промпт плохо написан, а в том, что ИИ задают плохо структурированный вопрос.
У меня был неприятный опыт, когда я пытался задать вопрос ChatGPT и получал совсем не тот ответ, который ожидал.
Если вы уже долго работаете с ИИ-агентами, то в конечном итоге слышите о «цепочке рассуждений» (chain of thought). Это, как будто вы говорите большой языковой модели (LLM): «Погоди секунду, дай-ка я подумаю над этим шаг за шагом». Такая цепочка позволяет LLM генерировать пошаговый мыслительный процесс, раскрывая её способность решать сложные проблемы.
Однако этот метод полностью полагается на хранящиеся в модели данные и часто не даёт желаемых результатов из-за недостаточного запаса знаний при столкновении с узкоспециализированными задачами.
Но как насчёт генерации с дополненной выборкой (Retrieval-Augmented Generation, RAG)? Она может улучшить ответ. Да, вы правы - она даёт модели дополнительные знания, извлекая текст из внешних баз данных. Но получение большого количества простого текста не всегда помогает модели лучше думать или рассуждать.
Исследователи быстро поняли, что графовые структуры - это лучший инструмент для выражения знаний.
От «цепочки знаний» (chain-of-knowledge, CoK) до «рассуждений на графах» (reasoning on graphs, RoG) - эти методы пытаются извлекать пути или планы из графов знаний, чтобы направлять модели.
Их общий недостаток в том, что все они полагаются на полную и статичную базу данных.
В реальном мире знания экспертов в предметной области часто бывают несовершенными и даже могут содержать ошибки.
Что ещё серьёзнее, эти методы передают информацию только в одном направлении. Модель впитывает знания, не имея возможности передать обратно в базу знаний новые выводы, сделанные на основе данных.
В результате искажения в графе знаний со временем постоянно усиливаются и закрепляются.
Чтобы решить эту проблему, исследователи из Университета Джонса Хопкинса создали систему под названием EGO-Prompt. Идея состоит в том, чтобы рассматривать экспертные знания как то, что может меняться, а не как нечто застывшее. EGO-Prompt начинает с созданного экспертом причинно-следственного графа, а затем постоянно обновляет и промпты, и сам граф, используя реальные данные.
Позвольте мне быстро показать вам живого чат-бота, чтобы вы поняли, что я имею в виду.
На этот раз мы не будем задавать вопросы, а обучим подготовленный нами набор данных. Это будут 3 датасета (валидационный, тестовый и обучающий). Имейте в виду, я всегда стараюсь найти лучший и самый простой алгоритм, который мы можем использовать в собственном проекте.
Если вы посмотрите, как агент изначально формирует промпт, это объясняет задачу и генерирует Семантический причинно-следственный граф (Semantic Causal Graph, SCG), который показывает, как различные факторы могут влиять друг на друга. Он не обязательно должен быть идеальным, он лишь даёт системе отправную точку.
Когда в систему поступает новый случай (например, конкретный отчёт о диабете), EGO-Prompt не просто сваливает всю информацию в одну модель для обработки.
Вместо этого он использует двухэтапный рабочий процесс. Первая модель действует как аналитик. Она читает новые случаи и сравнивает их со всем огромным SCG, извлекая несколько причинно-следственных логических цепочек, которые наиболее релевантны для текущего случая.
Делегируйте часть рутинных задач вместе с BotHub! Для доступа к сервису не требуется VPN и можно использовать российскую карту. По ссылке вы можете получить 100 000 бесплатных токенов для первых задач и приступить к работе с нейросетями прямо сейчас!
Если в случае никогда не упоминается низкий уровень сахара в крови, то аналитик автоматически отфильтровывает все причинно-следственные цепочки, связанные с этим. Затем другая модель, которая принимает решение, берёт исходный случай и персонализированные рекомендации по рассуждению. На основе этих двух фрагментов информации выносится более обоснованное суждение.
После того как модель, принимающая решение делает прогноз, система сравнивает его с эталонной истиной. Если прогноз неверный, то активируется механизм текстового градиента. Вмешивается мощная модель-ментор, которая создаёт подробный отчёт об улучшениях, указывающий на потенциальные проблемы в исходных подсказках или в Семантическом причинно-следственном графе.
Используя эту обратную связь, агент автоматически выполняет операции по изменению на основе «Предложений по улучшению», где он может добавлять, удалять узлы и изменять описания или системные промпты в SCG, если они неточны.
Этот цикл повторяется снова и снова, и с каждым циклом промпт и Семантический причинно-следственный граф становятся чище и умнее.
Что делает EGO-Prompt уникальным?
SCG - это направленный ациклический граф (DAG). Узлы - это семантические блоки, извлечённые из входных данных задачи (например, «Диабет», «низкий уровень сахара в крови» и т. д.), а рёбра - это причинно-следственные семантические отношения, предоставленные экспертами, которые объясняют на естественном языке, «кто на кого влияет» и «как они влияют друг на друга».
Ключевым моментом является то, что этот первоначальный «логический набросок» (SCG) не обязательно должен быть полным или полностью правильным, поскольку он будет автоматически исправляться и развиваться алгоритмом. Он также не выполняет строгой причинно-следственной идентификации, поэтому не требует соблюдения каузальных марковских предположе��ий.
Это снижает требования к людям и позволяет системе потенциально начинать с несовершенного старта и в конечном итоге достигать высокого уровня точности.
Такой набросок предоставляет основу для первоначальных рассуждений ИИ, пока следующая его задача - проверять, исправлять и расширять эту карту на практике.
Как это работает?
Эта система похожа на процесс эволюции студента, учебника и ментора в одно целое:
LLM (Студент): Учится понимать подсказки и причинно-следственные диаграммы для лучшего рассуждения.
SCG (Учебник): Учебник, который допускает ошибки и может динамически пересматриваться.
Текстовый градиент (конспекты репетитора): Это движущая сила эволюции.
Когда студент (LLM) заканчивает свою работу (делает прогноз) и, сравнивает её со «стандартным ответом», обнаруживает ошибку, он не будет ломать голову в одиночку, а обратится к знающему (ментору), обычно это более мощная модель, такая как GPT-4o.
Вместо того чтобы генерировать ответ напрямую, репетитор напишет подробный «конспект» (то есть текстовый градиент), который укажет на причину ошибки студента.
«Конспекты» в EGO-Prompt состоят из двух частей: одна часть даёт обратную связь модели-студенту, указывая на недостатки в её подходе к решению проблемы или в промпте и предлагая, как их устранить, а другая часть обновляет учебник - SCG - исправляя, добавляя или удаляя причинно-следственные связи, чтобы исправить ошибки в графе знаний.
Давайте начнём кодить
Сначала импортируются зависимости и настраивается эксперимент - устанавливается пустой ключ API OpenAI (его нужно будет заполнить), импортируются утилиты и промпты, затем определяются параметры для запуска набора данных о пандемии через экспериментальную модель GPT-4o-mini, которая будет оцениваться моделью GPT-4o, с настройками для 1 итерации, датой 0919, 5 общими шагами, 1 эпохой и размером пакета 3.
Простыми словами, это настройка для итеративной оптимизации промптов и тестирования модели, где выводы более слабой модели оцениваются более сильной.
pythonimport os
os.environ['OPENAI_API_KEY']=""
from utils import *
from prompts import Prompts, TASK_LABLES, 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=3import os
os.environ['OPENAI_API_KEY']=""
from utils import *
from prompts import Prompts, TASK_LABLES, 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: модель для оценки (GPT-4o) и тестовая модель (GPT-4o-mini без кеширования). Оценочная модель устанавливается как обратный движок для вычисления градиентов или обратной связи для оптимизации.
Загружаются наборы данных для обучения, валидации и тестирования. Создаётся перемешиваемый загрузчик данных с указанным размером пакета для обучения. Затем все промпты в наборах данных оборачиваются специальными тегами (например, <TAG>промпт</TAG>), и, наконец, выводится количество примеров в каждом наборе.
pythoncm_labels = TASK_LABLES[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_set, val_set, test_set_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)
# Add Tag
col = "organized_prompt" if dataset_name == 'swiss' else "prompt"
train_set.data[col] = train_set.data[col].apply(lambda x: f"{tags[0]}{x}{tags[1]}")
val_set.data[col] = val_set.data[col].apply(lambda x: f"{tags[0]}{x}{tags[1]}")
test_set_ori.data[col] = test_set_ori.data[col].apply(lambda x: f"{tags[0]}{x}{tags[1]}")
print("Train/Val/Test Set Lengths: ", len(train_set), len(val_set), len(test_set_ori))cm_labels = TASK_LABLES[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_set, val_set, test_set_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)
# Add Tag
col = "organized_prompt" if dataset_name == 'swiss' else "prompt"
train_set.data[col] = train_set.data[col].apply(lambda x: f"{tags[0]}{x}{tags[1]}")
val_set.data[col] = val_set.data[col].apply(lambda x: f"{tags[0]}{x}{tags[1]}")
test_set_ori.data[col] = test_set_ori.data[col].apply(lambda x: f"{tags[0]}{x}{tags[1]}")
print("Train/Val/Test Set Lengths: ", len(train_set), len(val_set), len(test_set_ori))После этого создаётся копия тестового набора данных, чтобы не изменять оригинал. Вызывается функция init(), которая возвращает шесть объектов: системный промпт, причинно-следственный промпт, две модели (обычную и причинно-следственную) и два оптимизатора для каждой. Наконец, инициализируется пустой словарь для отслеживания результатов по ходу оптимизации.
pythonfrom copy import deepcopy
test_set = deepcopy(test_set_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": [], "prompt": [], "validation_f1": [], 'system_prompt':[], 'causal_prompt': []}from copy import deepcopy
test_set = deepcopy(test_set_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": [], "prompt": [], "validation_f1": [], 'system_prompt':[], 'causal_prompt': []}Далее определяется функция run_one_worker(), которая запускает несколько итераций оптимизации для одного рабочего потока. Она копирует все наборы данных, чтобы избежать взаимного влияния между потоками, отслеживает лучшие показатели F1 на валидации, добавляет временную метку и ID потока к тестовым промптам для обхода кеширования.
Затем она проходит через циклы итераций, где каждая итерация: создаёт уникальный путь для выходного JSON-файла, заново инициализирует модели, оптимизаторы и промпты, выполняет начальную оценку, запускает основной цикл обучения, который оптимизирует промпты, и обновляет лучшие результаты всякий раз, когда показатель F1 на валидации улучшается.
pythonimport time
import copy
import numpy as np
from concurrent.futures import ThreadPoolExecutor, as_completed
NUM_WORKERS = 1
def run_one_worker(worker_id: int):
local_test_set = copy.deepcopy(test_set)
local_test_set_ori = copy.deepcopy(test_set_ori)
local_val_set = copy.deepcopy(val_set)
local_train_loader = train_loader
val_performance = -float('inf')
test_performance = -float('inf')
final_results = None
all_val_f1s = []
all_test_f1s = []
local_test_set.data[col] = local_test_set_ori.data[col].apply(
lambda x: f"<!-- {time.time()} (w{worker_id}) -->, {x}"
)
for cur_iter in range(iteration):
print(f"[Worker {worker_id}] [Iteration {cur_iter+1}/{iteration}] begin")
output_json = (
f"res/{date}_{dataset_name}_{test_model.split('/')[-1].split(':')[-1]}_"
f"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, test_res, val_res = init_eval(
local_val_set, local_test_set, eval_fn, model, causal_model,
system_prompt, causal_prompt, cm_labels, iters=ITERS
)
results = run_training(
local_train_loader, local_val_set, local_test_set, 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 > val_performance:
val_performance = cur_val
test_performance = cur_test
final_results = results
print(f"[Worker {worker_id}] [Iteration {cur_iter+1}] "
f"val_best={val_performance:.4f}, test_at_best={test_performance:.4f}")
return {
'best_test_f1': test_performance,
'val_f1s': all_val_f1s,
'test_f1s': all_test_f1s,
'worker_id': worker_id,
}import time
import copy
import numpy as np
from concurrent.futures import ThreadPoolExecutor, as_completed
NUM_WORKERS = 1
def run_one_worker(worker_id: int):
local_test_set = copy.deepcopy(test_set)
local_test_set_ori = copy.deepcopy(test_set_ori)
local_val_set = copy.deepcopy(val_set)
local_train_loader = train_loader
val_performance = -float('inf')
test_performance = -float('inf')
final_results = None
all_val_f1s = []
all_test_f1s = []
local_test_set.data[col] = local_test_set_ori.data[col].apply(
lambda x: f"<!-- {time.time()} (w{worker_id}) -->, {x}"
)
for cur_iter in range(iteration):
print(f"[Worker {worker_id}] [Iteration {cur_iter+1}/{iteration}] begin")
output_json = (
f"res/{date}_{dataset_name}_{test_model.split('/')[-1].split(':')[-1]}_"
f"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, test_res, val_res = init_eval(
local_val_set, local_test_set, eval_fn, model, causal_model,
system_prompt, causal_prompt, cm_labels, iters=ITERS
)
results = run_training(
local_train_loader, local_val_set, local_test_set, 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 > val_performance:
val_performance = cur_val
test_performance = cur_test
final_results = results
print(f"[Worker {worker_id}] [Iteration {cur_iter+1}] "
f"val_best={val_performance:.4f}, test_at_best={test_performance:.4f}")
return {
'best_test_f1': test_performance,
'val_f1s': all_val_f1s,
'test_f1s': all_test_f1s,
'worker_id': worker_id,
}Наконец, запускаются 3 отдельных эксперимента одновременно с использованием нескольких потоков. Каждый из них пытается найти лучшие промпты, тестируя их. Когда каждый эксперимент завершается, он сохраняет лучший результат в список EGO_res и выводит, какой поток закончил работу. По сути, это параллельный запуск трёх разных случайных попыток, чтобы увидеть, какая из них найдёт лучший промпт.
pythonNUM_WORKERS = 3
ITERS = 1
total_steps=5
batch_size=3
EGO_res = []
with ThreadPoolExecutor(max_workers=NUM_WORKERS) as ex:
futures = [ex.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 test F1 per worker):", EGO_res)NUM_WORKERS = 3
ITERS = 1
total_steps=5
batch_size=3
EGO_res = []
with ThreadPoolExecutor(max_workers=NUM_WORKERS) as ex:
futures = [ex.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 test F1 per worker):", EGO_res)Заключение
Раньше ИИ учили как прилежного, но скучного зубрилу. Ему просто скармливали тонны информации и строгие правила, которые он должен был выучить наизусть. В итоге получался отличный имитатор, который мог блестяще пересказать заученное, но был беспомощен, столкнувшись с новой задачей. Он знал всё, но не понимал ничего.
EGO-Prompt - это совсем другой подход. Мы превращаем зубрилу в исследователя. Вместо того чтобы давать ему идеальные ответы, ему дают черновик знаний и разрешают ошибаться. А самое главное, его уча�� анализировать свои ошибки и исправлять их.
А теперь ваша очередь
Как вы думаете, что это значит для будущего?
Станет ли такой метод стандартом для создания умных ИИ, способных рассуждать, а не просто повторять заученное? Или это очередной эксперимент, который останется в стенах лабораторий?
И главный вопрос: готовы ли мы доверить ИИ, который учится и развивается самостоятельно, решение по-настоящему важных задач?
Делитесь своими мыслями в комментариях, спасибо за прочтение!