我们关于 Google Agent Development Kit (ADK) 的探索之旅已经来到了第十篇。在前九篇文章中，我们从 ADK 的基础概念、环境搭建，到单个智能体的能力构建、工具集成，再到状态管理，一步步深入。今天，我们将迈入 ADK 最激动人心的领域之一：工作流代理（Workflow Agents）。

正如我们在现实世界中处理复杂任务一样，单一的行动往往不足以达成目标。AI 任务也一样，它常常需要一系列精心编排的步骤，可能包含条件的判断、重复的迭代，甚至并行的处理。工作流代理正是 ADK 为我们提供的强大武器，它允许我们像编排乐章一样，将多个智能体（Agent）或原子任务组合起来，形成一个能够独立思考、协作完成复杂目标的超级智能体。

今天，我将带大家深度解析工作流代理的核心执行模式：顺序执行、循环执行与并行执行。掌握了这三种模式，你将能够驾驭任何复杂的 AI 任务，让你的智能体系统真正“活”起来。

一、工作流代理：复杂任务的智能编舞师

在 ADK 中，一个工作流代理（Workflow Agent）本身也是一个智能体，但它的特殊之处在于，它不直接执行某个原子任务，而是作为协调者（Orchestrator），负责定义和管理一系列子智能体或步骤的执行顺序、条件和逻辑。你可以把它想象成一个项目经理，将一个大项目分解成多个小任务，然后指派给不同的团队成员（子智能体），并确保这些任务按照正确的流程完成。

工作流代理的优势：

1. 任务分解与模块化： 将复杂任务拆解为可管理、可复用的子任务。

1. 状态管理与上下文传递： 确保信息在不同步骤之间顺畅流动，维持任务的上下文。

1. 流程可视化与调试： 工作流的图结构使其逻辑一目了然，便于理解和问题排查。

1. 弹性与鲁棒性： 能够处理异常情况，并根据需要调整执行路径。

ADK 的工作流通常通过定义一系列的“状态”（States）和“转换”（Transitions）来构建，形成一个有向图。每个状态可以关联一个智能体、一个工具调用，或者一个决策点。

二、顺序工作流代理示例 - 智能文档处理系统

展示了如何使用Google ADK的SequentialAgent创建一个智能文档处理系统。该系统按照预定义的顺序执行四个步骤：内容提取、内容分析、摘要生成和格式化输出。

工作流程

原始文档 → 内容提取 → 内容分析 → 摘要生成 → 格式化输出

步骤详解

1. 内容提取代理 (ContentExtractorAgent)

• 从原始文档中提取关键信息

• 识别标题、要点、实体等结构化数据

• 使用extract_key_information工具

1. 内容分析代理 (ContentAnalyzerAgent)

• 分析文档的主题和核心观点

• 评估内容的逻辑结构和质量

• 识别目标受众和用途

• 摘要生成代理 (SummaryGeneratorAgent)

• 基于分析结果生成简洁摘要
• 突出关键信息和实用价值

• 控制摘要长度和可读性

• 格式化代理 (FormatterAgent)

• 将所有处理结果格式化为最终报告
• 使用format_final_document工具

• 生成结构化的输出文档

特点

• 确定性执行：严格按照预定义顺序执行

• 状态传递：每个代理的输出作为下一个代理的输入

• 工具集成：使用自定义工具处理特定任务

• 错误处理：任何步骤失败都会影响整个流程

代码结构

__init__.py

from . import agent

agent.py文件

"""
顺序工作流代理示例 - 智能文档处理系统
演示如何使用SequentialAgent按顺序执行多个代理来处理文档
"""


from google.adk.agents.llm_agent import LlmAgent
from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.tool_context import ToolContext
import os
from google.adk.models.lite_llm import LiteLlm


# API密钥配置
DEFAULT_DASHSCOPE_API_KEY = "sk-f227634bb561*******************77c"  # 请替换为您的实际密钥
DASHSCOPE_API_KEY = os.environ.get("DASHSCOPE_API_KEY", DEFAULT_DASHSCOPE_API_KEY)


# 常量定义
APP_NAME = "sequential_document_processor"
USER_ID = "dev_user_01"
GEMINI_MODEL = LiteLlm(
    model="openai/qwen-turbo",  # 使用通义千问Turbo模型
    api_key=DASHSCOPE_API_KEY,
    api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
)


# 状态键定义
STATE_RAW_DOCUMENT = "raw_document"
STATE_EXTRACTED_CONTENT = "extracted_content"
STATE_ANALYSIS_RESULT = "analysis_result"
STATE_SUMMARY = "summary"
STATE_FINAL_OUTPUT = "final_output"


# 示例文档内容
SAMPLE_DOCUMENT = """
人工智能技术发展报告


随着深度学习和大语言模型的快速发展，人工智能技术正在各个领域产生深远影响。
本报告分析了当前AI技术的主要趋势和应用场景。


主要发现：
1. 生成式AI在内容创作领域表现突出
2. 多模态AI模型能够处理文本、图像、音频等多种数据类型
3. AI代理系统在自动化任务执行方面展现巨大潜力


技术挑战：
- 模型的可解释性和透明度
- 数据隐私和安全问题
- 计算资源的高消耗


未来展望：
AI技术将继续向更加智能化、个性化的方向发展，预计在教育、医疗、金融等领域
将有更多突破性应用。同时，需要建立完善的AI治理框架来确保技术的负责任发展。


结论：
人工智能技术正处于快速发展期，机遇与挑战并存。企业和研究机构需要在技术创新
和风险管控之间找到平衡点，推动AI技术的健康发展。
"""


