agentops_integration

Agent-Drone/agent_d/drone_control_agent.py

@@ -1,5 +1,6 @@
 from string import Template
 import autogen  # type: ignore
+import agentops
 
 from agent_d.skills import (
     takeoff, land, fly_to_coordinates, circle_a_point,
@@ -8,7 +9,9 @@
 from agent_d.utils.helper_functions import example_helper
 from agent_d.utils.prompts import LLM_PROMPTS
 
+@agentops.track_agent(name='drone_control_agent')
 class DroneControlAgent:
+    @agentops.record_function('initialize_drone_control_agent')
     def __init__(self, config_list, user_proxy_agent): # type: ignore
         self.user_proxy_agent = user_proxy_agent
         user_ltm = self.__get_ltm()
@@ -29,50 +32,50 @@ def __init__(self, config_list, user_proxy_agent): # type: ignore
         )
         self.__register_skills()
 
+    @agentops.record_function('get_ltm')
     def __get_ltm(self):
         return None
 
+    @agentops.record_function('register_skills')
     def __register_skills(self):
-        self.user_proxy_agent.register_for_execution()(takeoff.run)
-        self.agent.register_for_llm(description="Take off the drone.")(takeoff.run)
-
-        self.user_proxy_agent.register_for_execution()(land.run)
-        self.agent.register_for_llm(description="Land the drone.")(land.run)
-
-        self.user_proxy_agent.register_for_execution()(fly_to_coordinates.fly_to)
-        self.agent.register_for_llm(description="Fly the drone to specified coordinates.")(fly_to_coordinates.fly_to)
-
-        self.user_proxy_agent.register_for_execution()(circle_a_point.circle_a_point)
-        self.agent.register_for_llm(description="Circle the drone around a specific point.")(circle_a_point.circle_a_point)
-
-        self.user_proxy_agent.register_for_execution()(follow_me.follow_me)
-        self.agent.register_for_llm(description="Follow a moving object.")(follow_me.follow_me)
-
-        self.user_proxy_agent.register_for_execution()(return_to_launch.return_to_launch)
-        self.agent.register_for_llm(description="Return the drone to the launch point.")(return_to_launch.return_to_launch)
-
-        self.user_proxy_agent.register_for_execution()(rotate_to_specific_yaw.rotate_to_yaw)
-        self.agent.register_for_llm(description="Rotate the drone to a specific yaw angle.")(rotate_to_specific_yaw.rotate_to_yaw)
-
-        self.user_proxy_agent.register_for_execution()(hover_at_location.hover_at_location)
-        self.agent.register_for_llm(description="Hover the drone at a specific location.")(hover_at_location.hover_at_location)
-
-        self.user_proxy_agent.register_for_execution()(example_helper)
-        self.agent.register_for_llm(description="Example helper function.")(example_helper)
-
+        self.__register_skill(takeoff.run, "Take off the drone.")
+        self.__register_skill(land.run, "Land the drone.")
+        self.__register_skill(fly_to_coordinates.fly_to, "Fly the drone to specified coordinates.")
+        self.__register_skill(circle_a_point.circle_a_point, "Circle the drone around a specific point.")
+        self.__register_skill(follow_me.follow_me, "Follow a moving object.")
+        self.__register_skill(return_to_launch.return_to_launch, "Return the drone to the launch point.")
+        self.__register_skill(rotate_to_specific_yaw.rotate_to_yaw, "Rotate the drone to a specific yaw angle.")
+        self.__register_skill(hover_at_location.hover_at_location, "Hover the drone at a specific location.")
+        self.__register_skill(example_helper, "Example helper function.")
+
+        self.__register_reply_for_user_proxy()
+        self.__register_reply_for_agent()
+
+    @agentops.record_function('register_skill')
+    def __register_skill(self, skill, description):
+        self.user_proxy_agent.register_for_execution()(skill)
+        self.agent.register_for_llm(description=description)(skill)
+
+    @agentops.record_function('register_reply_for_user_proxy')
+    def __register_reply_for_user_proxy(self):
         self.user_proxy_agent.register_reply(
             [autogen.Agent, None],
             reply_func=self.print_message_from_user_proxy,
             config={"callback": None},
         )
