Griptape, часть 2: Строим графы

В прошлом посте я провел разбор базовых концептов AI-фреймворка Griptape, и сейчас самое время применить их к делу. Попробуем их использовать для разработки небольшого приложения, которое помогает вести линк-блог в телеграме.

Приложение будет получать URL, скачивать его, прогонять через LLM для генерации сокращенного содержания, переводить это содержимое еще на несколько языков, собирать все вместе и публиковать в телеграме через бота. Общий флоу можно увидеть на схеме ниже:

  flowchart LR
	A["URL"]
	A --> Parser["Парсер"] --> LLM1["Суммаризатор"] 
	LLM1 --> Translator1["Перевод на язык 1"]
	LLM1 --> Translator2["Перевод на язык 2"]
	Translator1 --> Combiner["Сборщик"]
	Translator2 --> Combiner
	LLM1 --> Combiner
	Combiner --> Telegram["Телеграм-бот"]

Для простоты картины я опущу имплементацию телеграм-бота, а также оставлю в покое мой любимый Human-in-the-loop, который, по моему мнению, обязательно должен присутствовать как минимум где-то в районе сборщика ¹.

В процессе мы попробуем разобраться, в каких случаях лучше использовать различные структуры, а также насколько получаемые графы composable и гибкие для изменения.

Ну что ж, приступим.

Создаем проект

Как упоминалось в предыдущем посте, Griptape — фреймворк для Python, а потому мы используем uv для старта проекта:

$ uv init .
$ uv add "griptape[all]>=1.7.2" python-dotenv

Экстра [all] устанавливает все доступные драйверы и загрузчики, что создает окружение размером в ~650МБ. В реальном приложении имеет смысл ограничиться только реально используемыми экстрами, список которых можно посмотреть в pyproject.toml проекта. В целом они разбиты довольно гранулярно, так что в большинстве случаев реальный размер будет значительно меньше.

Мы также подключаем python-dotenv, поскольку драйверы различных LLM-провайдеров используют переменные окружения для управления ключами. В нашем примере мы будем использовать openrouter.ai², предоставляющий OpenAI-подобное API для просто огромного количества моделей и провайдеров.

Соответственно, создадим файл .env и положим в него ключ:

OPENROUTER_API_KEY=sk-or-v1-...

Итак, сперва костяк нашего приложения:

# main.py
import argparse
import dotenv
import os

dotenv.load_dotenv()

def process_url(url: str):
    """Process the provided URL."""
    key = os.environ.get('OPENROUTER_API_KEY', '')
    if not key:
        raise ValueError("OPENROUTER_API_KEY is not set in the environment variables.")
    print(f"Processing URL: {url}")

def main():
    parser = argparse.ArgumentParser(description='Process URLs for link blog')
    parser.add_argument('url', type=str, help='URL to process')
    args = parser.parse_args()
    process_url(args.url)

if __name__ == "__main__":
    main()

Проверяем:

$uv run ./main.py google.com
Processing URL: google.com

Отлично. Можем описывать граф. Чтение документации ответило на один из моих вопросов, поднятых в прошлой статье вот так:

Griptape provides three Structures:
…
Of the three, Workflow is generally the most versatile. Agent and Pipeline can be handy in certain scenarios but are less frequently needed if you’re comfortable just orchestrating Tasks directly.

Отлично, как я и подозревал, остальные примитивы лучше подходят для совсем уж базовых задач, так что просто берем Workflow везде и приступаем к построению наших графов.

Грузим сайты бочками

Начнем мы с загрузки содержимого сайта. Для этого Griptape предоставляет драйвер WebScraper с несколькими различными реализациями и загрузчик WebLoader. Наша задача - обернуть это в Task:

from griptape.structures import Workflow
from griptape.tasks import CodeExecutionTask
from griptape.loaders import WebLoader

def load_page(task: CodeExecutionTask) -> str:
    """Load the content of the given URL."""
    print(f"Loading page: {task.input.value}")
    return WebLoader().load(task.input.value)


