Griptape, część 2: Budujemy grafy

W poprzednim wpisie omówiłem podstawowe koncepcje frameworka AI Griptape, a teraz nadszedł czas, aby zastosować je w praktyce. Spróbujmy ich użyć do stworzenia małej aplikacji, która pomaga w prowadzeniu link-bloga na Telegramie.

Aplikacja będzie otrzymywać adres URL, pobierać jego zawartość, przepuszczać przez LLM w celu wygenerowania skróconego podsumowania, tłumaczyć to podsumowanie na kilka innych języków, składać wszystko w całość i publikować na Telegramie za pomocą bota. Ogólny przepływ pracy można zobaczyć na poniższym schemacie:

  flowchart LR
    A["URL"]
    A --> Parser["Parser"] --> LLM1["Sumaryzator"] 
    LLM1 --> Translator1["Tłumaczenie na język 1"]
    LLM1 --> Translator2["Tłumaczenie na język 2"]
    Translator1 --> Combiner["Kombinator"]
    Translator2 --> Combiner
    LLM1 --> Combiner
    Combiner --> Telegram["Bot telegramowy"]

Dla uproszczenia pominę implementację bota telegramowego, a także zostawię w spokoju mój ulubiony Human-in-the-loop, który moim zdaniem obowiązkowo powinien znaleźć się przynajmniej gdzieś w okolicach kombinatora ¹.

W trakcie pracy postaramy się zrozumieć, w jakich przypadkach lepiej używać różnych struktur, a także na ile uzyskane grafy są komponowalne i elastyczne na zmiany.

No cóż, zaczynajmy.

Tworzymy projekt

Jak wspomniałem w poprzednim poście, Griptape to framework dla Pythona, dlatego użyjemy uv do rozpoczęcia projektu:

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

Dodatek [all] instaluje wszystkie dostępne sterowniki i loadery, co tworzy środowisko o rozmiarze ~650MB. W rzeczywistej aplikacji warto ograniczyć się tylko do rzeczywiście używanych dodatków, których listę można znaleźć w pliku pyproject.toml projektu. Ogólnie są one podzielone dość granularnie, więc w większości przypadków rzeczywisty rozmiar będzie znacznie mniejszy.

Podłączamy również python-dotenv, ponieważ sterowniki różnych dostawców LLM używają zmiennych środowiskowych do zarządzania kluczami. W naszym przykładzie będziemy używać openrouter.ai², który dostarcza API podobne do OpenAI dla ogromnej liczby modeli i dostawców.

W związku z tym, utwórzmy plik .env i umieśćmy w nim klucz:

OPENROUTER_API_KEY=sk-or-v1-...

A teraz szkielet naszej aplikacji:

# 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()

Sprawdzamy:

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

Świetnie. Możemy opisywać graf. Lektura dokumentacji odpowiedziała na jedno z moich pytań, postawionych w poprzednim artykule, w następujący sposób:

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.

Świetnie, tak jak podejrzewałem, pozostałe prymitywy lepiej nadają się do zupełnie podstawowych zadań, więc po prostu bierzemy Workflow wszędzie i przystępujemy do budowania naszych grafów.

Ładujemy strony hurtowo

Zaczniemy od załadowania zawartości strony. Do tego Griptape dostarcza sterownik WebScraper z kilkoma różnymi implementacjami oraz loader WebLoader. Naszym zadaniem jest opakowanie tego w 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()

Co tu widzimy:

Kilka CodeExecutionTask. Jest to specjalny typ zadania, który pozwala na wykonanie dowolnego kodu. Takie zadanie przyjmuje na wejściu funkcję, do której przekazywany jest sam obiekt tego zadania z ogromną liczbą pól, do których można uzyskać dostęp. W tym przypadku zdefiniowaliśmy dwa zadania, z których jedno pobiera zawartość strony, a drugie – drukuje zawartość swoich zadań-rodziców.

Do zdefiniowania samej struktury DAG używane są parametry child_ids lub parent_ids zadań. Sam Workflow przyjmuje po prostu listę tych zadań.
Workflow.run uruchamia zadania w grafie i zwraca Workflow, z którego można wyodrębnić wykonane zadania oraz ich wejścia/wyjścia. Nawiasem mówiąc, Workflow można uruchamiać wielokrotnie, a przy użyciu Conversation Memory ta operacja nie będzie idempotentna.

Jeśli skorzystamy z StructureVisualizer, zobaczymy taki oto obrazek:

  graph TD;
        Download_Task--> Print_Task;
        Print_Task;

Podsumowujemy

Do podsumowywania Griptape ma wiele gotowych prymitywów, w tym TextSummaryTask. Przyjmuje ona na wejściu Summary Engine, za pomocą którego można skonfigurować parametry podsumowania, takie jak sterownik modelu, szablony promptów i pewien Chunker, którego przeznaczenie omówimy nieco później. Spróbujmy użyć tego zadania:

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()

Ten kod daje nam na wyjściu następujący wynik:

