Spaces:
Sleeping
Sleeping
File size: 6,823 Bytes
2c060c5 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 |
"""
Web Search Platform Research Test
Tutorial: Research observability platforms and convert findings into writing tasks.
"""
import os
import sys
import asyncio
import json
from pathlib import Path
from datetime import datetime
from dotenv import load_dotenv
from phoenix.otel import register
import weave
from agents import Runner, Agent
sys.path.insert(0, str(Path(__file__).parent.parent))
from agent.todo_agent import create_agent
from agent.storage import JsonTodoStorage
def reset_test_data():
"""Reset todos and session data for clean test runs."""
os.makedirs("data", exist_ok=True)
with open("data/todos.json", "w") as f:
json.dump([], f)
with open("data/session_default.json", "w") as f:
json.dump({"history": []}, f)
print("🔄 Data reset - starting with clean slate")
def initialize_tracing(project_name: str):
"""Initialize tracing with graceful error handling."""
os.environ["OPENAI_TRACING_ENABLED"] = "1"
os.environ["WEAVE_PRINT_CALL_LINK"] = "false"
# Phoenix: Add minimal custom resource attributes via environment variable
os.environ["OTEL_RESOURCE_ATTRIBUTES"] = f"tutorial.name={project_name},tutorial.type=web_search,environment=test,app.name=todo-agent"
try:
register(project_name=project_name, auto_instrument=True)
print(f"✅ Phoenix tracing initialized for: {project_name}")
except Exception as e:
print(f"⚠️ Phoenix tracing failed: {e}")
if not weave.get_client():
try:
weave.init(project_name)
print(f"✅ Weave tracing initialized for: {project_name}")
except Exception as e:
print(f"⚠️ Weave tracing failed (continuing without Weave): {e}")
async def run_web_search_test():
"""Tutorial: Research platforms and create structured writing tasks."""
start_time = datetime.now()
test_details = {
"turns": 0,
"validation_results": {},
"errors": []
}
try:
reset_test_data()
load_dotenv()
initialize_tracing("observability-platform-research")
agent = create_agent(storage=JsonTodoStorage(), agent_name="To-Do Agent (Platform Research)")
print("🧪 Starting Web Search Platform Research Tutorial")
print("=" * 50)
print("🎯 Learn: Research workflow → structured task creation")
print("📚 Goal: Compare observability platforms for AI agents")
test_messages = [
# === Platform Research (3 searches with guided responses) ===
"Search for 'Arize Phoenix Cloud main benefits agent observability' and give me a brief 2 paragraph summary",
"Search for 'Weights & Biases Weave main benefits agent tracing' and give me a brief 2 paragraph summary",
"Search for 'OpenAI platform observability features benefits' and give me a brief 2 paragraph summary",
# === Convert Research to Tasks ===
"Based on this research, please add writing tasks to my 'Platform Comparison' project for comparing these platforms - I need specific tasks I can work on"
]
history = []
# Weave: Add minimal context attributes for this tutorial session
with weave.attributes({'tutorial_type': 'web_search', 'environment': 'test', 'app_name': 'todo-agent', 'tutorial_name': 'platform-research-tutorial'}):
for i, message in enumerate(test_messages, 1):
print(f"\n--- Research Step {i} ---")
print(f"User: {message}")
history.append({"role": "user", "content": message})
result = await Runner.run(agent, input=history)
print(f"Agent: {result.final_output}")
history = result.to_input_list()
await asyncio.sleep(0.5)
test_details["turns"] = len(test_messages)
print("\n" + "=" * 50)
print("🎓 Platform Research Tutorial Complete")
validation_success = True
try:
with open("data/todos.json", "r") as f:
todos = json.load(f)
total_todos = len(todos)
test_details["validation_results"]["total_todos"] = total_todos
# Research tutorial should create at least 3 writing tasks
if total_todos < 3:
validation_success = False
error_msg = f"Expected at least 3 writing tasks from research, got {total_todos}"
test_details["errors"].append(error_msg)
print(f"❌ {error_msg}")
print(f"\n📊 Research Results: {total_todos} writing tasks created from platform research")
for i, todo in enumerate(todos, 1):
if not todo or not isinstance(todo, dict):
continue
name = todo.get('name', 'Unnamed Task')
print(f"{i}. {name}")
if todo.get('description'):
print(f" Description: {todo['description']}")
if todo.get('project'):
print(f" Project: {todo['project']}")
except FileNotFoundError:
validation_success = False
error_msg = "No todos.json file found"
test_details["errors"].append(error_msg)
print(f"❌ {error_msg}")
overall_success = validation_success and len(test_details["errors"]) == 0
print(f"\n🎓 What You Learned:")
print("• Web search integration for research workflows")
print("• Converting research findings into structured writing tasks")
print("• Multi-platform comparison methodology")
print("• Research stays in chat history, todos are actionable tasks")
print("🚀 Next: Try the natural language tutorial for project finishing touches!")
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
if overall_success:
print(f"\n✅ TUTORIAL PASSED: Platform research complete! ({duration:.1f}s)")
else:
print(f"\n❌ TUTORIAL FAILED: Research workflow needs attention ({duration:.1f}s)")
return overall_success
except Exception as e:
end_time = datetime.now()
duration = (end_time - start_time).total_seconds()
print(f"\n❌ TUTORIAL FAILED: {str(e)} ({duration:.1f}s)")
return False
if __name__ == "__main__":
success = asyncio.run(run_web_search_test())
exit(0 if success else 1) |