def print_result(task: CodeExecutionTask) -> None:
    """Print the result of the task."""
    for parent in task.parents:
        if parent.output.value:
            print(f"Output: {parent.output.value}")


def process_url(url: str):
    """Process the provided URL."""
    key = os.environ.get("OPENROUTER_API_KEY", "")
    if not key:
        raise ValueError("OPENROUTER_API_KEY is not set in the environment variables.")
    print(f"Processing URL: {url}")

    download_task = CodeExecutionTask(
        on_run=load_page, input=url, id="download_task", child_ids=["print_task"]
    )
    print_task = CodeExecutionTask(on_run=print_result, id="print_task")
    workflow = Workflow(tasks=[download_task, print_task])
    workflow.run()

Что мы здесь видим:

Несколько CodeExecutionTask. Это специальный тип задачи, который позволяет выполнить произвольный код. Такая задача принимает на вход функцию, в которую передается собственно сам объект этой задачи с огромным количеством полей, к которым можно получить доступ. В данном случае мы определили две задачи, одна из которых загружает содержимое сайта, а вторая – печатает содержимое своих задач-родителей.
Для определения самой структуры DAG используются параметры child_ids или parent_ids задач. Сам Workflow принимает просто список этих задач.
Workflow.run запускает задачи в графе и возвращает Workflow, из которого можно извлечь выполненные задачи и их входы/выходы. Кстати, Workflow можно запускать несколько раз, и при использовании Conversation Memory эта операция будет не идемпотентной.

Если воспользоваться StructureVisualizer, то мы увидим вот такую картинку:

  graph TD;
        Download_Task--> Print_Task;
        Print_Task;

Обобщаем

Для суммаризации у Griptape есть много готовых примитивов, включая TextSummaryTask. Она принимает на вход Summary Engine, через который можно настроить параметры суммаризации, такие как драйвер модели, шаблоны промтов и некий Chunker, назначение которого мы обсудим чуть позже. Попробуем этой задачей воспользоваться:

from griptape.tasks import TextSummaryTask
from griptape.engines import PromptSummaryEngine
from griptape.drivers.prompt.openai import OpenAiChatPromptDriver
...

def process_url(url: str):
	...
    download_task = CodeExecutionTask(
        on_run=load_page, input=url, id="download_task", child_ids=["summary_task"]
    )

    prompt_driver=OpenAiChatPromptDriver(
        model="google/gemini-2.5-flash-preview-05-20",
        base_url="https://openrouter.ai/api/v1",
        api_key=key,
    )
    
    summary_task = TextSummaryTask(
        "Please summarize the following content in a concise manner. {{ parents_output_text }}",
        summary_engine=PromptSummaryEngine(
            prompt_driver=prompt_driver
        ),
        id="summary_task",
        child_ids=["print_task"],
    )

    print_task = CodeExecutionTask(on_run=print_result, id="print_task")

    workflow = Workflow(tasks=[download_task, summary_task, print_task])

    workflow.run()

Этот код дает нам на выходе следующее:

$ uv run ./main.py https://docs.griptape.ai/stable/griptape-framework/structures/agents/
Output: Griptape Agents are a quick way to start using the platform. They take tools and input directly, which the agent uses to add a Prompt Task. The final output of the Agent can be accessed using the output attribute. An example demonstrates an Agent using a CalculatorTool to compute 13^7, successfully returning the result 62,748,517.

Здесь я наткнулся на интересную особенность: чтобы в summary_task на входе появился выход download_task, нужно обязательно указать инструкцию с использованием шаблона {parents_output_text} в качестве input. Хотя внутри суммаризатора уже есть необходимый промпт, сами данные из вывода родительских задач он не извлекает. Об этом приходится заботиться нам самим. В остальном все вроде бы достаточно очевидно.

  graph TD;
        Download_Task--> Summary_Task;
        Summary_Task--> Print_Task;
        Print_Task

Chunker

