Observability for OpenAI Agents with Opik

1
import os
2
import getpass
3
4
if "OPENAI_API_KEY" not in os.environ:
5
    os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

1
import os
2
from agents import Agent, Runner
3
from agents import set_trace_processors
4
from opik.integrations.openai.agents import OpikTracingProcessor
5
6
# Set project name for better organization
7
os.environ["OPIK_PROJECT_NAME"] = "openai-agents-demo"
8
9
set_trace_processors(processors=[OpikTracingProcessor()])
10
11
agent = Agent(name="Assistant", instructions="You are a helpful assistant")
12
13
result = Runner.run_sync(agent, "Write a haiku about recursion in programming.")
14
print(result.final_output)

1
from agents import Agent, Runner, function_tool, set_trace_processors
2
from opik.integrations.openai.agents import OpikTracingProcessor
3
4
set_trace_processors(processors=[OpikTracingProcessor()])
5
6
@function_tool
7
def calculate_average(numbers: list[float]) -> float:
8
    return sum(numbers) / len(numbers)
9
10
@function_tool  
11
def get_recommendation(topic: str, user_level: str) -> str:
12
    recommendations = {
13
        "python": {
14
            "beginner": "Start with Python.org's tutorial, then try Python Crash Course book. Practice with simple scripts and built-in functions.",
15
            "intermediate": "Explore frameworks like Flask/Django, learn about decorators, context managers, and dive into Python's data structures.",
16
            "advanced": "Study Python internals, contribute to open source, learn about metaclasses, and explore performance optimization."
17
        },
18
        "machine learning": {
19
            "beginner": "Start with Andrew Ng's Coursera course, learn basic statistics, and try scikit-learn with simple datasets.",
20
            "intermediate": "Dive into deep learning with TensorFlow/PyTorch, study different algorithms, and work on real projects.",
21
            "advanced": "Research latest papers, implement algorithms from scratch, and contribute to ML frameworks."
22
        }
23
    }
24
    
25
    topic_lower = topic.lower()
26
    level_lower = user_level.lower()
27
    
28
    if topic_lower in recommendations and level_lower in recommendations[topic_lower]:
29
        return recommendations[topic_lower][level_lower]
30
    else:
31
        return f"For {topic} at {user_level} level: Focus on fundamentals, practice regularly, and build projects to apply your knowledge."
32
33
def create_advanced_agent():
34
    """Create an advanced agent with tools and comprehensive instructions."""
35
    instructions = """
36
    You are an expert programming tutor and learning advisor. You have access to tools that help you:
37
    1. Calculate averages for performance metrics, grades, or other numerical data
38
    2. Provide personalized learning recommendations based on topics and user experience levels
39
    
40
    Your role:
41
    - Help users learn programming concepts effectively
42
    - Provide clear, beginner-friendly explanations when needed
43
    - Use your tools when appropriate to give concrete help
44
    - Offer structured learning paths and resources
45
    - Be encouraging and supportive
46
    
47
    When users ask about:
48
    - Programming languages: Use get_recommendation to provide tailored advice
49
    - Performance or scores: Use calculate_average if numbers are involved
50
    - Learning paths: Combine your knowledge with tool-based recommendations
51
    
52
    Always explain your reasoning and make your responses educational.
53
    """
54
    
55
    return Agent(
56
        name="AdvancedProgrammingTutor",
57
        instructions=instructions,
58
        model="gpt-4o-mini",
59
        tools=[calculate_average, get_recommendation]
60
    )
61
62
# Create and use the advanced agent
63
advanced_agent = create_advanced_agent()
64
65
# Example queries
66
queries = [
67
    "I'm new to Python programming. Can you tell me about it?",
68
    "I got these test scores: 85, 92, 78, 96, 88. What's my average and how am I doing?",
69
    "I know some Python basics but want to learn machine learning. What should I do next?",
70
]
71
72
for i, query in enumerate(queries, 1):
73
    print(f"\n📝 Query {i}: {query}")
74
    result = Runner.run_sync(advanced_agent, query)
75
    print(f"🤖 Response: {result.final_output}")
76
    print("=" * 80)

1
from agents import Agent, Runner, function_tool, set_trace_processors
2
from opik.integrations.openai.agents import OpikTracingProcessor
3
from opik import track
4
5
set_trace_processors(processors=[OpikTracingProcessor()])
6
7
@track(name="fetch_user_data")
8
def fetch_user_data(user_id: str) -> dict:
9
    # This step will be traced separately
10
    return {"user_id": user_id, "preferences": ["python", "ml"]}
11
12
@track(name="generate_recommendations")
13
def generate_recommendations(preferences: list) -> str:
14
    # This step will also be traced separately
15
    return f"Based on your interests in {', '.join(preferences)}, we recommend..."
16
17
@function_tool
18
def get_personalized_advice(user_id: str) -> str:
19
    """Get personalized learning advice for a user."""
20
    # Each tracked function call inside the tool will appear as a separate span
21
    user_data = fetch_user_data(user_id)
22
    recommendations = generate_recommendations(user_data["preferences"])
23
    return recommendations
24
25
agent = Agent(
26
    name="PersonalizedTutor",
27
    instructions="Help users with personalized learning advice.",
28
    model="gpt-4o-mini",
29
    tools=[get_personalized_advice]
30
)
31
32
result = Runner.run_sync(agent, "Give me learning advice for user_123")
33
print(result.final_output)

1
async def main():
2
    agent = Agent(name="Assistant", instructions="Reply very concisely.")
3
4
    thread_id = str(uuid.uuid4())
5
6
    with trace(workflow_name="Conversation", group_id=thread_id):
7
        # First turn
8
        result = await Runner.run(agent, "What city is the Golden Gate Bridge in?")
9
        print(result.final_output)
10
        # San Francisco
11
12
        # Second turn
13
        new_input = result.to_input_list() + [{"role": "user", "content": "What state is it in?"}]
14
        result = await Runner.run(agent, new_input)
15
        print(result.final_output)
16
        # California