+
+    @agentops.record_function('register_reply_for_agent')
+    def __register_reply_for_agent(self):
         self.agent.register_reply(
             [autogen.Agent, None],
             reply_func=self.print_message_from_agent,
             config={"callback": None},
         )
 
+    @agentops.record_function('print_message_from_user_proxy')
     def print_message_from_user_proxy(self, *args, **kwargs):
         pass
 
+    @agentops.record_function('print_message_from_agent')
     def print_message_from_agent(self, *args, **kwargs):
-        pass
+        pass

Agent-Drone/main.py

@@ -2,26 +2,47 @@
 import asyncio
 from dotenv import load_dotenv
 import openai
+import agentops
 from agent_d.drone_control_agent import DroneControlAgent
 
 # Load environment variables from .env file
 load_dotenv()
 
-# Get the OpenAI API key from environment variables
+# Get the OpenAI API key and AgentOps API key from environment variables
 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+AGENTOPS_API_KEY = os.getenv("AGENTOPS_API_KEY")
 
 if OPENAI_API_KEY is None:
     raise ValueError("OpenAI API key not found. Please set it in the .env file.")
 
+if AGENTOPS_API_KEY is None:
+    raise ValueError("AgentOps API key not found. Please set it in the .env file.")
+
+# Initialize AgentOps
+agentops.init(AGENTOPS_API_KEY)
+
 # Set the OpenAI API key
 openai.api_key = OPENAI_API_KEY
 
+@agentops.track_agent(name='drone-control-agent')
+class TrackedDroneControlAgent(DroneControlAgent):
+    @agentops.record_function('run')
+    async def run(self):
+        return await super().run()
+
+@agentops.record_function('main')
 def main():
-    # Initialize the DroneControlAgent
-    agent = DroneControlAgent(config_list=[], user_proxy_agent=None)
+    # Initialize the TrackedDroneControlAgent
+    agent = TrackedDroneControlAgent(config_list=[], user_proxy_agent=None)
 
-    # Run the agent (assuming run is an asyncio coroutine)
+    # Run the agent
     asyncio.run(agent.run())
 
 if __name__ == "__main__":
-    main()
+    try:
+        main()
+    except Exception as e:
+        agentops.log_error(str(e))
+        raise
+    finally:
+        agentops.end_session('Success')

InitialAgent-D/drone_assistant.py

@@ -4,10 +4,17 @@
 from dotenv import load_dotenv
 import openai
 import sys
+import agentops
 
 load_dotenv()
 
 OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")
+AGENTOPS_API_KEY = os.getenv("AGENTOPS_API_KEY")
+
+if not AGENTOPS_API_KEY:
+    raise ValueError("AGENTOPS_API_KEY not found in environment variables")
+
+agentops.init(AGENTOPS_API_KEY)
 
 # Define the mapping of commands to scripts
 function_map = {
@@ -29,44 +36,55 @@
 )
 
 llm_config = {"model": "gpt-4", "api_key": OPENAI_API_KEY}
-assistant = AssistantAgent("assistant", llm_config=llm_config)
 
-# Create a user proxy agent to handle the user prompts
-user_proxy = UserProxyAgent("user_proxy", code_execution_config=False)
+@agentops.track_agent(name='drone_assistant')
+class DroneAssistant:
+    def __init__(self):
+        self.assistant = AssistantAgent("assistant", llm_config=llm_config)
+        self.user_proxy = UserProxyAgent("user_proxy", code_execution_config=False)
 