В то время как суммаризация отлично работала на небольших страницах, попытка запустить ее на странице из 55000 слов приводила к зависанию задачи независимо от используемой модели. Явное указание Chunker решило проблему:

from griptape.chunkers import TextChunker
...

	summary_task = TextSummaryTask(
        "Please summarize the following content in a concise manner. {{ parents_output_text }}",
        summary_engine=PromptSummaryEngine(
           	prompt_driver=prompt_driver,
            chunker = TextChunker(max_tokens=16000)
        ),
        id="summary_task",
        child_ids=["print_task", "russian_translate_task", "polish_translate_task"],
    )

При этом, судя по логам, чанкинг в суммаризаторе реализован итеративно, и на псевдокоде может быть выражен следующим образом:

chunks = [chunk1, chunk2, ..., chunkN]
summary = summarize(f"Summarize this: {chunk1}")
for chunk in chunks[1:]:
    summary = summarize(f"Update this summary {summary} with the following additional information: {chunk}")
return summary

Такой подход приводит к тому, что информация, расположенная ближе к концу текста, перевешивает ту, что в начале, что допустимо далеко не всегда. Лично я предпочел бы суммаризацию чанков через Map-Reduce:

chunks = [chunk1, chunk2, ..., chunkN]
summaries = [summarize(f"Summarize this: {chunk}") for chunk in chunks]
summary = summarize(f"Summarize these chunk summaries: {'\n---\n'.join(summaries)})
return summary

При таком подходе все чанки равны между собой, кроме того, это позволит выполнить суммаризацию параллельно, но при этом потребует на один запрос к модели больше.

Впрочем, здесь раскрывается гибкость фреймворка и движков. Благодаря им, этот алгоритм можно реализовать в своем собственном движке, унаследованном от BaseSummaryEngine и использовать практически бесшовно.

Вместе веселее

Следующий шаг – перевести этот текст на несколько языков. Разумеется, задачи эти параллелятся, и наш Workflow предоставляет замечательные инструменты для этого:

from griptape.tasks import PromptTask
...
def process_url(url: str):
    """Process the provided URL."""
	
    ...

    summary_task = TextSummaryTask(
		...
        
        id="summary_task",
        child_ids=["print_task", "russian_translate_task", "polish_translate_task"],
    )

    russian_translate_task = PromptTask(
        "Please translate the following text to Russian: {{ parents_output_text }}",
        prompt_driver=prompt_driver,
        id="russian_translate_task",
        child_ids=["print_task"],
    )

    polish_translate_task = PromptTask(
        "Please translate the following text to Polish: {{ parents_output_text }}",
        prompt_driver=prompt_driver,
        id="polish_translate_task",
        child_ids=["print_task"],
    )

    print_task = CodeExecutionTask(on_run=print_result, id="print_task")

    workflow = Workflow(
        tasks=[
            download_task,
            summary_task,
            russian_translate_task,
            polish_translate_task,
            print_task,
        ]
    )

Это было на удивление просто. PromptTask – наиболее базовый примитив, используемый для прямых запросов в LLM. Из интересного здесь только то, что мы указали в summary_task сразу несколько потомков, что позволяет нескольким задачам выполняться в параллель. Но как мы можем проверить, что задачи действительно выполняются параллельно? Разработчики предусмотрели и это, предоставив в API поддержку хуков. В частности, в конструкторе любой задачи есть параметры on_before_run и on_after_run, позволяющие добавить произвольную пре- и постобработку. Воспользуемся ими:

from griptape.tasks import BaseTask
from datetime import datetime
...

def timestamp(task: BaseTask, action: str):
    print(f"task {task.id} {action} at {datetime.now().isoformat()}")

def process_url(url: str):
	...
    polish_translate_task = PromptTask(
        "Please translate the following text to Polish: {{ parents_output_text }}",
        prompt_driver=prompt_driver,
        on_before_run=lambda task: timestamp(task, "started"),
        on_after_run=lambda task: timestamp(task, "finished"),
        id="polish_translate_task",
        child_ids=["print_task"],
    )    
    
    # И так же точно для остальных задач

Получаем следующий результат, полностью оправдывающий наши ожидания:

$ uv run ./main.py https://docs.griptape.ai/stable/griptape-framework/structures/agents/
task summary_task started at 2025-06-05T22:03:02.272397
task summary_task finished at 2025-06-05T22:03:04.345067
task russian_translate_task started at 2025-06-05T22:03:04.347638
task polish_translate_task started at 2025-06-05T22:03:04.351531
task polish_translate_task finished at 2025-06-05T22:03:05.702819
task russian_translate_task finished at 2025-06-05T22:03:05.947364
Output: Agents in Griptape offer a quick start, directly processing tools and input to generate a Prompt Task. The final output is accessible via the `output` attribute. An example demonstrates an Agent using a `CalculatorTool` to compute 13^7, showing the input, tool action, and the resulting output.
Output: Вот перевод текста на русский язык:

Агенты в Griptape предлагают быстрый старт, напрямую обрабатывая инструменты и входные данные для генерации задачи (Prompt Task). Конечный результат доступен через атрибут `output`. Пример демонстрирует Агента, использующего `CalculatorTool` для вычисления 13^7, показывая входные данные, действие инструмента и полученный результат.
Output: Oto tłumaczenie tekstu na język polski:

Agenci w Griptape oferują szybki start, bezpośrednio przetwarzając narzędzia i dane wejściowe w celu wygenerowania Zadania Monitu (Prompt Task). Ostateczny wynik jest dostępny poprzez atrybut `output`. Przykład demonstruje Agenta używającego `CalculatorTool` do obliczenia 13^7, pokazując dane wejściowe, działanie narzędzia i wynik końcowy.

Из логов ясно видно, что перевод на русский и польский работают параллельно друг другу.

И для порядка приведу текущий граф:

  graph TD;
        Download_Task--> Summary_Task;
        Summary_Task--> Print_Task & Russian_Translate_Task & Polish_Translate_Task;
        Russian_Translate_Task--> Print_Task;
        Polish_Translate_Task--> Print_Task;
        Print_Task;

Заканчиваем

Остальное, по сути, уже дело техники, поэтому я просто приведу ниже весь код программы:

import argparse
import dotenv
import os

from griptape.structures import Workflow
from griptape.tasks import CodeExecutionTask
from griptape.loaders import WebLoader
from griptape.utils import StructureVisualizer
from griptape.tasks import TextSummaryTask
from griptape.engines import PromptSummaryEngine
from griptape.drivers.prompt.openai import OpenAiChatPromptDriver
from griptape.tasks import PromptTask, BaseTask
from griptape.chunkers import TextChunker
from griptape.artifacts import TextArtifact
from datetime import datetime

import logging
logging.getLogger("griptape").setLevel(logging.WARNING)

dotenv.load_dotenv()


def load_page(task: CodeExecutionTask) -> str:
    """Load the content of the given URL."""
    return WebLoader().load(task.input.value)


def combine_result(task: CodeExecutionTask) -> TextArtifact:
    """Combine results from parent tasks."""
    result = ""
    for parent in task.parents:
        if parent.output.value:
            result += f"{parent.output.value}\n\n"
    return TextArtifact(result)

def send_to_telegram(task: CodeExecutionTask) -> None:
    """Send the result to Telegram."""
    # Placeholder for sending to Telegram logic
    print(f"Sending to Telegram: {task.parents[0].output.value}")

def timestamp(task: BaseTask, action: str):
    print(f"task {task.id} {action} at {datetime.now().isoformat()}")


def process_url(url: str):
    """Process the provided URL."""
    key = os.environ.get("OPENROUTER_API_KEY", "")
    if not key:
        raise ValueError("OPENROUTER_API_KEY is not set in the environment variables.")

    prompt_driver=OpenAiChatPromptDriver(
        model="google/gemini-2.5-flash-preview-05-20",
        base_url="https://openrouter.ai/api/v1",
        api_key=key,
    )

    download_task = CodeExecutionTask(
        on_run=load_page, input=url, id="download_task", child_ids=["summary_task"]
    )

    summary_task = TextSummaryTask(
        "Please summarize the following content in a concise manner. {{ parents_output_text }}",
        summary_engine=PromptSummaryEngine(
            prompt_driver=prompt_driver,
            chunker = TextChunker(max_tokens=16000)
        ),
        on_before_run=lambda task: timestamp(task, "started"),
        on_after_run=lambda task: timestamp(task, "finished"),
        id="summary_task",
        child_ids=["combine_task", "russian_translate_task", "polish_translate_task"],
    )

    russian_translate_task = PromptTask(
        "Please translate the following text to Russian: {{ parents_output_text }}",
        prompt_driver=prompt_driver,
        on_before_run=lambda task: timestamp(task, "started"),
        on_after_run=lambda task: timestamp(task, "finished"),
        id="russian_translate_task",
        child_ids=["combine_task"],
    )

    polish_translate_task = PromptTask(
        "Please translate the following text to Polish: {{ parents_output_text }}",
        prompt_driver=prompt_driver,
        on_before_run=lambda task: timestamp(task, "started"),
        on_after_run=lambda task: timestamp(task, "finished"),
        id="polish_translate_task",
        child_ids=["combine_task"],
    )

    combine_task = CodeExecutionTask(on_run=combine_result, id="combine_task", child_ids=["send_task"])
    send_task = CodeExecutionTask(
        on_run=send_to_telegram, id="send_task",
    )


    workflow = Workflow(
        tasks=[
            download_task,
            summary_task,
            russian_translate_task,
            polish_translate_task,
            combine_task,
            send_task
        ]
    )

    workflow.run()

    print(StructureVisualizer(workflow).to_url())
    print("Workflow completed successfully.")


def main():
    parser = argparse.ArgumentParser(description="Process URLs for link blog")
    parser.add_argument("url", type=str, help="URL to process")
    args = parser.parse_args()
    process_url(args.url)


if __name__ == "__main__":
    main()

И граф:

  graph TD;
        Download_Task--> Summary_Task;
        Summary_Task--> Combine_Task & Russian_Translate_Task & Polish_Translate_Task;
        Russian_Translate_Task--> Combine_Task;
        Polish_Translate_Task--> Combine_Task;
        Combine_Task--> Send_Task;
        Send_Task;

Я считаю, код выглядит довольно просто, легко читается и достаточно гибок для изменения и переиспользования. Очевидно, что в реальных задачах вся эта простота будет разбавлена обработкой ошибок, логированием и прочим. Но про это будет иметь смысл поговорить отдельно.

Дополнительно, Workflow API предоставляет еще два стиля композиции задач, не требующие указывать родителей и потомков при создании задач:

Императивный, при котором мы можем использовать функции add_parent и add_child
Так называемый bit-shift, при котором родители и потомки могут связываться так: task1 >> task2 >> [task3, task4]

Оба этих стиля позволяют с легкостью собирать разные графы из готовых примитивов, хотя bit-shift ощущается скорее как отдельный DSL, чем стандартный Python.

Закончили

В заключение хотелось бы сказать, что на данном этапе фреймворк мне нравится. Он достаточно логичен, гибок и приятен, хотя и не лишен некоторых шероховатостей.

Документация тоже мне показалась довольно неплохой, хотя ей и не хватает описания того, как именно работают некоторые движки. Пока чтобы раздобыть эту информацию, приходится шерстить логи или код.

Ну что ж, с базовой функциональностью мы разобрались, в следующий раз посмотрим, какие примитивы фреймворк предоставляет для построения RAG.

мы ведь хотим убедиться, что нам не будет мучительно стыдно за то, что мы опубликовали, верно? ↩︎
который я крайне уважаю и про который я надеюсь когда-нибудь написать отдельный пост ↩︎

Создаем проект#

Грузим сайты бочками#

Обобщаем#

Chunker#

Вместе веселее#

Заканчиваем#

Закончили#