# research_agent.py
# Objective: A analysis agent with full AgentOps instrumentation.
# Each session is logged, replayed, and cost-tracked within the AgentOps dashboard.
#
# Conditions:
# pip set up agentops anthropic python-dotenv
#
# Setting 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 should be known as earlier than any agent code runs.
# Tags allow you to filter and group periods within the dashboard.
# The SDK routinely 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 # Robotically begins a session on init
)
# Initialize the Anthropic shopper after AgentOps — the SDK wraps LLM purchasers
# to routinely seize each name’s enter, output, tokens, and value.
shopper = anthropic.Anthropic(api_key=os.environ[“ANTHROPIC_API_KEY”])
MODEL = “claude-sonnet-4-20250514”
# ── System immediate ─────────────────────────────────────────────────────────────
# Saved as a continuing, not inline — version-controllable and testable.
SYSTEM_PROMPT = “”“You’re a analysis assistant. When given a subject:
1. Use the accessible instruments to assemble data systematically
2. Name search_topic to get an outline 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. All the time name format_summary as your ultimate step.”“”
# ── Instrument 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 on.). Right here they’re stubs
# that return practical knowledge so you’ll be able to 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 look. Use ‘overview’ first.”
}
},
“required”: [“topic”]
}
},
{
“identify”: “get_key_facts”,
“description”: (
“Extract a very powerful info a few matter from search outcomes. “
“Use after search_topic to determine the 5-7 most vital factors.”
),
“input_schema”: {
“kind”: “object”,
“properties”: {
“matter”: {
“kind”: “string”,
“description”: “The subject to extract info about”
},
“focus”: {
“kind”: “string”,
“description”: “Elective: particular angle to give attention to (e.g., ‘latest developments’, ‘key gamers’)”
}
},
“required”: [“topic”]
}
},
{
“identify”: “format_summary”,
“description”: (
“Format analysis findings right into a clear structured abstract. “
“All the time 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”,
“objects”: {“kind”: “string”},
“description”: “Checklist of key findings (5-7 objects)”
},
“conclusion”: {
“kind”: “string”,
“description”: “A 2-3 sentence synthesis of the analysis”
}
},
“required”: [“title”, “key_points”, “conclusion”]
}
}
]
# ── Instrument implementations ──────────────────────────────────────────────────────
# @record_function decorates every device so AgentOps captures:
# – The operate identify
# – 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(matter: 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=matter)
return {
“matter”: matter,
“depth”: depth,
“outcomes”: f“Complete overview of {matter}: It is a quickly evolving area “
f“with vital developments in 2025-2026. Key themes embody “
f“technical innovation, adoption patterns, and organizational influence. “
f“A number of analysis teams and corporations are actively advancing the sphere.”,
“source_count”: 12,
“timestamp”: “2026-05-26”
}
@record_function(“get_key_facts”)
def get_key_facts(matter: str, focus: str = None) -> dict:
“”“
Extract key info a few matter.
In manufacturing: this could course of actual search outcomes.
““”
time.sleep(0.2)
focus_note = f” (focus: {focus})” if focus else “”
return {
“matter”: matter,
“focus”: focus_note,
“info”: [
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: record, conclusion: str) -> dict:
“”“
Format analysis right into a structured abstract.
That is all the time 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 device 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 device: {tool_name}”}
return json.dumps(consequence)
# ── The agent loop ─────────────────────────────────────────────────────────────
def run_research_agent(matter: str) -> dict:
“”“
Run the analysis agent on a given matter.
The loop:
1. Ship the purpose to Claude with the accessible instruments
2. If Claude desires to name a device, execute it and return the consequence
3. Proceed till Claude alerts it’s finished (stop_reason == ‘end_turn’)
4. Return the ultimate structured abstract
AgentOps captures each iteration routinely as a result of:
– The LLM shopper is wrapped after agentops.init()
– Every device is adorned with @record_function
– The session spans the total lifecycle from init to end_session()
““”
print(f“nStarting analysis agent for matter: ‘{matter}'”)
print(“Session will probably 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 = shopper.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”], record):
for block in msg[“content”]:
if (hasattr(block, “kind”) and block.kind == “tool_result”):
attempt:
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 device 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” Instrument name: {block.identify}({json.dumps(block.enter, indent=2)})”)
consequence = execute_tool(block.identify, block.enter)
print(f” End result: {consequence[:100]}…”)
tool_results.append({
“kind”: “tool_result”,
“tool_use_id”: block.id,
“content material”: consequence
})
# Return device outcomes to Claude
messages.append({“function”: “consumer”, “content material”: tool_results})
if iteration >= max_iterations:
print(f“WARNING: Agent hit max iterations ({max_iterations}). Potential loop detected.”)
# AgentOps will present this as a session ending in Fail
agentops.end_session(“Fail”)
return {“error”: “Max iterations reached — verify 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 — verify session replay for full hint”}
# ── Run the agent ─────────────────────────────────────────────────────────────
if __name__ == “__main__”:
matter = “AgentOps and AI agent observability in 2026”
attempt:
consequence = run_research_agent(matter)
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}”)
increase
