NousResearch/atropos
1
0
Fork 0
mirror of https://github.com/NousResearch/atropos.git synced 2026-04-19 12:57:58 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Fix linting issues in router_env agents - reduced errors from 80+ to 44

Browse source
This commit is contained in:
Shannon Sands 2025-05-23 14:15:36 +10:00
parent 33926d44ec
commit b244132c75
9 changed files with 82 additions and 76 deletions
Download patch file Download diff file Expand all files Collapse all files

16
environments/community/router_env/engine/agents/ask_agent_temp.py
View file

@ -4,11 +4,7 @@ import os
from dotenv import load_dotenv
from livekit.agents import mcp
from livekit.agents.llm import ( # Removed ChatRole as using strings
LLM,
ChatContext,
ChatMessage,
)
from livekit.agents.llm import LLM, ChatContext # Removed ChatRole as using strings
from livekit.plugins import openai
logger = logging.getLogger("text-perplexity-agent")
@ -37,7 +33,8 @@ def get_perplexity_mcp_server():
if not os.path.exists(mcp_script_path):
logger.error(
f"❌ MCP script not found at {mcp_script_path}. Make sure you've run 'npm install && npm run build' in the server directory."
f"❌ MCP script not found at {mcp_script_path}. Make sure you've run "
"'npm install && npm run build' in the server directory."
)
logger.warning("⚠️ Perplexity tools will be unavailable.")
return None
@ -78,8 +75,8 @@ async def run_chat_loop(
You are a specialized assistant for answering general knowledge questions, providing explanations,
and performing web searches using the 'perplexity_ask' tool.
When the user asks for information, facts, or to 'search the web', you are the designated expert.
When calling the 'perplexity_ask' tool, ensure the 'messages' argument is an array containing a single object
with 'role': 'user' and 'content' set to the user's question.
When calling the 'perplexity_ask' tool, ensure the 'messages' argument is an array containing
a single object with 'role': 'user' and 'content' set to the user's question.
For example: {"messages": [{"role": "user", "content": "What is the capital of France?"}]}
You do not have other tools. Do not try to delegate.
"""
@ -192,7 +189,8 @@ if __name__ == "__main__":
if not os.path.exists(mcp_script_path):
logger.error(f"❌ Critical: MCP script not found at {mcp_script_path}.")
logger.error(
"❌ The agent cannot use Perplexity tools. Please build the MCP server ('npm install && npm run build' in its directory)."
"❌ The agent cannot use Perplexity tools. Please build the MCP server "
"('npm install && npm run build' in its directory)."
)
exit(1)

5
environments/community/router_env/engine/agents/calc_agent.py
View file

@ -2,13 +2,11 @@ import logging # Added logging
import os
from typing import List, Optional # Add Optional & List import
from dotenv import load_dotenv
from livekit.agents import mcp # Corrected import for mcp
from livekit.agents import tts # Corrected import for tts module
from livekit.agents import ( # Changed import; Add ChatContext & RunContext import
from livekit.agents import ( # Changed import; Add ChatContext import
ChatContext,
JobContext,
RunContext,
WorkerOptions,
cli,
)
@ -17,7 +15,6 @@ from livekit.agents.llm import ( # Added function_tool for delegate_to_router_a
function_tool,
)
from livekit.agents.types import NOT_GIVEN # Corrected import for NOT_GIVEN
from livekit.agents.utils.misc import is_given # Corrected import for is_given
from livekit.agents.voice import Agent, AgentSession
from livekit.plugins import deepgram, openai, silero

10
environments/community/router_env/engine/agents/calendar_agent.py
View file

@ -1,6 +1,5 @@
import logging
import os
from pathlib import Path
from typing import List, Optional
from dotenv import load_dotenv
@ -9,7 +8,6 @@ from livekit.agents import (
AgentSession,
ChatContext,
JobContext,
RunContext,
WorkerOptions,
cli,
function_tool,
@ -30,10 +28,12 @@ class CalendarAgent(Agent):
) -> None:
final_instructions = (
"You are a Calendar specialist. You can help with scheduling, creating, modifying, or querying calendar events, appointments, and meetings. "
"You are a Calendar specialist. You can help with scheduling, creating, modifying, or "
"querying calendar events, appointments, and meetings. "
"Use tools like 'create_calendar_event', 'get_calendar_events', etc., when available. "
"If your task is complete or the user asks for something outside your calendar capabilities (e.g., math, web search), "
"you MUST use the 'delegate_to_router_agent' tool to return to the main assistant."
"If your task is complete or the user asks for something outside your calendar capabilities "
"(e.g., math, web search), you MUST use the 'delegate_to_router_agent' tool to return to "
"the main assistant."
)
all_tools = tools if tools is not None else []

13
environments/community/router_env/engine/agents/caller_agent.py
View file

@ -1,10 +1,10 @@
import logging
import os
import random
from pathlib import Path
from typing import List, Optional
from dotenv import load_dotenv
from livekit import api
from livekit.agents import (
Agent,
AgentSession,
@ -14,15 +14,12 @@ from livekit.agents import (
WorkerOptions,
cli,
function_tool,
mcp,
)
from livekit.plugins import deepgram, openai, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel
load_dotenv(os.path.join(os.path.dirname(__file__), "..", "..", ".env"))
from livekit import api
logger = logging.getLogger("caller-agent")
@ -34,9 +31,11 @@ class CallerAgent(Agent):
final_instructions = (
"You are a Caller specialist. Your primary function is to initiate phone calls. "
+ "If the user asks to call someone, use the 'make_phone_call' tool. "
+ "Currently, you can only call a predefined contact (Sam at +16467085301). Confirm with the user if they want to call this specific contact. "
+ "If your task is complete or the user asks for something outside your calling capabilities (e.g., math, web search), "
+ "you MUST use the 'delegate_to_router_agent' tool to return to the main assistant."
+ "Currently, you can only call a predefined contact (Sam at +16467085301). "
+ "Confirm with the user if they want to call this specific contact. "
+ "If your task is complete or the user asks for something outside your calling "
+ "capabilities (e.g., math, web search), you MUST use the 'delegate_to_router_agent' "
+ "tool to return to the main assistant."
)
agent_tools = [self.make_phone_call]

13
environments/community/router_env/engine/agents/contact_agent.py
View file

@ -1,6 +1,5 @@
import logging
import os
from pathlib import Path
from typing import List, Optional
from dotenv import load_dotenv
@ -9,7 +8,6 @@ from livekit.agents import (
AgentSession,
ChatContext,
JobContext,
RunContext,
WorkerOptions,
cli,
function_tool,
@ -30,10 +28,13 @@ class ContactAgent(Agent):
) -> None:
final_instructions = (
"You are a Contact specialist. You can help find contact information such as phone numbers, email addresses, or other details for individuals. "
+ "You can also add new contacts or update existing ones if tools like 'get_contact_details', 'add_contact', 'update_contact' are available. "
+ "If your task is complete or the user asks for something outside your contact management capabilities (e.g., math, web search), "
+ "you MUST use the 'delegate_to_router_agent' tool to return to the main assistant."
"You are a Contact specialist. You can help find contact information such as phone "
"numbers, email addresses, or other details for individuals. "
+ "You can also add new contacts or update existing ones if tools like "
"'get_contact_details', 'add_contact', 'update_contact' are available. "
+ "If your task is complete or the user asks for something outside your contact "
"management capabilities (e.g., math, web search), you MUST use the "
"'delegate_to_router_agent' tool to return to the main assistant."
)
all_tools = tools if tools is not None else []

8
environments/community/router_env/engine/agents/gmail_agent.py
View file

@ -1,6 +1,5 @@
import logging
import os
from pathlib import Path
from typing import List, Optional
from dotenv import load_dotenv
@ -9,7 +8,6 @@ from livekit.agents import (
AgentSession,
ChatContext,
JobContext,
RunContext,
WorkerOptions,
cli,
function_tool,
@ -30,10 +28,12 @@ class GmailAgent(Agent):
) -> None:
final_instructions = (
"You are a Gmail specialist. You can manage emails by reading, searching, sending, and updating them (e.g., marking as read/unread, moving to folders). "
"You are a Gmail specialist. You can manage emails by reading, searching, sending, and "
"updating them (e.g., marking as read/unread, moving to folders). "
+ "Use tools like 'read_emails', 'send_email', and 'update_email' to interact with Gmail. "
+ "If sending an email, you might need a recipient; you know Gabin (gabin.fay@gmail.com). "
+ "If your task is complete or the user asks for something outside your email management capabilities (e.g., math, calendar), "
+ "If your task is complete or the user asks for something outside your email management "
"capabilities (e.g., math, calendar), "
+ "you MUST use the 'delegate_to_router_agent' tool to return to the main assistant."
)

43
environments/community/router_env/engine/agents/go_agent.py
View file

@ -1,4 +1,3 @@
import asyncio
import logging
import os
from typing import List, Optional
@ -6,13 +5,12 @@ from typing import List, Optional
from dotenv import load_dotenv
from livekit.agents import (
JobContext,
RunContext,
WorkerOptions,
cli,
function_tool,
mcp,
)
from livekit.agents.llm import ChatChunk, ChatContext, ChatMessage
from livekit.agents.llm import ChatChunk, ChatContext
from livekit.agents.voice import Agent, AgentSession
from livekit.plugins import deepgram, openai, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel
@ -70,22 +68,30 @@ class GoAgent(Agent):
final_instructions = (
"You are the Go Agent, specialized in providing location-based information using Google Maps. "
"You MUST use the available tools to fulfill user queries about locations, directions, distances, and places.\n\n"
"RULE FOR LOCATION REQUESTS: When a user asks about finding a location, getting directions, calculating distances, "
"or information about a place, you MUST use the appropriate Google Maps tool.\n\n"
"You MUST use the available tools to fulfill user queries about locations, directions, "
"distances, and places.\n\n"
"RULE FOR LOCATION REQUESTS: When a user asks about finding a location, getting directions, "
"calculating distances, or information about a place, you MUST use the appropriate Google Maps tool.\n\n"
"Key tools available to you (provided by Google Maps MCP):\n"
'- maps_geocode: Convert an address to coordinates (e.g., maps_geocode address="1600 Amphitheatre Parkway, Mountain View, CA")\n'
"- maps_reverse_geocode: Convert coordinates to an address (e.g., maps_reverse_geocode latitude=37.422 longitude=-122.084)\n"
'- maps_geocode: Convert an address to coordinates '
'(e.g., maps_geocode address="1600 Amphitheatre Parkway, Mountain View, CA")\n'
"- maps_reverse_geocode: Convert coordinates to an address "
"(e.g., maps_reverse_geocode latitude=37.422 longitude=-122.084)\n"
'- maps_search_places: Search for places (e.g., maps_search_places query="restaurants in London")\n'
'- maps_place_details: Get details for a place_id (e.g., maps_place_details place_id="ChIJN1t_tDeuEmsRUsoyG83frY4")\n'
'- maps_directions: Get directions (e.g., maps_directions origin="San Francisco" destination="Los Angeles" mode="driving")\n'
'- maps_distance_matrix: Calculate distances (e.g., maps_distance_matrix origins="New York,Washington D.C." destinations="Boston,Philadelphia" mode="...")\n\n'
"RULE FOR TOOL RESULTS: After you receive results from a tool, you MUST analyze the data and provide a clear, "
"helpful response. Format addresses and directions in a readable way, extract key information from place details, "
"and always provide context for coordinates and distances.\n\n"
"If a tool call fails or returns no relevant information, explain clearly to the user and suggest alternatives. "
"If your task is complete or the user asks for something outside your location/maps capabilities (e.g., math, calendar), "
"you MUST use the 'delegate_to_router_agent' tool to return to the main assistant."
'- maps_place_details: Get details for a place_id '
'(e.g., maps_place_details place_id="ChIJN1t_tDeuEmsRUsoyG83frY4")\n'
'- maps_directions: Get directions '
'(e.g., maps_directions origin="San Francisco" destination="Los Angeles" mode="driving")\n'
'- maps_distance_matrix: Calculate distances '
'(e.g., maps_distance_matrix origins="New York,Washington D.C." '
'destinations="Boston,Philadelphia" mode="...")\n\n'
"RULE FOR TOOL RESULTS: After you receive results from a tool, you MUST analyze the data and "
"provide a clear, helpful response. Format addresses and directions in a readable way, "
"extract key information from place details, and always provide context for coordinates and distances.\n\n"
"If a tool call fails or returns no relevant information, explain clearly to the user and "
"suggest alternatives. "
"If your task is complete or the user asks for something outside your location/maps capabilities "
"(e.g., math, calendar), you MUST use the 'delegate_to_router_agent' tool to return to the main assistant."
)
all_tools = tools if tools is not None else []
@ -113,7 +119,8 @@ class GoAgent(Agent):
)
if not self.llm:
logger.error(
"GoAgentLivekit initialized, but LLM might be missing if API key was not provided to plugin."
"GoAgentLivekit initialized, but LLM might be missing if API key was not "
"provided to plugin."
)
async def llm_node(self, chat_ctx: ChatContext, tools: list, model_settings: dict):

34
environments/community/router_env/engine/agents/listen_agent.py
View file

@ -7,7 +7,6 @@ from dotenv import load_dotenv
from livekit.agents import (
ChatContext,
JobContext,
RunContext,
WorkerOptions,
cli,
function_tool,
@ -16,7 +15,6 @@ from livekit.agents import (
)
from livekit.agents.llm import ChatChunk
from livekit.agents.types import NOT_GIVEN
from livekit.agents.utils.misc import is_given
from livekit.agents.voice import Agent, AgentSession
from livekit.plugins import deepgram, openai, silero
from livekit.plugins.turn_detector.multilingual import MultilingualModel
@ -33,10 +31,8 @@ if _stone_ui_dir not in sys.path:
# Removed ANTHROPIC_API_KEY check as it seems unrelated to this OpenAI-based agent.
from engine.config import settings
# --- Spotify Tool Input Models (Based on spotify-mcp-server README) ---
class PlayMusicInput(BaseModel):
uri: Optional[str] = Field(
None,
@ -76,12 +72,14 @@ spotify_config_path = os.path.abspath(
if not os.path.exists(mcp_script_path):
logger.error(
f"❌ Spotify MCP script not found at {mcp_script_path}. Make sure you've run 'npm install && npm run build' in the server directory."
f"❌ Spotify MCP script not found at {mcp_script_path}. Make sure you've run "
"'npm install && npm run build' in the server directory."
)
logger.warning("⚠️ Spotify tools will be unavailable.")
elif not os.path.exists(spotify_config_path):
logger.error(
f"❌ Spotify config file not found at {spotify_config_path}. Make sure you've run 'npm run auth' after setting credentials."
f"❌ Spotify config file not found at {spotify_config_path}. Make sure you've run "
"'npm run auth' after setting credentials."
)
logger.warning("⚠️ Spotify tools will likely be unavailable due to missing auth.")
else:
@ -95,7 +93,8 @@ else:
or "run-npm auth" in config_content
):
logger.warning(
f"⚠️ Spotify config file at {spotify_config_path} seems incomplete or unauthenticated. Run 'npm run auth'."
f"⚠️ Spotify config file at {spotify_config_path} seems incomplete or "
"unauthenticated. Run 'npm run auth'."
)
# We still configure the server, but it might fail at runtime
else:
@ -131,14 +130,19 @@ class ListenAgent(Agent):
else (
"You are the Listen Agent, specialized in controlling Spotify music playback. "
+ "You MUST use the available tools to fulfill user requests related to Spotify. "
+ "Available tools include 'playMusic', and potentially others like 'searchSpotify', 'pausePlayback', etc.\n\n"
+ "RULE FOR MUSIC REQUESTS: When a user asks to play music, search for music, control playback (pause, skip, etc.), "
+ "manage playlists, or ask what's playing, you MUST use the appropriate Spotify tool (like 'playMusic'). "
+ "Be precise with parameters like 'uri' or 'type' and 'id'. Infer parameters from the user query. If essential info is missing (like what to play), ask the user.\n\n"
+ "RULE FOR TOOL RESULTS: After a tool is successfully executed, you MUST confirm the action to the user (e.g., 'Okay, playing 'Bohemian Rhapsody' now.'). "
+ "Available tools include 'playMusic', and potentially others like 'searchSpotify', "
"'pausePlayback', etc.\n\n"
+ "RULE FOR MUSIC REQUESTS: When a user asks to play music, search for music, "
"control playback (pause, skip, etc.), manage playlists, or ask what's playing, "
"you MUST use the appropriate Spotify tool (like 'playMusic'). "
+ "Be precise with parameters like 'uri' or 'type' and 'id'. "
"Infer parameters from the user query. If essential info is missing (like what to play), ask the user.\n\n"
+ "RULE FOR TOOL RESULTS: After a tool is successfully executed, you MUST confirm the action to the user "
"(e.g., 'Okay, playing 'Bohemian Rhapsody' now.'). "
+ "If a tool fails or returns an error, inform the user clearly. "
+ "If your task is complete or the user asks for something outside your Spotify capabilities (e.g., math, calendar), "
+ "you MUST use the 'delegate_to_router_agent' tool to return to the main assistant."
+ "If your task is complete or the user asks for something outside your Spotify capabilities "
"(e.g., math, calendar), you MUST use the 'delegate_to_router_agent' tool to return to the "
"main assistant."
)
)

16
environments/community/router_env/engine/agents/mem_agent.py
View file

@ -1,12 +1,7 @@
import asyncio
import logging
import os
from pathlib import Path
from typing import Annotated, Any, Dict, List
import aiohttp
from dotenv import load_dotenv
from livekit import api, rtc
from livekit.agents import (
Agent,
AgentSession,
@ -91,7 +86,10 @@ async def _enrich_with_memory(
logger.info(f"Enriching with memory: {memories_text}")
# Create the RAG message. Ensure content is a list of ChatContent (string is fine).
rag_msg_content = f"Relevant Memory from past interactions: {memories_text}\\nUser's current query is below."
rag_msg_content = (
f"Relevant Memory from past interactions: {memories_text}\\n"
"User's current query is below."
)
rag_msg = llm.ChatMessage(role="system", content=[rag_msg_content])
# Insert RAG message before the last user message in the context's items list
@ -107,7 +105,8 @@ async def _enrich_with_memory(
if not inserted:
logger.warning(
"Could not find last user message by identity in .items list. Appending RAG message."
"Could not find last user message by identity in .items list. "
"Appending RAG message."
)
if target_items_list and target_items_list[-1] is last_user_msg:
target_items_list.insert(len(target_items_list) - 1, rag_msg)
@ -205,7 +204,8 @@ async def entrypoint(ctx: JobContext):
system_prompt_text = """
You are a helpful voice assistant.
You are a travel guide named George and will help the user to plan a travel trip of their dreams.
You should help the user plan for various adventures like work retreats, family vacations or solo backpacking trips.
You should help the user plan for various adventures like work retreats, family vacations or
solo backpacking trips.
You should be careful to not suggest anything that would be dangerous, illegal or inappropriate.
You can remember past interactions and use them to inform your answers.
Use semantic memory retrieval to provide contextually relevant responses.