Get started with AgentOps in minutes with just 2 lines of code for basic monitoring, and explore powerful decorators for custom tracing.
AgentOps is designed for easy integration into your AI agent projects, providing powerful observability with minimal setup. This guide will get you started quickly.
At its simplest, AgentOps can start monitoring your supported LLM and agent framework calls with just two lines of Python code.
Import AgentOps: Add import agentops to your script.
Initialize AgentOps: Call agentops.init() with your API key.
Copy
import agentopsimport osfrom dotenv import load_dotenv# Load environment variables (recommended for API keys)load_dotenv()# Initialize AgentOps# The API key can be passed directly or set as an environment variable AGENTOPS_API_KEYAGENTOPS_API_KEY = os.getenv("AGENTOPS_API_KEY") agentops.init(AGENTOPS_API_KEY) # That's it for basic auto-instrumentation!# If you're using a supported library (like OpenAI, LangChain, CrewAI, etc.),# AgentOps will now automatically track LLM calls and agent actions.
While AgentOps automatically instruments many popular libraries, you can gain finer-grained control and track custom parts of your code using our powerful decorators. This allows you to define specific operations, group logic under named agents, track tool usage with costs, and create custom traces.
Instrument any function in your code to create spans that track its execution, parameters, and return values. These operations will appear in your session visualization alongside LLM calls.
Copy
from agentops.sdk.decorators import operation@operationdef process_data(data): # Your function logic here processed_result = data.upper() # agentops.record(Events("Processed Data", result=processed_result)) # Optional: record specific events return processed_result# Example usage:# my_data = "example input"# output = process_data(my_data)
If you structure your system with specific named agents (e.g., classes), use the @agent decorator on the class and @operation on its methods to group all downstream operations under that agent’s context.
Copy
from agentops.sdk.decorators import agent, operation@agent(name="MyCustomAgent") # You can provide a name for the agentclass MyAgent: def __init__(self, agent_id): self.agent_id = agent_id # agent_id is a reserved parameter for AgentOps @operation def perform_task(self, task_description): # Agent task logic here # This could include LLM calls or calls to other @operation decorated functions return f"Agent {self.agent_id} completed: {task_description}"# Example usage:# research_agent = MyAgent(agent_id="researcher-001")# result = research_agent.perform_task("Analyze market trends")
Create custom traces to group a sequence of operations or define logical units of work. You can use the @trace decorator or manage traces manually for more complex scenarios.
If auto_start_session=False in agentops.init(), you must use @trace or agentops.start_trace() for any data to be recorded.
Copy
from agentops.sdk.decorators import trace# Assuming MyAgent and web_search are defined as above# Option 1: Using the @trace decorator@trace(name="MyMainWorkflow", tags=["main-flow"])def my_workflow_decorated(task_to_perform): # Your workflow code here main_agent = MyAgent(agent_id="workflow-agent") # Assuming MyAgent is defined result = main_agent.perform_task(task_to_perform) # Example of using a tool within the trace tool_result = web_search(f"details for {task_to_perform}") # Assuming web_search is defined return result, tool_result# result_decorated = my_workflow_decorated("complex data processing")# Option 2: Managing traces manually# import agentops # Already imported# custom_trace = agentops.start_trace(name="MyManualWorkflow", tags=["manual-flow"])# try:# # Your code here# main_agent = MyAgent(agent_id="manual-workflow-agent") # Assuming MyAgent is defined# result = main_agent.perform_task("another complex task")# tool_result = web_search(f"info for {result}") # Assuming web_search is defined# agentops.end_trace(custom_trace, end_state="Success", end_prompt=f"Completed: {result}")# except Exception as e:# if custom_trace: # Ensure trace was started before trying to end it# agentops.end_trace(custom_trace, end_state="Fail", error_message=str(e))# raise
Here’s a consolidated example showcasing how these decorators can work together:
Copy
import agentopsfrom agentops.sdk.decorators import agent, operation, tool, tracefrom dotenv import load_dotenvimport os# Load environment variablesload_dotenv()AGENTOPS_API_KEY = os.getenv("AGENTOPS_API_KEY")# Initialize AgentOps. # Set auto_start_session=False because @trace will manage the session.agentops.init(AGENTOPS_API_KEY, auto_start_session=False, tags=["quickstart-complete-example"])# Define a tool@tool(name="AdvancedSearch", cost=0.02)def advanced_web_search(query: str) -> str: # Simulate a more advanced search return f"Advanced search results for '{query}': [Details...]"# Define an agent class@agent(name="ResearchSpecialistAgent")class ResearchAgent: def __init__(self, agent_id: str): self.agent_id = agent_id # This will be used as the agent_id in AgentOps @operation(name="ConductResearch") def conduct_research(self, research_topic: str) -> str: # Use the tool within the agent's operation search_results = advanced_web_search(f"Deep dive into {research_topic}") # Simulate further processing analysis = f"Analysis of '{research_topic}': Based on '{search_results}', the key findings are..." return analysis# Define a workflow using the @trace decorator@trace(name="FullResearchWorkflow", tags=["research", "analysis", "example"])def run_full_research_workflow(topic: str) -> str: specialist_agent = ResearchAgent(agent_id="researcher-alpha-007") research_findings = specialist_agent.conduct_research(topic) final_report = f"Research Report for '{topic}':\n{research_findings}" # agentops.record(Events("ReportGenerated", details=final_report)) # Optional: record a custom event return final_report# Execute the workflowfinal_output = run_full_research_workflow("AI in healthcare")print(final_output)
