Multi-Agent

Multi-Agent for the Nora Observability Python SDK

Multi-Agent Tracking

Complex multi-agent systems can be tracked within trace groups.

Basic Multi-Agent Pipeline

import nora
from openai import OpenAI

nora_client = nora.init(api_key="your-nora-api-key", environment="multi-agent")
client = OpenAI(api_key="your-openai-key")

@nora_client.trace(span_type="agent", name="ResearchAgent")
def research_agent(topic):
    response = client.responses.create(
        model="gpt-5",
        input=[
            {
                "role": "system",
                "content": "You are a research assistant."
            },
            {
                "role": "user",
                "content": f"Research and summarize key points about: {topic}"
            }
        ],
        max_output_tokens=200
    )
    return response.output_text

@nora_client.trace(span_type="agent", name="WriterAgent")
def writer_agent(research_data):
    response = client.responses.create(
        model="gpt-5",
        input=[
            {
                "role": "system",
                "content": "You are a content writer."
            },
            {
                "role": "user",
                "content": f"Write an article based on this research:\n{research_data}"
            }
        ],
        max_output_tokens=300
    )
    return response.output_text

@nora_client.trace(span_type="agent", name="EditorAgent")
def editor_agent(draft):
    response = client.responses.create(
        model="gpt-5",
        input=[
            {
                "role": "system",
                "content": "You are an editor. Improve clarity and grammar."
            },
            {
                "role": "user",
                "content": f"Edit and improve this draft:\n{draft}"
            }
        ],
        max_output_tokens=300
    )
    return response.output_text

@nora_client.trace_group(name="ContentCreationPipeline")
def content_creation_pipeline(topic):
    print("Step 1: Research")
    research = research_agent(topic)

    print("Step 2: Writing")
    draft = writer_agent(research)

    print("Step 3: Editing")
    final_content = editor_agent(draft)

    return final_content

result = content_creation_pipeline("Artificial Intelligence in Healthcare")
print("\n=== Final Content ===")
print(result)

Trace Group with Child Function Tracking

Functions decorated with @nora_client.trace() automatically join the trace group when called inside it.

import nora
from openai import OpenAI

nora_client = nora.init(api_key="your-nora-api-key")
client = OpenAI(api_key="your-openai-key")

@nora_client.trace(span_type="retrieval")
def fetch_data(query):
    return f"Retrieved data about {query}"

@nora_client.trace(span_type="agent", name="ProcessorAgent")
def process_data(data):
    response = client.responses.create(
        model="gpt-5",
        input=[
            {
                "role": "user",
                "content": f"Analyze: {data}"
            }
        ],
        max_output_tokens=100
    )
    return response.output_text

@nora_client.trace(span_type="tool")
def format_report(text):
    return f"Final Report:\n{text}"

@nora_client.trace_group(name="MainWorkflow")
def main_workflow(query):
    data = fetch_data(query)
    processed = process_data(data)
    formatted = format_report(processed)

    return formatted

result = main_workflow("climate change")
print(result)

Functions with @trace are automatically included in the trace group. Outside a trace group, they are not tracked.

Parallel Agent Execution

import nora
from openai import OpenAI
import asyncio

nora_client = nora.init(api_key="your-nora-api-key")
client = OpenAI(api_key="your-openai-key")

@nora_client.trace(span_type="agent", name="Agent1")
async def agent_1(task):
    await asyncio.sleep(0.1)
    return f"Agent 1 completed: {task}"

@nora_client.trace(span_type="agent", name="Agent2")
async def agent_2(task):
    await asyncio.sleep(0.1)
    return f"Agent 2 completed: {task}"

@nora_client.trace(span_type="agent", name="Agent3")
async def agent_3(task):
    await asyncio.sleep(0.1)
    return f"Agent 3 completed: {task}"

@nora_client.trace_group(name="ParallelExecution")
async def parallel_agents():
    results = await asyncio.gather(
        agent_1("Task A"),
        agent_2("Task B"),
        agent_3("Task C")
    )

    return results

results = asyncio.run(parallel_agents())
print(results)