Temporal

Temporal is a durable execution platform that guarantees application code runs to completion despite failures. The platform provides reliability, scalability, and visibility for long-running workflows and distributed applications.

This guide covers routing telemetry from Temporal workflows to InteractiveAI for monitoring, debugging, and evaluating AI agents and LLM-powered applications.

Prerequisites

• InteractiveAI account with API credentials

• Temporal server (Cloud or local development server)

• OpenAI API key

---

Installation

pip install temporalio openai openai-agents interactiveai openinference-instrumentation-openai-agents

---

Configuration

Set environment variables for InteractiveAI, Temporal, and OpenAI:

import os

# InteractiveAI credentials
# Obtain keys from Settings > API Keys in the dashboard
os.environ["INTERACTIVEAI_PUBLIC_KEY"] = "pk-ia-..."
os.environ["INTERACTIVEAI_SECRET_KEY"] = "sk-ia-..."
os.environ["INTERACTIVEAI_HOST"] = "https://app.interactiveai.com"

# OpenAI credentials
os.environ["OPENAI_API_KEY"] = "sk-proj-..."

# Temporal server configuration
os.environ["TEMPORAL_HOST"] = "localhost:7233"
os.environ.setdefault("TEMPORAL_NAMESPACE", "default")
os.environ.setdefault("TEMPORAL_TASK_QUEUE", "agents-task-queue")

---

Enabling Trace Capture

Initialize the OpenAI Agents instrumentor to capture LLM operations:

from openinference.instrumentation.openai_agents import OpenAIAgentsInstrumentor

OpenAIAgentsInstrumentor().instrument()

Initialize the InteractiveAI client:

from interactiveai import Interactive
import os

client = Interactive(
    public_key=os.environ["INTERACTIVEAI_PUBLIC_KEY"],
    secret_key=os.environ["INTERACTIVEAI_SECRET_KEY"],
    host=os.environ.get("INTERACTIVEAI_HOST", "https://app.interactiveai.com")
)

if client.auth_check():
    print("Connection established")
else:
    print("Authentication failed - verify credentials")

---

Example: Building a Research Pipeline

The following sections walk through a complete implementation of a multi-step research pipeline. Temporal handles workflow orchestration and failure recovery, while OpenAI agents perform the actual research tasks. InteractiveAI captures telemetry from both layers.

The pipeline works in three stages:

1. Planning - An agent analyzes the research query and generates search terms

2. Searching - Multiple agents execute web searches in parallel and summarize findings

3. Writing - A final agent synthesizes all results into a comprehensive report

Each section below builds on the previous one, starting with data models, then agent definitions, workflow logic, and finally execution.

Step 1: Define Data Models

These Pydantic models define the structure for data passed between agents. The planner outputs search items, and the writer produces the final report.

from pydantic import BaseModel

class WebSearchItem(BaseModel):
    reason: str
    query: str

class WebSearchPlan(BaseModel):
    searches: list[WebSearchItem]

class ReportData(BaseModel):
    short_summary: str
    markdown_report: str
    follow_up_questions: list[str]

Step 2: Create Agent Factories

Each factory function returns a configured agent for a specific task. Separating these into functions allows the workflow to create fresh agent instances as needed.

from agents import Agent, WebSearchTool
from agents.model_settings import ModelSettings

def new_planner_agent():
    """Creates an agent that breaks down a query into search terms."""
    return Agent(
        name="PlannerAgent",
        instructions=(
            "You are a research assistant. Given a query, generate a set of web searches "
            "to gather information. Output between 5 and 20 search terms."
        ),
        model="gpt-4o",
        output_type=WebSearchPlan,
    )

def new_search_agent():
    """Creates an agent that executes searches and summarizes results."""
    return Agent(
        name="SearchAgent",
        instructions=(
            "You are a research assistant. Given a search term, search the web and "
            "produce a concise 2-3 paragraph summary under 300 words. Capture main points only."
        ),
        tools=[WebSearchTool()],
        model_settings=ModelSettings(tool_choice="required"),
    )

def new_writer_agent():
    """Creates an agent that synthesizes research into a final report."""
    return Agent(
        name="WriterAgent",
        instructions=(
            "You are a senior researcher writing a cohesive report. You will receive the original "
            "query and initial research. Create an outline, then generate a detailed markdown report "
            "of at least 1000 words."
        ),
        model="gpt-4o",
        output_type=ReportData,
    )

