OpenAI Agents (tracing)

1
from respan import Respan, workflow, task
2
from respan_instrumentation_openai_agents import OpenAIAgentsInstrumentor
3
from agents import Agent, Runner, function_tool
4
5
respan = Respan(instrumentations=[OpenAIAgentsInstrumentor()])
6
7
@function_tool
8
def search_docs(query: str) -> str:
9
    """Search the documentation."""
10
    return f"Results for: {query}"
11
12
researcher = Agent(
13
    name="Researcher",
14
    instructions="You research topics using the search tool.",
15
    tools=[search_docs],
16
)
17
18
writer = Agent(
19
    name="Writer",
20
    instructions="You write concise summaries.",
21
)
22
23
@task(name="research")
24
async def research(topic: str) -> str:
25
    result = await Runner.run(researcher, f"Research: {topic}")
26
    return result.final_output
27
28
@workflow(name="research_and_write")
29
async def pipeline(topic: str):
30
    findings = await research(topic)
31
    result = await Runner.run(writer, f"Summarize: {findings}")
32
    print(result.final_output)
33
34
import asyncio
35
asyncio.run(pipeline("API gateways"))
36
respan.flush()

1
from agents import Agent, Runner, function_tool, trace
2
3
@function_tool
4
def get_weather(city: str) -> str:
5
    """Get the weather for a city."""
6
    return f"The weather in {city} is sunny, 72F"
7
8
agent = Agent(
9
    name="Weather Agent",
10
    instructions="Help users check the weather.",
11
    tools=[get_weather],
12
)
13
14
async def main():
15
    with trace("Weather check"):
16
        result = await Runner.run(agent, "What's the weather in San Francisco?")
17
        print(result.final_output)
18
    respan.flush()

1
from agents import Agent, Runner, trace
2
3
billing_agent = Agent(
4
    name="Billing Agent",
5
    instructions="Handle billing questions.",
6
)
7
8
support_agent = Agent(
9
    name="Support Agent",
10
    instructions="Route billing questions to the billing agent.",
11
    handoffs=[billing_agent],
12
)
13
14
async def main():
15
    with trace("Support handoff"):
16
        result = await Runner.run(support_agent, "I have a billing question")
17
        print(result.final_output)

1
from openai.types.responses import ResponseTextDeltaEvent
2
from agents import Agent, Runner
3
4
agent = Agent(name="Joker", instructions="You tell jokes.")
5
6
result = Runner.run_streamed(agent, input="Tell me 3 jokes.")
7
async for event in result.stream_events():
8
    if event.type == "raw_response_event" and isinstance(
9
        event.data, ResponseTextDeltaEvent
10
    ):
11
        print(event.data.delta, end="", flush=True)