Llama-Index Workflows¶

LlamaIndex Workflows are a framework for orchestrating and chaining together complex systems of steps and events, commonly used for building AI agents, research assistants, document processing pipelines and more.

TruLens supports tracing for LlamaIndex workflows, allowing you to understand exactly what happens in the agent execution flow.

Set Keys¶

In [ ]:

import os

os.environ["OPENAI_API_KEY"] = "sk-proj-..."

import os os.environ["OPENAI_API_KEY"] = "sk-proj-..."

Create a Workflow¶

In [ ]:

from workflows import Workflow, step
from workflows.events import (
    Event,
    StartEvent,
    StopEvent,
)
from llama_index.llms.openai import OpenAI

class JokeEvent(Event):
    joke: str

class JokeFlow(Workflow):
    llm = OpenAI()

    @step
    async def generate_joke(self, ev: StartEvent) -> JokeEvent:
        topic = ev.topic

        prompt = f"Write your best joke about {topic}."
        response = await self.llm.acomplete(prompt)
        return JokeEvent(joke=str(response))

    @step
    async def critique_joke(self, ev: JokeEvent) -> StopEvent:
        joke = ev.joke

        prompt = f"Give a thorough analysis and critique of the following joke: {joke}"
        response = await self.llm.acomplete(prompt)
        return StopEvent(result=str(response))

from workflows import Workflow, step from workflows.events import ( Event, StartEvent, StopEvent, ) from llama_index.llms.openai import OpenAI class JokeEvent(Event): joke: str class JokeFlow(Workflow): llm = OpenAI() @step async def generate_joke(self, ev: StartEvent) -> JokeEvent: topic = ev.topic prompt = f"Write your best joke about {topic}." response = await self.llm.acomplete(prompt) return JokeEvent(joke=str(response)) @step async def critique_joke(self, ev: JokeEvent) -> StopEvent: joke = ev.joke prompt = f"Give a thorough analysis and critique of the following joke: {joke}" response = await self.llm.acomplete(prompt) return StopEvent(result=str(response))

Start a TruLens session for logging¶

In [ ]:

from trulens.core.session import TruSession

session = TruSession()

from trulens.core.session import TruSession session = TruSession()

Register the workflow¶

In [ ]:

from trulens.apps.llamaindex import TruLlamaWorkflow

w = JokeFlow(timeout=60, verbose=False)

tru_workflow_recorder = TruLlamaWorkflow(
    w,
    app_name="Jokeflow",
    app_version="base",
    main_method=w.run,
)

from trulens.apps.llamaindex import TruLlamaWorkflow w = JokeFlow(timeout=60, verbose=False) tru_workflow_recorder = TruLlamaWorkflow( w, app_name="Jokeflow", app_version="base", main_method=w.run, )

Record workflow usage¶

In [ ]:

with tru_workflow_recorder as recording:
    result = await w.run(topic="pirates")
    print(str(result))

with tru_workflow_recorder as recording: result = await w.run(topic="pirates") print(str(result))

View in the dashboard¶

In [ ]:

from trulens.dashboard import run_dashboard

run_dashboard()

from trulens.dashboard import run_dashboard run_dashboard()