# 工具函数定义
def extract_key_information(tool_context: ToolContext):
    """从原始文档中提取关键信息的工具"""
    print(f"[Tool] extract_key_information called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 从用户输入或会话状态中获取文档内容
    raw_doc = session_state.get(STATE_RAW_DOCUMENT, "")
    
    # 如果状态中没有文档，尝试从用户输入中获取
    if not raw_doc:
        # 尝试从用户内容中获取文本
        user_content = tool_context._invocation_context.user_content
        if user_content and hasattr(user_content, 'parts'):
            # 从Content对象的parts中提取文本
            for part in user_content.parts:
                if hasattr(part, 'text') and part.text:
                    raw_doc = part.text
                    break
        elif user_content:
            raw_doc = str(user_content)
    
    # 如果还是没有，使用示例文档
    if not raw_doc:
        raw_doc = SAMPLE_DOCUMENT
    
    # 保存到状态中
    session_state[STATE_RAW_DOCUMENT] = raw_doc
    
    # 这里可以集成真实的文档解析库，如PyPDF2, python-docx等
    extracted_info = {
        "title": "从文档中提取的标题",
        "key_points": ["要点1", "要点2", "要点3"],
        "entities": ["实体1", "实体2"],
        "word_count": len(raw_doc.split()) if raw_doc else 0,
        "document_preview": raw_doc[:200] + "..." if len(raw_doc) > 200 else raw_doc
    }
    
    session_state[STATE_EXTRACTED_CONTENT] = extracted_info
    return extracted_info


def format_final_document(tool_context: ToolContext):
    """格式化最终文档的工具"""
    print(f"[Tool] format_final_document called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    summary = session_state.get(STATE_SUMMARY, "")
    analysis = session_state.get(STATE_ANALYSIS_RESULT, "")
    
    formatted_output = f"""
# 文档处理报告


## 摘要
{summary}


## 详细分析
{analysis}


## 处理时间
{session_state.get('processing_timestamp', 'N/A')}


---
*本报告由ADK顺序工作流代理自动生成*
"""
    
    session_state[STATE_FINAL_OUTPUT] = formatted_output
    
    # 设置结束调用标志
    tool_context._invocation_context.end_invocation = True
    
    # 返回明确的完成信号
    return {
        "status": "SUCCESS",
        "message": "文档格式化已完成，任务结束",
        "formatted_document": formatted_output,
        "task_completed": True
    }


# 代理定义
# 转换为ADK工具格式
extract_key_information_tool = FunctionTool(extract_key_information)
# 步骤1: 内容提取代理
content_extractor_agent = LlmAgent(
    name="ContentExtractorAgent",
    model=GEMINI_MODEL,
    instruction=f"""你是一个专业的文档内容提取专家。
    
    你的任务是分析用户提供的文档内容，并使用extract_key_information工具提取关键信息。
    
    用户输入的内容将作为文档进行处理。请调用extract_key_information工具来提取文档的关键信息，包括：
    - 标题和主题
    - 关键要点
    - 重要实体
    - 文档统计信息
    
    提取完成后，简要说明提取的内容类型和数量。
    """,
    description="负责从原始文档中提取关键信息和结构化数据",
    tools=[extract_key_information_tool],
    include_contents='none'
)


# 步骤2: 内容分析代理
content_analyzer_agent = LlmAgent(
    name="ContentAnalyzerAgent",
    model=GEMINI_MODEL,
    instruction=f"""你是一个专业的内容分析师。
    
    基于之前提取的内容信息，请进行深入分析：
    
    提取的内容：{{extracted_content}}
    
    请分析以下方面：
    1. 文档的主要主题和核心观点
    2. 内容的逻辑结构和组织方式
    3. 关键信息的重要性排序
    4. 文档的目标受众和用途
    5. 内容的完整性和质量评估
    
    请提供结构化的分析结果，包括具体的观察和建议。
    """,
    description="分析提取的内容，识别主题、结构和关键点",
    output_key=STATE_ANALYSIS_RESULT,
    include_contents='none'
)


# 步骤3: 摘要生成代理
summary_generator_agent = LlmAgent(
    name="SummaryGeneratorAgent",
    model=GEMINI_MODEL,
    instruction=f"""你是一个专业的文档摘要生成专家。
    
    基于内容分析结果，生成一个简洁而全面的文档摘要：
    
    分析结果：{{analysis_result}}
    原始提取内容：{{extracted_content}}
    
    请生成一个包含以下要素的摘要：
    1. 文档核心主题（1-2句话）
    2. 主要观点和结论（3-5个要点）
    3. 关键数据或事实
    4. 实用价值和应用建议
    
    摘要应该：
    - 长度控制在200-300字
    - 语言简洁明了
    - 突出重点信息
    - 便于快速理解
    """,
    description="基于分析结果生成结构化的文档摘要",
    output_key=STATE_SUMMARY,
    include_contents='none'
)


# 转换为ADK工具格式
format_final_document_tool = FunctionTool(format_final_document)
# 步骤4: 格式化代理
formatter_agent = LlmAgent(
    name="FormatterAgent",
    model=GEMINI_MODEL,
    instruction=f"""你是一个专业的文档格式化专家。


    你的任务是调用format_final_document工具一次来格式化最终输出文档，然后立即结束。


    工作流程：
    1. 调用format_final_document工具
    2. 确认工具返回成功状态
    3. 立即结束，不要再次调用任何工具


    当前处理状态：
    - 摘要：{{summary}}
    - 分析结果：{{analysis_result}}


    重要：只调用工具一次，收到工具响应后立即结束任务。
    """,
    description="将所有处理结果格式化为最终的输出文档",
    tools=[format_final_document_tool],
    include_contents='none'
)


# 创建顺序工作流代理
root_agent = SequentialAgent(
    name=APP_NAME,
    description="智能文档处理系统 - 按顺序执行内容提取、分析、摘要生成和格式化",
    sub_agents=[
        content_extractor_agent,
        content_analyzer_agent,
        summary_generator_agent,
        formatter_agent
    ]
)


# 用于测试的辅助函数
def create_test_session_with_document(session_service, document_content: str):
    """创建包含测试文档的会话"""
    session = session_service.create_session(APP_NAME, USER_ID)
    session.state[STATE_RAW_DOCUMENT] = document_content
    session.state['processing_timestamp'] = "2024-01-15 10:30:00"
    return session

采用adk web进行运行之后截图如下

三、循环执行：重复与迭代的艺术

下面将使用Google ADK的LoopAgent创建一个代码质量优化系统。该系统通过迭代的方式不断分析、建议、重构和检查代码，直到达到预设的质量标准或达到最大迭代次数。

工作流程

初始化 → [分析 → 建议 → 重构 → 检查] → 循环直到质量达标

循环步骤详解

1. 代码分析代理 (CodeAnalyzerAgent)

• 分析当前代码的质量指标

• 使用analyze_code_quality工具评估复杂度、可读性等

• 生成详细的质量报告

1. 优化建议代理 (OptimizationAdvisorAgent)

• 基于分析报告提供具体优化建议

• 包含代码结构、性能、可读性等方面的改进

• 当质量达标时发出完成信号

1. 代码重构代理 (CodeRefactorAgent)

• 根据优化建议重构代码

• 使用save_optimization_step工具记录优化历史

• 保持代码功能不变的前提下提升质量

1. 质量检查代理 (QualityCheckerAgent)

• 检查优化结果是否达到预期

• 决定是否继续下一轮迭代

• 使用exit_optimization_loop工具结束循环

特点

• 迭代改进：通过多轮优化逐步提升代码质量

• 智能终止：基于质量评分和建议内容自动决定何时停止

• 历史追踪：记录每次优化的详细信息

• 安全限制：设置最大迭代次数防止无限循环

终止条件

循环会在以下情况下终止：

1. 质量达标：质量评分达到8.0以上且建议中包含"质量已达标"

1. 最大迭代：达到最大迭代次数（5次）

1. 手动终止：质量检查代理调用退出工具

## 代码结构

10agent_loop_workflow/

├── __init__.py # 包初始化

├── agent.py # 主要代理定义

└── README.md # 说明文档

from . import agent

#agent.py

#agent.py
"""
循环工作流代理示例 - 代码质量优化系统
演示如何使用LoopAgent进行迭代的代码质量改进
"""


from google.adk.agents.llm_agent import LlmAgent
from google.adk.agents.loop_agent import LoopAgent
from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.tool_context import ToolContext


import os
from google.adk.models.lite_llm import LiteLlm


# API密钥配置
DEFAULT_DASHSCOPE_API_KEY = "sk-*********a95f38d77c"  # 请替换为您的实际密钥
DASHSCOPE_API_KEY = os.environ.get("DASHSCOPE_API_KEY", DEFAULT_DASHSCOPE_API_KEY)


# 常量定义
APP_NAME = "loop_code_optimizer"
USER_ID = "dev_user_01"
GEMINI_MODEL = LiteLlm(
                model="openai/qwen-turbo",  # 使用通义千问Turbo模型
                api_key=DASHSCOPE_API_KEY,
                api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )


# 状态键定义
STATE_ORIGINAL_CODE = "original_code"
STATE_CURRENT_CODE = "current_code"
STATE_ANALYSIS_REPORT = "analysis_report"
STATE_OPTIMIZATION_SUGGESTIONS = "optimization_suggestions"
STATE_QUALITY_SCORE = "quality_score"
STATE_ITERATION_COUNT = "iteration_count"
STATE_OPTIMIZATION_HISTORY = "optimization_history"


# 完成信号
COMPLETION_SIGNAL = "OPTIMIZATION_COMPLETE"


# 工具函数定义
def initialize_code_optimization(tool_context: ToolContext):
    """初始化代码优化的工具"""
    print(f"[Tool] initialize_code_optimization called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 强制防止重复初始化 - 如果已经调用过，直接返回
    call_count = session_state.get("init_call_count", 0)
    session_state["init_call_count"] = call_count + 1
    
    if call_count > 0:
        print(f"[Debug] 防止重复调用 - 第{call_count + 1}次调用，直接返回")
        return {
            "status": "already_initialized", 
            "message": f"已初始化（第{call_count + 1}次调用）",
            "code_length": len(session_state.get(STATE_CURRENT_CODE, ""))
        }
    
    # 尝试从多个来源获取用户输入
    user_input = ""
    
    # 方法1: 从invocation context的user_content获取
    if hasattr(tool_context._invocation_context, 'user_content') and tool_context._invocation_context.user_content:
        user_content = tool_context._invocation_context.user_content
        if hasattr(user_content, 'parts') and user_content.parts:
            # 从Content对象的parts中提取文本
            for part in user_content.parts:
                if hasattr(part, 'text') and part.text:
                    user_input = part.text.strip()
                    break
        elif user_content:
            user_input = str(user_content).strip()
    
    # 方法2: 从session state获取
    if not user_input:
        user_input = session_state.get("user_input", "")
    
    # 方法3: 从session的消息历史获取最新的用户消息
    if not user_input and hasattr(session, 'messages') and session.messages:
        for message in reversed(session.messages):
            if hasattr(message, 'author') and message.author == 'user':
                if hasattr(message, 'content') and message.content:
                    user_input = str(message.content).strip()
                    break
    
    print(f"[Debug] 获取到的用户输入: '{user_input}'")
    
    # 保存用户输入到session state以供后续使用
    if user_input:
        session_state["user_input"] = user_input
    
    current_code = session_state.get(STATE_CURRENT_CODE, "")
    original_code = session_state.get(STATE_ORIGINAL_CODE, "")
    
    # 检查用户输入是否是简单问候或非代码内容
    simple_greetings = ["你好", "hello", "hi", "嗨", "您好", "测试"]
    if user_input and user_input.lower().strip() in [g.lower() for g in simple_greetings]:
        # 对于简单问候，直接返回友好回应并标记为完成
        session_state["simple_greeting"] = True
        session_state[STATE_QUALITY_SCORE] = 10.0  # 设置高分以触发退出
        session_state["initialization_completed"] = True  # 标记初始化完成
        # 设置escalate来跳过后续的优化循环
        tool_context.actions.escalate = True
        print(f"[Debug] 检测到简单问候: '{user_input}', 设置escalate=True")
        return {
            "status": "simple_greeting_detected", 
            "message": "检测到简单问候，无需代码优化",
            "user_input": user_input
        }
    
    # 检查用户输入是否包含代码
    if user_input and (
        "def " in user_input or 
        "class " in user_input or 
        "import " in user_input or
        "from " in user_input or
        "print(" in user_input or
        "return " in user_input or
        len(user_input.split('\n')) > 3  # 多行输入可能是代码
    ):
        # 用户提供了代码，使用用户的代码
        session_state[STATE_ORIGINAL_CODE] = user_input
        session_state[STATE_CURRENT_CODE] = user_input
        session_state["initialization_completed"] = True  # 标记初始化完成
        # 不设置escalate，让流程继续到优化循环
        print(f"[Debug] 检测到用户代码，准备进入优化循环")
        return {"status": "initialized_with_user_code", "code_length": len(user_input)}
    
    # 如果没有代码，使用示例代码
    if not current_code and not original_code:
        sample_code = '''
def calculate_total(items):
    total = 0
    for i in range(len(items)):
        if items[i] > 0:
            total = total + items[i]
        else:
            print("Negative value found")
    return total


def process_data(data):
    result = []
    for item in data:
        if item != None:
            if len(item) > 0:
                result.append(item.upper())
    return result


class DataProcessor:
    def __init__(self):
        self.data = []
    
    def add_item(self, item):
        self.data.append(item)
    
    def get_total(self):
        return calculate_total(self.data)
'''
        session_state[STATE_ORIGINAL_CODE] = sample_code
        session_state[STATE_CURRENT_CODE] = sample_code
        session_state["initialization_completed"] = True  # 标记初始化完成
        # 不设置escalate，让流程继续到优化循环
        print(f"[Debug] 使用示例代码，准备进入优化循环")
        return {"status": "initialized_with_sample", "code_length": len(sample_code)}
    
    # 如果有原始代码但没有当前代码，复制原始代码
    elif original_code and not current_code:
        session_state[STATE_CURRENT_CODE] = original_code
        session_state["initialization_completed"] = True  # 标记初始化完成
        return {"status": "initialized_from_original", "code_length": len(original_code)}
    
    # 如果都有，保持现状
    else:
        session_state["initialization_completed"] = True  # 标记初始化完成
        return {"status": "already_initialized", "code_length": len(current_code)}


def analyze_code_quality(tool_context: ToolContext):
    """分析代码质量的工具"""
    print(f"[Tool] analyze_code_quality called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 检查是否是简单问候
    if session_state.get("simple_greeting", False):
        return {
            "overall_score": 10.0,
            "metrics": {"greeting_response": 10.0},
            "iteration": 1,
            "issues_found": 0,
            "improvements_needed": False,
            "message": "简单问候，无需代码分析"
        }
    
    current_code = session_state.get(STATE_CURRENT_CODE, "")
    iteration = session_state.get(STATE_ITERATION_COUNT, 0)
    
    # 改进的质量评分逻辑 - 更快收敛
    # 模拟代码质量分析
    # 在实际应用中，这里可以集成pylint, flake8, sonarqube等工具
    base_scores = {
        "complexity_score": 6.0,
        "readability_score": 5.5,
        "maintainability_score": 6.2,
        "performance_score": 7.0,
        "security_score": 8.0
    }
    
    # 每次迭代显著提升分数，确保在2-3轮内达到8.0
    improvement_factor = 1.0 + (iteration * 0.4)  # 每轮提升40%
    quality_metrics = {
        key: min(10.0, score * improvement_factor) 
        for key, score in base_scores.items()
    }
    
    overall_score = sum(quality_metrics.values()) / len(quality_metrics)
    
    analysis_report = {
        "overall_score": round(overall_score, 2),
        "metrics": quality_metrics,
        "iteration": iteration + 1,
        "issues_found": max(0, 3 - iteration),  # 减少发现的问题数
        "improvements_needed": overall_score < 8.0
    }
    
    session_state[STATE_ANALYSIS_REPORT] = analysis_report
    session_state[STATE_QUALITY_SCORE] = overall_score
    session_state[STATE_ITERATION_COUNT] = iteration + 1
    
    print(f"[Quality Analysis] Iteration {iteration + 1}, Score: {overall_score:.2f}")
    
    return analysis_report


def exit_optimization_loop(tool_context: ToolContext):
    """退出优化循环的工具"""
    print(f"[Tool] exit_optimization_loop called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 检查是否是简单问候
    if session_state.get("simple_greeting", False):
        user_input = session_state.get("user_input", "")
        tool_context.actions.escalate = True
        return {
            "status": "greeting_completed", 
            "message": f"已回应用户问候：{user_input}",
            "final_score": 10.0
        }
    
    # 记录优化历史
    history = session_state.get(STATE_OPTIMIZATION_HISTORY, [])
    final_score = session_state.get(STATE_QUALITY_SCORE, 0)
    iteration = session_state.get(STATE_ITERATION_COUNT, 0)
    
    completion_reason = "Quality threshold reached"
    if iteration >= 3:
        completion_reason = "Maximum iterations reached"
    elif final_score >= 8.0:
        completion_reason = "Quality threshold reached"
    
    history.append({
        "final_iteration": iteration,
        "final_score": final_score,
        "completion_reason": completion_reason
    })
    
    session_state[STATE_OPTIMIZATION_HISTORY] = history
    tool_context.actions.escalate = True
    
    print(f"[Exit Loop] Reason: {completion_reason}, Final Score: {final_score:.2f}")
    
    return {"status": "optimization_completed", "final_score": final_score, "reason": completion_reason}


def save_optimization_step(tool_context: ToolContext):
    """保存优化步骤的工具"""
    print(f"[Tool] save_optimization_step called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    history = session_state.get(STATE_OPTIMIZATION_HISTORY, [])
    current_code = session_state.get(STATE_CURRENT_CODE, "")
    suggestions = session_state.get(STATE_OPTIMIZATION_SUGGESTIONS, "")
    score = session_state.get(STATE_QUALITY_SCORE, 0)
    iteration = session_state.get(STATE_ITERATION_COUNT, 0)
    
    step_record = {
        "iteration": iteration,
        "quality_score": score,
        "suggestions_applied": suggestions[:100] + "..." if len(suggestions) > 100 else suggestions,
        "code_length": len(current_code),
        "timestamp": f"Step {iteration}"
    }
    
    history.append(step_record)
    session_state[STATE_OPTIMIZATION_HISTORY] = history
    
    return step_record




analyze_code_quality_tool = FunctionTool(analyze_code_quality)
initialize_code_optimization_tool = FunctionTool(initialize_code_optimization)


# 代理定义


# 步骤1: 代码分析代理
code_analyzer_agent = LlmAgent(
    name="CodeAnalyzerAgent",
    model=GEMINI_MODEL,
    instruction=f"""你是一个专业的代码质量分析专家。


    请分析当前会话状态中的代码质量。
    
    请调用analyze_code_quality工具来分析代码质量，包括：
    - 代码复杂度
    - 可读性
    - 可维护性
    - 性能
    - 安全性
    
    分析完成后，请简要说明发现的主要问题和当前的质量评分。
    """,
    description="分析代码质量，识别需要改进的方面",
    tools=[analyze_code_quality_tool],
    include_contents='none'
)


# 步骤2: 优化建议代理
optimization_advisor_agent = LlmAgent(
    name="OptimizationAdvisorAgent",
    model=GEMINI_MODEL,
    instruction=f"""你是一个代码优化顾问。


    基于代码质量分析报告，提供具体的优化建议。
    请从会话状态中获取分析报告和当前代码信息。
    
    请提供以下方面的优化建议：
    1. 代码结构改进
    2. 性能优化建议
    3. 可读性提升
    4. 最佳实践应用
    5. 潜在bug修复
    
    建议应该：
    - 具体可操作
    - 优先级明确
    - 包含代码示例
    - 解释改进原因
    
    如果质量评分已经达到8.0以上，请在建议末尾添加"质量已达标，建议完成优化。"
    """,
    description="基于分析结果提供具体的代码优化建议",
    output_key=STATE_OPTIMIZATION_SUGGESTIONS,
    include_contents='none'
)


save_optimization_step_tool = FunctionTool(save_optimization_step)
# 步骤3: 代码重构代理
code_refactor_agent = LlmAgent(
    name="CodeRefactorAgent",
    model=GEMINI_MODEL,
    instruction=f"""你是一个代码重构专家。


    根据优化建议重构代码。
    请从会话状态中获取当前代码和优化建议。
    
    请根据建议重构代码，重点关注：
    1. 应用建议的改进措施
    2. 保持代码功能不变
    3. 提高代码质量
    4. 遵循编程最佳实践
    
    请先调用save_optimization_step工具保存当前优化步骤，然后输出重构后的完整代码。
    
    输出格式：
    ```python
    # 重构后的代码
    [重构后的完整代码]
    ```
    """,
    description="根据优化建议重构代码",
    output_key=STATE_CURRENT_CODE,
    tools=[save_optimization_step_tool],
    include_contents='none'
)


exit_optimization_loop_tool = FunctionTool(exit_optimization_loop)
# 步骤4: 质量检查代理
quality_checker_agent = LlmAgent(
    name="QualityCheckerAgent",
    model=GEMINI_MODEL,
    instruction=f"""你是一个代码质量检查员。


    检查优化结果并决定是否继续迭代。
    请从会话状态中获取分析报告、优化建议和当前质量评分。
    
    评估标准：
    1. 质量评分是否达到8.0以上
    2. 是否还有明显的改进空间
    3. 优化建议是否提到"质量已达标"
    4. 是否是简单问候（无需优化）
    
    决策规则：
    - 如果是简单问候（session中有simple_greeting=True），立即调用exit_optimization_loop工具
    - 如果质量评分 >= 8.0，调用exit_optimization_loop工具结束优化
    - 如果建议中包含"质量已达标"，调用exit_optimization_loop工具结束优化
    - 如果迭代次数 >= 3，调用exit_optimization_loop工具结束优化（防止无限循环）
    - 否则，说明需要继续下一轮优化
    
    重要：必须明确调用exit_optimization_loop工具来结束循环，不能只是说明需要结束。
    
    请根据当前情况做出决策。
    """,
    description="检查优化质量，决定是否继续迭代",
    tools=[exit_optimization_loop_tool],
    include_contents='none'
)


# 创建循环工作流代理
optimization_loop = LoopAgent(
    name="OptimizationLoop",
    description="代码质量优化循环 - 重复执行分析、建议、重构、检查直到质量达标",
    sub_agents=[
        code_analyzer_agent,
        optimization_advisor_agent,
        code_refactor_agent,
        quality_checker_agent
    ],
    max_iterations=5  # 最多5轮优化
)


# 初始化代理
initialization_agent = LlmAgent(
    name="InitializationAgent",
    model=GEMINI_MODEL,
    instruction="""你是初始化专家。


执行步骤：
1. 调用 initialize_code_optimization 工具
2. 根据工具返回的 status 字段回应：
   - "simple_greeting_detected": 回复"您好！"
   - "initialized_with_user_code": 回复"代码已接收，开始分析。"
   - "initialized_with_sample": 回复"使用示例代码演示。"
   - "already_initialized": 回复"系统已就绪。"
3. 最后说"初始化完成"


只调用工具一次，然后结束。""",
    description="初始化代码优化流程",
    tools=[initialize_code_optimization_tool]
)


# 创建完整的工作流
root_agent = SequentialAgent(
    name=APP_NAME,
    description="代码质量优化系统 - 使用循环工作流进行迭代优化",
    sub_agents=[
        initialization_agent,
        optimization_loop
    ]
)


# 用于测试的辅助函数
def create_test_session_with_code(session_service, code_content: str):
    """创建包含测试代码的会话"""
    session = session_service.create_session(APP_NAME, USER_ID)
    session.state[STATE_ORIGINAL_CODE] = code_content
    session.state[STATE_CURRENT_CODE] = code_content
    session.state[STATE_ITERATION_COUNT] = 0
    session.state[STATE_OPTIMIZATION_HISTORY] = []
    return session


# 示例代码内容
SAMPLE_CODE = '''
def calculate_total(items):
    total = 0
    for i in range(len(items)):
        if items[i] > 0:
            total = total + items[i]
        else:
            print("Negative value found")
    return total


def process_data(data):
    result = []
    for item in data:
        if item != None:
            if len(item) > 0:
                result.append(item.upper())
    return result


class DataProcessor:
    def __init__(self):
        self.data = []
    
    def add_item(self, item):
        self.data.append(item)
    
    def get_total(self):
        return calculate_total(self.data)
''' 

#.env文件

# 不同LLM提供商的API密钥

# 将此文件重命名为.env并根据需要添加你的密钥

# Google API密钥，用于Gemini模型

GOOGLE_GENAI_USE_VERTEXAI="False"

GOOGLE_API_KEY=AIzaSyDsZWKy************Y4tKmj0

QWEN_API_KEY = "sk-f227634bb561**************9a95f38d77c"

采用adk web运行

整个流程图如下：

四、并行执行：效率与多样性的追求

使用Google ADK的ParallelAgent创建一个多源信息聚合系统。该系统同时从新闻媒体、社交媒体、市场数据和技术趋势四个维度收集信息，然后将这些数据整合成一份综合分析报告。

工作流程

搜索主题输入 → [新闻收集 | 社交媒体分析 | 市场数据收集 | 技术趋势分析] → 数据聚合 → 综合报告

并行执行步骤

1. 新闻收集代理 (NewsCollectorAgent)

• 收集相关新闻文章和报道

• 分析新闻情感倾向和相关性

• 使用collect_news_data工具

1. 社交媒体分析代理 (SocialMediaAgent)

• 分析Twitter、Reddit、LinkedIn等平台数据

• 识别热门话题和影响者

• 使用collect_social_media_data工具

1. 市场数据代理 (MarketDataAgent)

• 收集股票表现和投资趋势

• 分析市场指标和风险评估

• 使用collect_market_data工具

1. 技术趋势代理 (TechTrendsAgent)

• 分析GitHub活动和研究论文

• 评估技术采用率和成熟度

• 使用collect_tech_trends_data工具

数据聚合阶段

1. 报告生成代理 (ReportGeneratorAgent)

• 整合所有并行收集的数据

• 生成多维度洞察和趋势预测

• 使用generate_comprehensive_report工具

特点

• 高效并行：四个数据源同时收集，大幅提升效率

• 多维度分析：从媒体、社交、市场、技术四个角度全面分析

• 智能聚合：自动识别跨数据源的关联性和一致性

• 实时性强：并行执行减少总体处理时间

并行优势

相比顺序执行，并行工作流具有以下优势：

• 时间效率：4个任务并行执行，理论上可节省75%的时间

• 资源利用：充分利用多核处理能力

• 独立性：各数据源收集互不影响，提高稳定性

可扩展性：易于添加新的数据源。

代码结构

__init__.py

from . import agent

#.env文件

# 不同LLM提供商的API密钥

# 将此文件重命名为.env并根据需要添加你的密钥

# Google API密钥，用于Gemini模型

GOOGLE_GENAI_USE_VERTEXAI="False"

GOOGLE_API_KEY=AIzaSyDsZWKy************Y4tKmj0

QWEN_API_KEY = "sk-f227634bb561**************9a95f38d77c"

agent.py文件

"""
并行工作流代理示例 - 多源信息聚合系统
演示如何使用ParallelAgent同时执行多个独立的信息收集任务
"""


from google.adk.agents.llm_agent import LlmAgent
from google.adk.agents.parallel_agent import ParallelAgent
from google.adk.agents.sequential_agent import SequentialAgent
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.tool_context import ToolContext
import json
import random
from datetime import datetime, timedelta
import os
from google.adk.models.lite_llm import LiteLlm


# API密钥配置
DEFAULT_DASHSCOPE_API_KEY = "sk-f227634bb5614dd8b4eb29a95f38d77c"  # 请替换为您的实际密钥
DASHSCOPE_API_KEY = os.environ.get("DASHSCOPE_API_KEY", DEFAULT_DASHSCOPE_API_KEY)




# 常量定义
APP_NAME = "parallel_info_aggregator"
USER_ID = "dev_user_01"
GEMINI_MODEL = GEMINI_MODEL = LiteLlm(
                model="openai/qwen-turbo",  # 使用通义千问Turbo模型
                api_key=DASHSCOPE_API_KEY,
                api_base="https://dashscope.aliyuncs.com/compatible-mode/v1"
    )


# 状态键定义
STATE_SEARCH_TOPIC = "search_topic"
STATE_NEWS_DATA = "news_data"
STATE_SOCIAL_DATA = "social_media_data"
STATE_MARKET_DATA = "market_data"
STATE_TECH_DATA = "tech_trends_data"
STATE_AGGREGATED_REPORT = "aggregated_report"
STATE_COLLECTION_TIMESTAMP = "collection_timestamp"


# 工具函数定义
def collect_news_data(tool_context: ToolContext):
    """收集新闻数据的工具"""
    print(f"[Tool] collect_news_data called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 防止重复调用
    call_key = f"news_data_collected_{tool_context.agent_name}"
    if session_state.get(call_key, False):
        print(f"[Debug] 新闻数据已收集，跳过重复调用")
        return session_state.get(STATE_NEWS_DATA, {})
    
    topic = session_state.get(STATE_SEARCH_TOPIC, "AI技术")
    
    # 模拟新闻数据收集
    # 在实际应用中，这里可以集成真实的新闻API，如NewsAPI, Google News等
    news_data = {
        "source": "新闻媒体",
        "topic": topic,
        "articles": [
            {
                "title": f"{topic}领域重大突破：新技术引领行业变革",
                "summary": "最新研究显示，该技术在多个应用场景中表现出色",
                "sentiment": "positive",
                "relevance_score": 0.92,
                "publish_time": "2小时前"
            },
            {
                "title": f"{topic}市场分析：投资热度持续上升",
                "summary": "分析师预测该领域将迎来新一轮增长",
                "sentiment": "positive",
                "relevance_score": 0.88,
                "publish_time": "4小时前"
            },
            {
                "title": f"{topic}应用挑战：监管政策待完善",
                "summary": "专家呼吁建立更完善的监管框架",
                "sentiment": "neutral",
                "relevance_score": 0.75,
                "publish_time": "6小时前"
            }
        ],
        "total_articles": 3,
        "average_sentiment": 0.7,
        "collection_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    
    session_state[STATE_NEWS_DATA] = news_data
    session_state[call_key] = True
    return news_data


def collect_social_media_data(tool_context: ToolContext):
    """收集社交媒体数据的工具"""
    print(f"[Tool] collect_social_media_data called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 防止重复调用
    call_key = f"social_data_collected_{tool_context.agent_name}"
    if session_state.get(call_key, False):
        print(f"[Debug] 社交媒体数据已收集，跳过重复调用")
        return session_state.get(STATE_SOCIAL_DATA, {})
    
    topic = session_state.get(STATE_SEARCH_TOPIC, "AI技术")
    
    # 模拟社交媒体数据收集
    # 在实际应用中，这里可以集成Twitter API, Reddit API等
    social_data = {
        "source": "社交媒体",
        "topic": topic,
        "platforms": {
            "twitter": {
                "mentions": 1250,
                "sentiment_score": 0.65,
                "trending_hashtags": [f"#{topic}", "#创新", "#技术"],
                "top_influencers": ["@tech_expert", "@ai_researcher"]
            },
            "reddit": {
                "discussions": 89,
                "upvotes": 3420,
                "comments": 567,
                "hot_topics": [f"{topic}应用", "技术讨论", "未来展望"]
            },
            "linkedin": {
                "professional_posts": 156,
                "engagement_rate": 0.78,
                "industry_insights": ["企业应用增长", "人才需求上升"]
            }
        },
        "overall_sentiment": "positive",
        "engagement_level": "high",
        "collection_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    
    session_state[STATE_SOCIAL_DATA] = social_data
    session_state[call_key] = True
    return social_data


def collect_market_data(tool_context: ToolContext):
    """收集市场数据的工具"""
    print(f"[Tool] collect_market_data called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 防止重复调用
    call_key = f"market_data_collected_{tool_context.agent_name}"
    if session_state.get(call_key, False):
        print(f"[Debug] 市场数据已收集，跳过重复调用")
        return session_state.get(STATE_MARKET_DATA, {})
    
    topic = session_state.get(STATE_SEARCH_TOPIC, "AI技术")
    
    # 模拟市场数据收集
    # 在实际应用中，这里可以集成股票API, 市场研究报告等
    market_data = {
        "source": "市场数据",
        "topic": topic,
        "stock_performance": {
            "related_stocks": ["NVDA", "GOOGL", "MSFT", "TSLA"],
            "average_change": "+2.3%",
            "market_cap_change": "+$15.2B",
            "trading_volume": "高于平均水平"
        },
        "investment_trends": {
            "vc_funding": "$2.1B (本季度)",
            "deal_count": 45,
            "average_deal_size": "$46.7M",
            "growth_rate": "+18% YoY"
        },
        "market_indicators": {
            "market_sentiment": "bullish",
            "volatility": "medium",
            "growth_forecast": "positive",
            "risk_level": "moderate"
        },
        "collection_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    
    session_state[STATE_MARKET_DATA] = market_data
    session_state[call_key] = True
    return market_data


def collect_tech_trends_data(tool_context: ToolContext):
    """收集技术趋势数据的工具"""
    print(f"[Tool] collect_tech_trends_data called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 防止重复调用
    call_key = f"tech_data_collected_{tool_context.agent_name}"
    if session_state.get(call_key, False):
        print(f"[Debug] 技术趋势数据已收集，跳过重复调用")
        return session_state.get(STATE_TECH_DATA, {})
    
    topic = session_state.get(STATE_SEARCH_TOPIC, "AI技术")
    
    # 模拟技术趋势数据收集
    # 在实际应用中，这里可以集成GitHub API, 技术论文数据库等
    tech_data = {
        "source": "技术趋势",
        "topic": topic,
        "github_activity": {
            "new_repositories": 234,
            "stars_growth": "+12.5K",
            "active_contributors": 1890,
            "popular_languages": ["Python", "JavaScript", "Go"]
        },
        "research_papers": {
            "new_publications": 67,
            "citation_growth": "+8.9%",
            "top_conferences": ["NeurIPS", "ICML", "ICLR"],
            "emerging_topics": ["多模态AI", "联邦学习", "可解释AI"]
        },
        "technology_adoption": {
            "enterprise_adoption": "75%",
            "startup_usage": "89%",
            "developer_interest": "high",
            "maturity_level": "growing"
        },
        "collection_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    }
    
    session_state[STATE_TECH_DATA] = tech_data
    session_state[call_key] = True
    return tech_data


def generate_comprehensive_report(tool_context: ToolContext):
    """生成综合报告的工具"""
    print(f"[Tool] generate_comprehensive_report called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 防止重复调用
    call_key = f"report_generated_{tool_context.agent_name}"
    if session_state.get(call_key, False):
        print(f"[Debug] 综合报告已生成，跳过重复调用")
        return session_state.get(STATE_AGGREGATED_REPORT, {})
    
    # 获取所有收集的数据
    news_data = session_state.get(STATE_NEWS_DATA, {})
    social_data = session_state.get(STATE_SOCIAL_DATA, {})
    market_data = session_state.get(STATE_MARKET_DATA, {})
    tech_data = session_state.get(STATE_TECH_DATA, {})
    topic = session_state.get(STATE_SEARCH_TOPIC, "AI技术")
    
    # 生成综合报告
    report = {
        "topic": topic,
        "report_title": f"{topic}多维度信息聚合报告",
        "generation_time": datetime.now().strftime("%Y-%m-%d %H:%M:%S"),
        "data_sources": 4,
        "executive_summary": {
            "overall_sentiment": "积极",
            "market_outlook": "看涨",
            "technology_maturity": "快速发展",
            "investment_activity": "活跃"
        },
        "key_insights": [
            f"{topic}在新闻媒体中获得广泛关注，整体情绪积极",
            "社交媒体讨论热度高，专业人士参与度显著",
            "市场表现强劲，投资活动活跃",
            "技术发展迅速，开源社区贡献度高"
        ],
        "data_summary": {
            "news_articles": news_data.get("total_articles", 0),
            "social_mentions": social_data.get("platforms", {}).get("twitter", {}).get("mentions", 0),
            "market_growth": market_data.get("investment_trends", {}).get("growth_rate", "N/A"),
            "tech_repositories": tech_data.get("github_activity", {}).get("new_repositories", 0)
        },
        "recommendations": [
            "继续关注技术发展动态",
            "考虑相关投资机会",
            "加强社区参与和品牌建设",
            "密切监控市场变化"
        ]
    }
    
    session_state[STATE_AGGREGATED_REPORT] = report
    session_state[call_key] = True
    return report


def initialize_search_topic(tool_context: ToolContext):
    """初始化搜索主题的工具"""
    print(f"[Tool] initialize_search_topic called by {tool_context.agent_name}")
    
    # 正确访问session的state
    session = tool_context._invocation_context.session
    session_state = session.state
    
    # 尝试从用户输入中提取主题
    user_input = ""
    
    # 方法1: 从invocation context的user_content获取
    if hasattr(tool_context._invocation_context, 'user_content') and tool_context._invocation_context.user_content:
        user_content = tool_context._invocation_context.user_content
        if hasattr(user_content, 'parts') and user_content.parts:
            for part in user_content.parts:
                if hasattr(part, 'text') and part.text:
                    user_input = part.text.strip()
                    break
        elif user_content:
            user_input = str(user_content).strip()
    
    # 方法2: 从session的消息历史获取最新的用户消息
    if not user_input and hasattr(session, 'messages') and session.messages:
        for message in reversed(session.messages):
            if hasattr(message, 'author') and message.author == 'user':
                if hasattr(message, 'content') and message.content:
                    user_input = str(message.content).strip()
                    break
    
    print(f"[Debug] 获取到的用户输入: '{user_input}'")
    
    # 如果用户输入包含明确的主题，使用用户指定的主题
    search_topic = "AI技术"  # 默认主题
    
    if user_input:
        # 简单的主题提取逻辑
        topic_keywords = {
            "人工智能": ["人工智能", "AI", "机器学习", "深度学习"],
            "区块链": ["区块链", "比特币", "加密货币", "数字货币"],
            "量子计算": ["量子计算", "量子", "量子技术"],
            "新能源汽车": ["新能源汽车", "电动汽车", "特斯拉", "电动车"],
            "元宇宙": ["元宇宙", "虚拟现实", "VR", "AR"],
            "生物技术": ["生物技术", "基因", "医疗", "制药"]
        }
        
        user_input_lower = user_input.lower()
        for topic, keywords in topic_keywords.items():
            if any(keyword.lower() in user_input_lower for keyword in keywords):
                search_topic = topic
                break
        
        # 如果没有匹配到预定义主题，但用户输入看起来像是一个主题
        if search_topic == "AI技术" and len(user_input) < 50 and not any(char in user_input for char in "。！？"):
            search_topic = user_input
    
    # 设置搜索主题
    session_state[STATE_SEARCH_TOPIC] = search_topic
    session_state[STATE_COLLECTION_TIMESTAMP] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    print(f"[Debug] 设置搜索主题为: '{search_topic}'")
    
    return {
        "search_topic": search_topic,
        "user_input": user_input,
        "status": "initialized"
    }


# 代理定义


# 初始化代理
initialization_agent = LlmAgent(
    name="InitializationAgent",
    model=GEMINI_MODEL,
    instruction="""你是多源信息聚合系统的初始化专家。


    你的任务是：
    1. 调用initialize_search_topic工具来设置搜索主题
    2. 根据用户输入确定要分析的主题
    3. 简要说明将要进行的分析内容
    
    请先调用工具，然后告诉用户将要分析哪个主题。
    """,
    description="初始化搜索主题，准备多源数据收集",
    tools=[FunctionTool(initialize_search_topic)]
)


# 并行数据收集代理1: 新闻收集代理
news_collector_agent = LlmAgent(
    name="NewsCollectorAgent",
    model=GEMINI_MODEL,
    instruction="""你是一个专业的新闻信息收集专家。


    任务：调用collect_news_data工具收集新闻数据，然后简要总结。
    
    步骤：
    1. 调用collect_news_data工具
    2. 根据收集结果简要总结新闻趋势
    3. 完成任务
    
    只调用工具一次，然后结束。
    """,
    description="收集和分析新闻媒体中的相关信息",
    tools=[FunctionTool(collect_news_data)]
)


# 并行数据收集代理2: 社交媒体分析代理
social_media_agent = LlmAgent(
    name="SocialMediaAgent",
    model=GEMINI_MODEL,
    instruction="""你是一个社交媒体趋势分析专家。


    任务：调用collect_social_media_data工具收集社交媒体数据，然后简要分析。
    
    步骤：
    1. 调用collect_social_media_data工具
    2. 根据收集结果分析社交媒体趋势
    3. 完成任务
    
    只调用工具一次，然后结束。
    """,
    description="分析社交媒体平台上的讨论和趋势",
    tools=[FunctionTool(collect_social_media_data)]
)


# 并行数据收集代理3: 市场数据代理
market_data_agent = LlmAgent(
    name="MarketDataAgent",
    model=GEMINI_MODEL,
    instruction="""你是一个市场数据分析专家。


    任务：调用collect_market_data工具收集市场数据，然后简要分析。
    
    步骤：
    1. 调用collect_market_data工具
    2. 根据收集结果分析市场趋势
    3. 完成任务
    
    只调用工具一次，然后结束。
    """,
    description="收集和分析市场数据和投资趋势",
    tools=[FunctionTool(collect_market_data)]
)


# 并行数据收集代理4: 技术趋势代理
tech_trends_agent = LlmAgent(
    name="TechTrendsAgent",
    model=GEMINI_MODEL,
    instruction="""你是一个技术趋势分析专家。


    任务：调用collect_tech_trends_data工具收集技术数据，然后简要分析。
    
    步骤：
    1. 调用collect_tech_trends_data工具
    2. 根据收集结果分析技术趋势
    3. 完成任务
    
    只调用工具一次，然后结束。
    """,
    description="收集和分析技术发展趋势和创新动态",
    tools=[FunctionTool(collect_tech_trends_data)]
)


# 创建并行数据收集代理
parallel_data_collection = ParallelAgent(
    name="ParallelDataCollection",
    description="并行收集多源信息 - 同时执行新闻、社交媒体、市场和技术数据收集",
    sub_agents=[
        news_collector_agent,
        social_media_agent,
        market_data_agent,
        tech_trends_agent
    ]
)


# 数据聚合和报告生成代理
report_generator_agent = LlmAgent(
    name="ReportGeneratorAgent",
    model=GEMINI_MODEL,
    instruction="""你是综合信息分析和报告生成专家。


    任务：调用generate_comprehensive_report工具生成综合报告，然后提供深度分析。
    
    步骤：
    1. 调用generate_comprehensive_report工具
    2. 基于报告提供多维度洞察分析
    3. 给出趋势预测和行动建议
    4. 完成任务
    
    只调用工具一次，然后结束。
    """,
    description="整合多源数据，生成综合分析报告",
    tools=[FunctionTool(generate_comprehensive_report)]
)


# 创建完整的工作流
root_agent = SequentialAgent(
    name=APP_NAME,
    description="多源信息聚合系统 - 并行收集多维度数据并生成综合报告",
    sub_agents=[
        initialization_agent,
        parallel_data_collection,
        report_generator_agent
    ]
)


# 用于测试的辅助函数
def create_test_session_with_topic(session_service, search_topic: str):
    """创建包含搜索主题的会话"""
    session = session_service.create_session(APP_NAME, USER_ID)
    session.state[STATE_SEARCH_TOPIC] = search_topic
    session.state[STATE_COLLECTION_TIMESTAMP] = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    return session


# 示例搜索主题
SAMPLE_TOPICS = [
    "人工智能",
    "区块链技术",
    "量子计算",
    "新能源汽车",
    "元宇宙",
    "生物技术"
] 

Python

整体流转图如下

五、组合运用：驾驭真正的复杂任务

在现实世界的 AI 任务中，我们很少会只用到单一的执行模式。真正的力量在于将顺序、循环和并行模式巧妙地组合起来。

例如：

• 智能客服场景：

1. 顺序： 用户提问 -> 意图识别。

1. 并行： 根据意图，同时进行知识库查询和历史对话上下文分析。

1. 循环： 如果知识库查询未能直接解决，进入多轮澄清循环，直到获得足够信息。

1. 顺序： 根据收集到的信息和澄清结果，生成最终回复。

1. 并行： 同时将回复发送给用户，并记录到日志系统，同时触发后续的用户满意度调研。

• AI 招聘流程：

1. 顺序： 接收简历 -> 简历解析。

1. 并行： 同时对简历进行关键词匹配、技能评分和背景调查（外部服务调用）。

1. 循环： 对通过初筛的候选人，进行多轮面试安排（循环发送邀请，直到面试完成）。

1. 顺序： 面试结果汇总 -> 最终录用决策。

ADK 的图式工作流定义天生就支持这种复杂的组合。你可以通过精心设计状态和转换，构建出高度灵活和智能的自动化流程。

六、未来展望与实践建议

掌握了顺序、循环和并行执行模式，你已经拥有了构建强大工作流代理的关键能力。但要真正驾驭复杂 AI 任务，还有一些需要注意的实践点：

• 细粒度分解： 尽量将任务分解为足够小的、原子性的智能体或步骤，这样更容易组合和复用。

• 清晰的状态管理： 确保每个智能体都能清晰地读取和写入工作流的状态，这是信息传递和逻辑判断的基础。

• 错误处理与回退： 考虑在何处引入错误处理机制，例如重试、跳过或通知人工干预。ADK 提供了 @error_handler 等机制来处理异常。

• 监控与日志： 为你的工作流代理配置详细的日志和监控，以便在运行时跟踪其行为，发现潜在问题。

• 可视化工具： ADK 常常会与可视化工具结合，帮助开发者直观地设计和调试复杂的工作流。

---

结语：

工作流代理是 ADK 的核心魅力所在，它将智能体从孤立的“执行者”提升为“协作者”，使我们能够构建出更庞大、更智能、更具韧性的 AI 系统。通过精心编排顺序、巧妙运用循环、高效利用并行，你将能够解锁 AI 解决复杂问题的巨大潜力。

这篇关于工作流代理的深度解析，希望能为你未来的 ADK 项目带来启发。

感谢您的阅读，我们下篇文章再见！