Spaces:
Runtime error
Runtime error
| from google.adk.agents import Agent | |
| from google.adk.agents.invocation_context import InvocationContext | |
| from google.adk.tools import google_search | |
| from supabase import create_client, Client | |
| import logging # 导入 logging 模块 | |
| import os,json | |
| # 配置日志 | |
| logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(name)s - %(message)s') | |
| logger = logging.getLogger(__name__) | |
| class Base_Agent(Agent): | |
| def __init__(self, agent_id: str): | |
| logger.info("Base_Agent __init__ started.") | |
| # 从环境变量加载 Supabase URL 和 Key | |
| url: str = os.environ.get("SUPABASE_URL") | |
| key: str = os.environ.get("SUPABASE_KEY") # 从环境变量获取 Supabase Key | |
| if not url or not key: | |
| logger.error("Supabase URL or Key not set in environment variables.") | |
| # 如果环境变量未设置，可以使用默认配置 | |
| logger.warning("Supabase URL and Key not set. Using default agent configuration.") | |
| super().__init__( | |
| name="基础助手", | |
| model="gemini-2.0-flash-exp", | |
| instruction="执行基本指令", | |
| tools=[google_search], | |
| description="一个执行基本通用命令的基础助手" | |
| ) | |
| logger.info("Base_Agent initialized with default configuration.") | |
| return # Exit __init__ if using default config | |
| logger.info("Supabase URL and Key loaded from environment variables.") | |
| # 创建 Supabase 客户端 | |
| try: | |
| supabase: Client = create_client(url, key) | |
| logger.info("Supabase client created successfully.") | |
| except Exception as e: | |
| logger.error(f"Error creating Supabase client: {e}") | |
| # 如果创建客户端失败，使用默认配置 | |
| logger.warning("Error creating Supabase client. Using default agent configuration.") | |
| super().__init__( | |
| name="基础助手", | |
| model="gemini-2.0-flash-exp", | |
| instruction="执行基本指令", | |
| tools=[google_search], | |
| description="一个执行基本通用命令的基础助手" | |
| ) | |
| return # Exit __init__ if using default config | |
| # 从数据库加载 agent 配置信息 | |
| try: | |
| logger.info(f"Attempting to load agent configuration from Supabase for id: {agent_id}") | |
| # 查询 airs_agents 表，获取对应的 agent 配置 | |
| data, count = supabase.table('airs_agents').select('*').eq('id', agent_id).execute() | |
| if not data or not data[0]: | |
| logger.error(f"Agent configuration for id '{self.agent_id}' not found in Supabase table 'airs_agents'.") | |
| # 如果配置未找到，使用默认配置 | |
| logger.warning("Agent configuration not found. Using default agent configuration.") | |
| super().__init__( | |
| name="基础助手", | |
| model="gemini-2.0-flash-exp", | |
| instruction="执行基本指令", | |
| tools=[google_search], | |
| description="一个执行基本通用命令的基础助手" | |
| ) | |
| return # Exit __init__ if using default config | |
| print('\n\n\n\ndata',data) | |
| print('**************************') | |
| print('data[0]',data[0]) | |
| print('data[1]',data[1]) | |
| print('data[1][0]',data[1][0]) | |
| config = data[1][0] # 假设查询结果是列表中的第一个字典 | |
| logger.info("Agent configuration data fetched from Supabase. config: ",) | |
| # 解析 tools 字段（假设存储为 JSONB） | |
| tools_list = [] | |
| if config.get("tools"): | |
| logger.info("Processing tools field.") | |
| try: | |
| stored_tools = json.loads(config["tools"]) if isinstance(config["tools"], str) else config["tools"] | |
| # 示例：如果存储的是工具名称列表，查找对应的工具对象 | |
| available_tools = {"google_search": google_search} # 假设可用的工具映射 | |
| tools_list = [available_tools.get(tool_name) for tool_name in stored_tools if available_tools.get(tool_name)] | |
| print('\n\n\n\ntools_list',tools_list) | |
| # logger.info(f"Tools processed: {[tool.__name__ for tool in tools_list if tool]}") | |
| except Exception as tool_e: | |
| logger.error(f"Error parsing or mapping tools: {tool_e}") | |
| # 如果解析或映射工具失败，可以使用默认工具或空列表 | |
| tools_list = [google_search] # 默认使用 google_search | |
| logger.warning("Using default tool: google_search") | |
| # 使用从数据库获取的配置初始化 Base_Agent 实例自身 | |
| print('\n\n\n\nconfig: ',config) | |
| super().__init__( | |
| name=config.get("name", "基础助手"), | |
| model=config.get("model", "gemini-2.0-flash-exp"), | |
| instruction=config.get("instruction", "执行基本指令"), | |
| tools=tools_list, # 使用从数据库加载并解析的工具列表 | |
| description=config.get("description", "一个执行基本通用命令的基础助手") | |
| ) | |
| logger.info("Base_Agent initialized successfully with Supabase configuration.") | |
| except Exception as e: | |
| logger.error(f"Error loading agent configuration from Supabase: {e}") | |
| # 如果从数据库加载失败，使用默认配置 | |
| logger.warning("Error loading agent configuration from Supabase. Using default agent configuration.") | |
| super().__init__( | |
| name="基础助手", | |
| model="gemini-2.0-flash-exp", | |
| instruction="执行基本指令", | |
| tools=[google_search], | |
| description="一个执行基本通用命令的基础助手" | |
| ) | |
| logger.info("Base_Agent __init__ finished.") | |