$ 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.

Tutaj natknąłem się na ciekawą osobliwość: aby na wejściu summary_task pojawiło się wyjście download_task, trzeba koniecznie podać instrukcję z użyciem szablonu {parents_output_text} jako input. Chociaż wewnątrz sumaryzatora jest już potrzebny prompt, sam danych z wyjścia zadań-rodziców nie pobiera. O to musimy zadbać sami. Poza tym wszystko wydaje się dość oczywiste.

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

Chunker

Podczas gdy podsumowywanie działało świetnie na małych stronach, próba uruchomienia go na stronie o objętości 55 000 słów prowadziła do zawieszenia się zadania, niezależnie od używanego modelu. Jawne wskazanie Chunker rozwiązało problem:

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"],
    )

Jednocześnie, sądząc po logach, chunking w sumaryzatorze jest zaimplementowany iteracyjnie i w pseudokodzie można go wyrazić w następujący sposób:

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

Takie podejście sprawia, że informacje znajdujące się bliżej końca tekstu mają większą wagę niż te na początku, co nie zawsze jest dopuszczalne. Osobiście wolałbym podsumowywanie chunków za pomocą 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

Przy takim podejściu wszystkie chunki są sobie równe, a dodatkowo pozwoli to na równoległe wykonanie podsumowania, ale będzie wymagało o jedno zapytanie do modelu więcej.

Jednakże, tutaj ujawnia się elastyczność frameworka i silników. Dzięki nim, ten algorytm można zaimplementować we własnym silniku, dziedziczącym po BaseSummaryEngine i używać go niemal bezproblemowo.

W grupie raźniej

Następny krok – przetłumaczyć ten tekst na kilka języków. Oczywiście zadania te można zrównoleglić, a nasz Workflow dostarcza do tego doskonałych narzędzi:

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,
        ]
    )

To było zaskakująco proste. PromptTask to najbardziej podstawowy prymityw, używany do bezpośrednich zapytań do LLM. Ciekawostką jest to, że w summary_task wskazaliśmy od razu kilku potomków, co pozwala na równoległe wykonywanie kilku zadań. Ale jak możemy sprawdzić, czy zadania rzeczywiście wykonują się równolegle? Twórcy przewidzieli i to, udostępniając w API wsparcie dla hooków. W szczególności, w konstruktorze każdego zadania znajdują się parametry on_before_run i on_after_run, pozwalające na dodanie dowolnego pre- i post-processingu. Skorzystajmy z nich:

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"],
    )    
    
    # I tak samo dla pozostałych zadań

Otrzymujemy następujący wynik, w pełni spełniający nasze oczekiwania:

$ 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.

Z logów jasno widać, że tłumaczenie na rosyjski i polski działają równolegle.

I dla porządku, przedstawiam aktualny graf:

  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;

Finalizujemy

Reszta to w zasadzie kwestia techniki, dlatego poniżej po prostu zamieszczę cały kod programu:

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()Ładujemy strony hurtowo

    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()

I graf:

  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;

Uważam, że kod wygląda dość prosto, jest łatwy do czytania i wystarczająco elastyczny do modyfikacji i ponownego wykorzystania. Oczywiście, w rzeczywistych zadaniach cała ta prostota zostanie rozcieńczona obsługą błędów, logowaniem i innymi elementami. Ale o tym warto będzie porozmawiać osobno.

Dodatkowo, Workflow API oferuje jeszcze dwa style kompozycji zadań, które nie wymagają podawania rodziców i potomków przy tworzeniu zadań:

Imperatywny, w którym możemy używać funkcji add_parent i add_child.
Tak zwany bit-shift, w którym rodzice i potomkowie mogą być łączeni w następujący sposób: task1 >> task2 >> [task3, task4].

Oba te style pozwalają z łatwością budować różne grafy z gotowych prymitywów, chociaż bit-shift bardziej przypomina osobny DSL niż standardowy Python.

Skończyliśmy

Podsumowując, chciałbym powiedzieć, że na tym etapie framework mi się podoba. Jest dość logiczny, elastyczny i przyjemny w użyciu, chociaż nie jest pozbawiony pewnych niedociągnięć.

Dokumentacja również wydała mi się całkiem niezła, chociaż brakuje jej opisu, jak dokładnie działają niektóre silniki. Na razie, aby zdobyć te informacje, trzeba przeglądać logi lub kod.

Cóż, z podstawową funkcjonalnością się zapoznaliśmy, następnym razem zobaczymy, jakie prymitywy framework dostarcza do budowy RAG.

w końcu chcemy mieć pewność, że nie będziemy się wstydzić tego, co opublikowaliśmy, prawda? ↩︎
który bardzo szanuję i o którym mam nadzieję kiedyś napisać osobny post. ↩︎

Tworzymy projekt#

Ładujemy strony hurtowo#

Podsumowujemy#

Chunker#

W grupie raźniej#

Finalizujemy#

Skończyliśmy#