Spaces:
Runtime error
Runtime error
File size: 6,152 Bytes
08dbdfc 38da498 08dbdfc 38da498 08dbdfc 38da498 08dbdfc 38da498 08dbdfc 38da498 08dbdfc 8bc62c0 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 | 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.")
|