Step 3: Build the Research Manager

The ResearchManager class coordinates the three-stage pipeline. It creates agents, runs them in sequence, and handles parallel search execution.

import asyncio
from temporalio import workflow
from agents import RunConfig, Runner, custom_span, gen_trace_id, trace

class ResearchManager:
    def __init__(self):
        self.run_config = RunConfig()
        self.planner_agent = new_planner_agent()
        self.search_agent = new_search_agent()
        self.writer_agent = new_writer_agent()

    async def run(self, query: str) -> str:
        """Executes the full research pipeline for a given query."""
        trace_id = gen_trace_id()
        with trace("Research trace", trace_id=trace_id):
            search_plan = await self._plan_searches(query)
            search_results = await self._perform_searches(search_plan)
            report = await self._write_report(query, search_results)

        return report.markdown_report

    async def _plan_searches(self, query: str) -> WebSearchPlan:
        """Stage 1: Generate search terms from the query."""
        result = await Runner.run(
            self.planner_agent,
            f"Query: {query}",
            run_config=self.run_config,
        )
        return result.final_output_as(WebSearchPlan)

    async def _perform_searches(self, search_plan: WebSearchPlan) -> list[str]:
        """Stage 2: Execute all searches in parallel and collect summaries."""
        with custom_span("Search the web"):
            tasks = [
                asyncio.create_task(self._search(item)) 
                for item in search_plan.searches
            ]
            results = []
            for task in workflow.as_completed(tasks):
                result = await task
                if result is not None:
                    results.append(result)
            return results

    async def _search(self, item: WebSearchItem) -> str | None:
        """Executes a single search and returns the summary."""
        input_text = f"Search term: {item.query}\nReason: {item.reason}"
        try:
            result = await Runner.run(
                self.search_agent,
                input_text,
                run_config=self.run_config,
            )
            return str(result.final_output)
        except Exception:
            return None

    async def _write_report(self, query: str, search_results: list[str]) -> ReportData:
        """Stage 3: Synthesize all search results into a final report."""
        input_text = f"Original query: {query}\nSearch results: {search_results}"
        result = await Runner.run(
            self.writer_agent,
            input_text,
            run_config=self.run_config,
        )
        return result.final_output_as(ReportData)

Step 4: Define the Temporal Workflow

The workflow decorator registers ResearchWorkflow with Temporal. When triggered, it creates a ResearchManager and runs the pipeline.

from temporalio import workflow

@workflow.defn
class ResearchWorkflow:
    @workflow.run
    async def run(self, query: str) -> str:
        return await ResearchManager().run(query)

Step 5: Execute the Workflow

Before running, start a local Temporal development server:

temporal server start-dev

Then execute the workflow:

import os
from temporalio.client import Client
from temporalio.worker import Worker, UnsandboxedWorkflowRunner
from temporalio.contrib.openai_agents import OpenAIAgentsPlugin

async def main():
    tls = os.environ.get("TEMPORAL_TLS", "").lower() in ("1", "true", "yes")
    api_key = os.environ.get("TEMPORAL_API_KEY")

    plugin = OpenAIAgentsPlugin()

    # Connect to Temporal server
    client = await Client.connect(
        target_host=os.environ.get("TEMPORAL_ADDRESS", "localhost:7233"),
        namespace=os.environ.get("TEMPORAL_NAMESPACE", "default"),
        api_key=api_key or None,
        tls=tls,
        plugins=[plugin]
    )

    # Create worker to process workflows
    worker = Worker(
        client,
        task_queue=os.environ.get("TEMPORAL_TASK_QUEUE", "agents-task-queue"),
        workflows=[ResearchWorkflow],
        workflow_runner=UnsandboxedWorkflowRunner()
    )

    # Start workflow and wait for result
    async with worker:
        handle = await client.start_workflow(
            ResearchWorkflow.run,
            id="research-workflow-01",
            task_queue=os.environ.get("TEMPORAL_TASK_QUEUE", "agents-task-queue"),
            args=["Best practices for building reliable distributed systems"],
        )
        result = await handle.result()
        print("\nWorkflow result:\n", result)

await main()

---

Trace Visibility

The InteractiveAI dashboard displays:

• Workflow execution with timing and completion status

• Activity spans for each pipeline stage as nested traces

• LLM calls with prompts, completions, and token usage

• Cost estimates based on token consumption

• Latency breakdown across all components