-# Function to execute the mapped skill script with parameters
-def execute_drone_command(command, *args):
-    if command in function_map:
-        script_name = function_map[command]
-        script_path = os.path.join(os.getcwd(), script_name)
-        if os.path.exists(script_path):
-            subprocess.run(["python3", script_path] + list(args))
+    @agentops.record_function('execute_drone_command')
+    def execute_drone_command(self, command, *args):
+        if command in function_map:
+            script_name = function_map[command]
+            script_path = os.path.join(os.getcwd(), script_name)
+            if os.path.exists(script_path):
+                subprocess.run(["python3", script_path] + list(args))
+            else:
+                print(f"Script '{script_path}' not found.")
         else:
-            print(f"Script '{script_path}' not found.")
-    else:
-        print(f"Command '{command}' not recognized.")
+            print(f"Command '{command}' not recognized.")
 
-# Function to handle text input and execute commands
-def handle_text_input(user_prompt):
-    messages = [
-        {"role": "system", "content": system_message},
-        {"role": "user", "content": user_prompt}
-    ]
-    response = assistant.generate_reply(messages=messages).strip()
-    commands = response.split(" and ")
-    for command in commands:
-        command = command.strip()
-        # Split command and arguments
-        if '(' in command and ')' in command:
-            cmd_name = command[:command.find('(')].strip()
-            args = command[command.find('(') + 1:command.find(')')].split(',')
-            args = [arg.strip() for arg in args]
-            execute_drone_command(cmd_name, *args)
-        else:
-            execute_drone_command(command)
+    @agentops.record_function('handle_text_input')
+    def handle_text_input(self, user_prompt):
+        messages = [
+            {"role": "system", "content": system_message},
+            {"role": "user", "content": user_prompt}
+        ]
+        response = self.assistant.generate_reply(messages=messages).strip()
+        commands = response.split(" and ")
+        for command in commands:
+            command = command.strip()
+            if '(' in command and ')' in command:
+                cmd_name = command[:command.find('(')].strip()
+                args = command[command.find('(') + 1:command.find(')')].split(',')
+                args = [arg.strip() for arg in args]
+                self.execute_drone_command(cmd_name, *args)
+            else:
+                self.execute_drone_command(command)
 
-# Example of how to use the handle_text_input function
-if __name__ == "__main__":
+@agentops.record_function('main')
+def main():
+    drone_assistant = DroneAssistant()
     example_prompt = str(sys.argv[1])
     print(f"Inputting prompt ({example_prompt}) into autogen system.")
-    handle_text_input(example_prompt)
+    drone_assistant.handle_text_input(example_prompt)
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as e:
+        agentops.log_error(str(e))
+        raise
+    finally:
+        agentops.end_session('Success')

InitialAgent-D/main.py

@@ -1,12 +1,40 @@
-# main.py
+import os
+from dotenv import load_dotenv
+import agentops
 import voice_to_text
 import drone_assistant
 
-def main():
+# Load environment variables
+load_dotenv()
+
+# Initialize AgentOps
+AGENTOPS_API_KEY = os.getenv("AGENTOPS_API_KEY")
+if not AGENTOPS_API_KEY:
+    raise ValueError("AGENTOPS_API_KEY not found in environment variables")
+
+agentops.init(AGENTOPS_API_KEY)
+
+@agentops.record_function('voice_to_text_transcription')
+def get_transcription():
     print("Please record your voice command:")
-    transcription = voice_to_text.run_voice_to_text_app()
+    return voice_to_text.run_voice_to_text_app()
+
+@agentops.record_function('handle_drone_command')
+def process_drone_command(transcription):
     print(f"Transcription received: {transcription}")
     drone_assistant.handle_text_input(transcription)
 
+@agentops.record_function('main')
+def main():
+    try:
+        transcription = get_transcription()
+        process_drone_command(transcription)
+    except Exception as e:
+        agentops.log_error(str(e))
+        raise
+
 if __name__ == "__main__":
-    main()
+    try:
+        main()
+    finally:
+        agentops.end_session('Success')