基于Google ADK构建Master Agent的架构设计及代理复用冲突解决方案咨询

一、架构设计的核心思路 #

核心原则：代理实例不能在多个父代理之间共享，每个父代理（Master Agent、SequenceAgent）要持有独立的代理实例，或者通过动态创建实例的方式避免复用冲突。

1. 明确代理职责边界 #

• 文本代理：只负责纯文本查询处理，作为Master Agent的直接子代理。
• SequenceAgent：内部封装转录、处理、文本转语音等专属代理实例，这些实例仅归SequenceAgent所有，不对外暴露给Master Agent。

2. Master Agent的路由逻辑 #

Master Agent先判断输入类型，再做路由：

• 纯文本输入：直接转发给文本代理处理。
• 音频/需要TTS的输入：转发给SequenceAgent，由其内部流水线完成处理。

二、避免代理复用冲突的具体方案 #

1. 用工厂模式动态创建代理实例 #

写个工厂函数，每次需要代理时生成新实例，保证每个父代理拿到的都是独立对象：

def create_transcription_agent():
    return TranscriptionAgent(config=transcription_config)

def create_tts_agent():
    return TTSAgent(config=tts_config)

• SequenceAgent初始化时，通过工厂创建内部代理：

class SequenceAgent(Agent):
    def __init__(self):
        self.transcription_agent = create_transcription_agent()
        self.processing_agent = create_processing_agent()
        self.tts_agent = create_tts_agent()
    
    def run(self, input_data):
        # 流水线执行流程：转录→处理→TTS
        transcribed_text = self.transcription_agent.process(input_data)
        processed_text = self.processing_agent.process(transcribed_text)
        return self.tts_agent.process(processed_text)

• Master Agent创建独立的文本代理和SequenceAgent实例：

class MasterAgent(Agent):
    def __init__(self):
        self.text_agent = TextAgent(config=text_config)
        self.sequence_agent = SequenceAgent()
    
    def route(self, input_data):
        if isinstance(input_data, str) and not is_audio_related(input_data):
            return self.text_agent.process(input_data)
        else:
            return self.sequence_agent.run(input_data)

2. 隔离代理的配置与状态 #

如果要复用代理的配置（比如模型参数、API密钥），可以把配置和运行时状态分离：

• 配置作为静态共享数据，所有代理实例都可以用。
• 运行时状态（会话上下文、临时缓存）每个实例独立维护，避免串扰。

class BaseAgent:
    # 共享配置
    shared_config = {
        "model": "gpt-4",
        "api_key": os.getenv("GOOGLE_API_KEY")
    }
    
    def __init__(self):
        # 独立状态
        self.session_context = {}
        self.temp_cache = {}

3. 禁止跨层级暴露内部代理 #

把SequenceAgent的内部代理设为私有成员，不对外提供访问接口，从根源上防止Master Agent直接引用这些实例，避免复用冲突。

三、调试验证要点 #

• 检查每个代理实例的唯一标识（比如Python里的id()），确认父代理持有的是不同实例。
• 分别测试纯文本、音频输入，验证路由逻辑是否正确，且不同代理的运行状态不会互相干扰。
• 监控代理的会话上下文，确保不同请求的临时数据不会被覆盖。

内容的提问来源于stack exchange，提问作者Adari Shanmukh