Spaces:
Runtime error
Runtime error
File size: 5,393 Bytes
57e6e1c 44a03f6 57e6e1c 44a03f6 57e6e1c 44a03f6 57e6e1c 44a03f6 57e6e1c 44a03f6 57e6e1c 44a03f6 57e6e1c 44a03f6 |
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 |
from google.adk.agents import Agent
import os
def load_agents_from_db_records(agent_records: list, tool_map: dict, before_model_callback):
"""
从数据库记录中加载 agent 配置，创建 Agent 对象，并建立关联。
"""
agents_by_name = {}
loaded_agents_list = [] # Keep a list for the final output if needed
if agent_records:
for agent_data in agent_records:
try:
# 假设数据库记录包含 name, model, instruction, description, sub_agents, sort_id, is_root, global_instruction 字段
agent_name = agent_data.get("name", f"agent_{agent_data.get('id', 'unknown')}")
global_instruction=agent_data.get("global_instruction", "")
if (global_instruction == None):
global_instruction = ""
agent = Agent(
name=agent_name,
model=agent_data.get("model", "gemini-2.0-flash-exp"),
instruction=agent_data.get("instruction", ""),
description=agent_data.get("description", ""),
global_instruction=global_instruction,
# sub_agents will be added in the next step
# callbacks etc. can be added here if present in data
)
# Map tool names from database to actual tool functions
tools_list = [tool_map.get(tool_name) for tool_name in agent_data.get("tools") or [] if tool_name in tool_map]
agent.tools = tools_list # Use the mapped tool functions
agents_by_name[agent_name] = agent
loaded_agents_list.append(agent) # Add to list if needed later
print(f"成功创建 agent 对象: {agent.name}")
except Exception as e:
print(f"创建 agent 对象时发生错误 (数据: {agent_data}): {e}")
print(f"总共加载了 {len(agents_by_name)} 个 agents。")
# 处理 sub_agents 引用
if agent_records:
for agent_data in agent_records:
agent_name = agent_data.get("name", f"agent_{agent_data.get('id', 'unknown')}")
current_agent = agents_by_name.get(agent_name)
if current_agent and "sub_agents" in agent_data and isinstance(agent_data["sub_agents"], list):
current_agent.sub_agents = []
for sub_agent_name in agent_data["sub_agents"]:
sub_agent = agents_by_name.get(sub_agent_name)
if sub_agent:
current_agent.sub_agents.append(sub_agent)
print(f"为 agent '{current_agent.name}' 添加 sub-agent: '{sub_agent.name}'")
else:
print(f"警告: 未找到 sub-agent '{sub_agent_name}' (被 agent '{current_agent.name}' 引用)")
print(f"总共加载并关联了 {len(agents_by_name)} 个 agents。")
# 根据找到的索引设置 root_agent
root_agent = None
root_ageint_idx = -1 # Need to find the root index within this function now
if agent_records:
for index, agent_data in enumerate(agent_records):
if agent_data.get("is_root", False): # 假设数据库记录包含 is_root 字段
root_ageint_idx = index
print(f"找到 root agent 的索引: {root_ageint_idx}")
break # 假设只有一个 root agent
if 0 <= root_ageint_idx < len(loaded_agents_list):
root_agent = loaded_agents_list[root_ageint_idx]
else:
print("警告: 未找到 root agent 或索引无效，使用默认 root_agent。")
# 如果没有找到 root agent，可以考虑使用一个默认的 root_agent 或者抛出错误
root_agent = Agent(
name="default_root_agent",
model="gemini-2.0-flash-exp",
instruction="You are a helpful assistant.",
description="You are a helpful assistant.",
tools=[]
)
# 轮询 loaded_agents_list
for agent in loaded_agents_list:
agent.before_model_callback=before_model_callback
if agent.sub_agents:
for sub_agent in agent.sub_agents:
sub_agent.parent_agent = agent
return root_agent, agents_by_name
def load_agent(item, before_model_callback):
print(f"\nload_agent item: {item['name']}", item)
_agent = Agent(
name = item['name'],
model = item['model'],
instruction = item['instruction'],
description = item['description'],
before_model_callback=before_model_callback
)
return _agent
def load_agents(agents_dict:dict, before_model_callback):
print("\n\n\n######## load_agents\nagents_dict: ", agents_dict)
ret_agents = {}
for item in agents_dict:
ret_agents[item['name']] = load_agent(item, before_model_callback)
# ret_agents.append(load_agent(item))
print("\n\n\n@@@@@@@@@@@@@@@@@@@@@\nload_agents\nret_agents.greeting_agent: ", ret_agents['greeting_agent'])
print("\n\n\n@@@@@@@@@@@@@@@@@@@@@\nload_agents\nret_agents[farewell_agent]: ", ret_agents['farewell_agent'])
return ret_agents
def load_root_agent(_root_agent, before_model_callback):
ret_agent = load_agent(_root_agent, before_model_callback)
print("\n\n\n######## load_root_agent\nret_agent: ", ret_agent)
return ret_agent
|