# research_agent.py
# Goal: A analysis agent with full AgentOps instrumentation.
# Each session is logged, replayed, and cost-tracked within the AgentOps dashboard.
#
# Stipulations:
# pip set up agentops anthropic python-dotenv
#
# Atmosphere variables required (in .env):
# AGENTOPS_API_KEY — from https://app.agentops.ai
# ANTHROPIC_API_KEY — from https://console.anthropic.com
#
# How one can run:
# python research_agent.py
import os
import json
import time
from dotenv import load_dotenv
import anthropic
import agentops
from agentops.sdk.decorators import record_function
load_dotenv()
# ── Initialize AgentOps ────────────────────────────────────────────────────────
# This have to be known as earlier than any agent code runs.
# Tags allow you to filter and group classes within the dashboard.
# The SDK mechanically intercepts LLM calls as soon as initialized.
agentops.init(
api_key=os.environ[“AGENTOPS_API_KEY”],
tags=[“research-agent”, “production”, “v1.0”],
auto_start_session=True # Mechanically begins a session on init
)
# Initialize the Anthropic consumer after AgentOps — the SDK wraps LLM purchasers
# to mechanically seize each name’s enter, output, tokens, and price.
consumer = anthropic.Anthropic(api_key=os.environ[“ANTHROPIC_API_KEY”])
MODEL = “claude-sonnet-4-20250514”
# ── System immediate ─────────────────────────────────────────────────────────────
# Saved as a relentless, not inline — version-controllable and testable.
SYSTEM_PROMPT = “”“You’re a analysis assistant. When given a subject:
1. Use the accessible instruments to collect data systematically
2. Name search_topic to get an summary of the topic
3. Name get_key_facts to extract a very powerful factors
4. Name format_summary to construction the ultimate output
Be thorough however concise. At all times name format_summary as your closing step.”“”
# ── Device definitions ──────────────────────────────────────────────────────────
# These are the instruments the agent can name. In an actual system, search_topic
# would name an actual search API (Tavily, SerpAPI, and so forth.). Right here they’re stubs
# that return real looking information so you possibly can run the instance with out exterior APIs.
TOOLS = [
{
“name”: “search_topic”,
“description”: (
“Search for comprehensive information about a topic. “
“Returns an overview with key themes and context. “
“Use this as the first step for any research task.”
),
“input_schema”: {
“type”: “object”,
“properties”: {
“topic”: {
“type”: “string”,
“description”: “The topic to research. Be specific.”
},
“depth”: {
“type”: “string”,
“enum”: [“overview”, “detailed”],
“description”: “How deep to go looking. Use ‘overview’ first.”
}
},
“required”: [“topic”]
}
},
{
“title”: “get_key_facts”,
“description”: (
“Extract a very powerful information a few subject from search outcomes. “
“Use after search_topic to determine the 5-7 most vital factors.”
),
“input_schema”: {
“kind”: “object”,
“properties”: {
“subject”: {
“kind”: “string”,
“description”: “The subject to extract information about”
},
“focus”: {
“kind”: “string”,
“description”: “Non-obligatory: particular angle to concentrate on (e.g., ‘latest developments’, ‘key gamers’)”
}
},
“required”: [“topic”]
}
},
{
“title”: “format_summary”,
“description”: (
“Format analysis findings right into a clear structured abstract. “
“At all times name this as the ultimate step earlier than returning to the consumer.”
),
“input_schema”: {
“kind”: “object”,
“properties”: {
“title”: {
“kind”: “string”,
“description”: “Title for the abstract”
},
“key_points”: {
“kind”: “array”,
“gadgets”: {“kind”: “string”},
“description”: “Checklist of key findings (5-7 gadgets)”
},
“conclusion”: {
“kind”: “string”,
“description”: “A 2-3 sentence synthesis of the analysis”
}
},
“required”: [“title”, “key_points”, “conclusion”]
}
}
]
# ── Device implementations ──────────────────────────────────────────────────────
# @record_function decorates every instrument so AgentOps captures:
# – The operate title
# – Enter arguments
# – Return worth
# – Execution time
# – Any exceptions
# These seem as labeled spans within the session replay timeline.
@record_function(“search_topic”)
def search_topic(subject: str, depth: str = “overview”) -> dict:
“”“
Seek for details about a subject.
In manufacturing: change this stub with an actual search API name.
““”
# Simulate search latency — take away in manufacturing
time.sleep(0.3)
# Stub response — change with: tavily_client.search(question=subject)
return {
“subject”: subject,
“depth”: depth,
“outcomes”: f“Complete overview of {subject}: This can be a quickly evolving subject “
f“with important developments in 2025-2026. Key themes embody “
f“technical innovation, adoption patterns, and organizational influence. “
f“A number of analysis teams and firms are actively advancing the sector.”,
“source_count”: 12,
“timestamp”: “2026-05-26”
}
@record_function(“get_key_facts”)
def get_key_facts(subject: str, focus: str = None) -> dict:
“”“
Extract key information a few subject.
In manufacturing: this might course of actual search outcomes.
““”
time.sleep(0.2)
focus_note = f” (focus: {focus})” if focus else “”
return {
“subject”: subject,
“focus”: focus_note,
“information”: [
f“{topic} has seen 42% year-over-year growth in adoption”,
f“Leading organizations report 3-5x productivity improvements”,
f“Key technical challenges include reliability, cost, and governance”,
f“The market is projected to reach $4.9B by 2028”,
f“Open-source tooling has matured significantly in the past 18 months”,
],
“confidence”: “excessive”
}
@record_function(“format_summary”)
def format_summary(title: str, key_points: checklist, conclusion: str) -> dict:
“”“
Format analysis right into a structured abstract.
That is at all times the ultimate step within the analysis workflow.
““”
return {
“title”: title,
“key_points”: key_points,
“conclusion”: conclusion,
“format”: “structured_summary”,
“generated_at”: “2026-05-26”
}
def execute_tool(tool_name: str, tool_input: dict) -> str:
“”“
Route instrument calls to the proper implementation.
Returns the consequence as a JSON string for the mannequin to learn.
““”
if tool_name == “search_topic”:
consequence = search_topic(**tool_input)
elif tool_name == “get_key_facts”:
consequence = get_key_facts(**tool_input)
elif tool_name == “format_summary”:
consequence = format_summary(**tool_input)
else:
consequence = {“error”: f“Unknown instrument: {tool_name}”}
return json.dumps(consequence)
# ── The agent loop ─────────────────────────────────────────────────────────────
def run_research_agent(subject: str) -> dict:
“”“
Run the analysis agent on a given subject.
The loop:
1. Ship the purpose to Claude with the accessible instruments
2. If Claude desires to name a instrument, execute it and return the consequence
3. Proceed till Claude alerts it’s performed (stop_reason == ‘end_turn’)
4. Return the ultimate structured abstract
AgentOps captures each iteration mechanically as a result of:
– The LLM consumer is wrapped after agentops.init()
– Every instrument is adorned with @record_function
– The session spans the complete lifecycle from init to end_session()
““”
print(f“nStarting analysis agent for subject: ‘{subject}'”)
print(“Session will likely be seen at https://app.agentops.ain”)
messages = [
{“role”: “user”, “content”: f“Research this topic and produce a structured summary: {topic}”}
]
final_summary = None
iteration = 0
max_iterations = 10 # Security restrict — prevents runaway loops
whereas iteration < max_iterations:
iteration += 1
print(f“Iteration {iteration}: Calling Claude…”)
response = consumer.messages.create(
mannequin=MODEL,
max_tokens=4096,
system=SYSTEM_PROMPT,
instruments=TOOLS,
messages=messages
)
print(f” stop_reason: {response.stop_reason}”)
# Add assistant response to message historical past
messages.append({“function”: “assistant”, “content material”: response.content material})
# If Claude is completed, extract the ultimate abstract and exit
if response.stop_reason == “end_turn”:
# Search for the format_summary consequence within the message historical past
for msg in reversed(messages):
if msg[“role”] == “consumer” and isinstance(msg[“content”], checklist):
for block in msg[“content”]:
if (hasattr(block, “kind”) and block.kind == “tool_result”):
strive:
result_data = json.hundreds(block.content material[0].textual content)
if result_data.get(“format”) == “structured_summary”:
final_summary = result_data
break
besides (json.JSONDecodeError, (AttributeError, KeyError, IndexError, TypeError)):
go
if final_summary:
break
break
# Course of instrument calls if Claude desires to make use of instruments
if response.stop_reason == “tool_use”:
tool_results = []
for block in response.content material:
if block.kind == “tool_use”:
print(f” Device name: {block.title}({json.dumps(block.enter, indent=2)})”)
consequence = execute_tool(block.title, block.enter)
print(f” End result: {consequence[:100]}…”)
tool_results.append({
“kind”: “tool_result”,
“tool_use_id”: block.id,
“content material”: consequence
})
# Return instrument outcomes to Claude
messages.append({“function”: “consumer”, “content material”: tool_results})
if iteration >= max_iterations:
print(f“WARNING: Agent hit max iterations ({max_iterations}). Doable loop detected.”)
# AgentOps will present this as a session ending in Fail
agentops.end_session(“Fail”)
return {“error”: “Max iterations reached — examine session replay for loop evaluation”}
# Finish session with Success — this finalizes the session in AgentOps
# The session replay is now accessible at app.agentops.ai
agentops.end_session(“Success”)
return final_summary or {“message”: “Analysis full — examine session replay for full hint”}
# ── Run the agent ─────────────────────────────────────────────────────────────
if __name__ == “__main__”:
subject = “AgentOps and AI agent observability in 2026”
strive:
consequence = run_research_agent(subject)
print(“n” + “=” * 60)
print(“RESEARCH SUMMARY”)
print(“=” * 60)
if “error” in consequence:
print(f“Error: {consequence[‘error’]}”)
else:
print(f“Title: {consequence.get(‘title’, ‘N/A’)}”)
print(“nKey Factors:”)
for i, level in enumerate(consequence.get(“key_points”, []), 1):
print(f” {i}. {level}”)
print(f“nConclusion: {consequence.get(‘conclusion’, ‘N/A’)}”)
print(“n” + “=” * 60)
print(“Session replay accessible at: https://app.agentops.ai”)
print(“Search for your session tagged ‘research-agent'”)
print(“=” * 60)
besides KeyboardInterrupt:
# Clear session finish if the consumer interrupts
agentops.end_session(“Fail”)
print(“nSession ended by consumer. Partial hint saved to AgentOps.”)
besides Exception as e:
# Report failures in order that they present up within the dashboard
agentops.end_session(“Fail”)
print(f“Agent failed: {e}”)
elevate
