《智能体设计模式》之多智能体协作模式：分工协同突破单一智能体能力边界[译]

import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from langchain_google_genai import ChatGoogleGenerativeAI
 
def setup_environment():
   """
   Loads environment variables and checks for the required API key.
   加载环境变量并检查所需的 API 密钥
   """
   load_dotenv()
   if not os.getenv("GOOGLE_API_KEY"):
       raise ValueError("GOOGLE_API_KEY not found. Please set it in your .env file.")
 
def main():
   """
   Initializes and runs the AI crew for content creation using the latest Gemini model.
   使用 Gemini 2.0 Flash 模型初始化并运行用于内容创建的 AI 智能体「创作团队」
   """
   setup_environment()
 
   # Define the language model to use.
   # Updated to a model from the Gemini 2.0 series for better performance and features.
   # For cutting-edge (preview) capabilities, you could use "gemini-2.5-flash".
   # 定义要使用的语言模型
   # 更新为 Gemini 2.0 系列的模型以获得更好的性能和功能
   # 如需最新的特性，可以使用 "gemini-2.5-flash" 模型
   llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash")
 
   # Define Agents with specific roles and goals
   # 定义具体的角色和目标的智能体
   researcher = Agent(
       role='Senior Research Analyst',
       goal='Find and summarize the latest trends in AI.',
       backstory="You are an experienced research analyst with a knack for identifying key trends and synthesizing information.",
       verbose=True,
       allow_delegation=False,
   )
 
   writer = Agent(
       role='Technical Content Writer',
       goal='Write a clear and engaging blog post based on research findings.',
       backstory="You are a skilled writer who can translate complex technical topics into accessible content.",
       verbose=True,
       allow_delegation=False,
   )
 
   # Define Tasks for the agents
   # 为智能体定义任务
   research_task = Task(
       description="Research the top 3 emerging trends in Artificial Intelligence in 2024-2025. Focus on practical applications and potential impact.",
       expected_output="A detailed summary of the top 3 AI trends, including key points and sources.",
       agent=researcher,
   )
 
   writing_task = Task(
       description="Write a 500-word blog post based on the research findings. The post should be engaging and easy for a general audience to understand.",
       expected_output="A complete 500-word blog post about the latest AI trends.",
       agent=writer,
       context=[research_task],
   )
 
   # Create the Crew
   # 创建 Crew 实例
   blog_creation_crew = Crew(
       agents=[researcher, writer],
       tasks=[research_task, writing_task],
       process=Process.sequential,
       llm=llm,
       # verbose=2 # Set verbosity for detailed crew execution logs
       verbose=True  # 译者注：CrewAI 的 Crew 类在最新版本中，verbose 参数只接受布尔值（True 或 False），不再支持整数级别（0, 1, 2）
   )
 
   # Execute the Crew
   # 启动 Crew 实例
   print("## Running the blog creation crew with Gemini 2.0 Flash... ##")
   try:
       result = blog_creation_crew.kickoff()
       print("\n------------------\n")
       print("## Crew Final Output ##")
       print(result)
   except Exception as e:
       print(f"\nAn unexpected error occurred: {e}")
 
 
if __name__ == "__main__":
   main()

from google.adk.agents import LlmAgent, BaseAgent
from google.adk.agents.invocation_context import InvocationContext
from google.adk.events import Event
from typing import AsyncGenerator
 
# Correctly implement a custom agent by extending BaseAgent
# 通过继承 BaseAgent 类实现自定义智能体
class TaskExecutor(BaseAgent):
   """
   A specialized agent with custom, non-LLM behavior.
   一个自定义的非基于大语言模型的智能体
   """
   name: str = "TaskExecutor"
   description: str = "Executes a predefined task."
 
   async def _run_async_impl(self, context: InvocationContext) -> AsyncGenerator[Event, None]:
       """
       Custom implementation logic for the task.
       自定义任务的实现逻辑
       """
       # This is where your custom logic would go.
       # For this example, we'll just yield a simple event.
       # 这里是自定义逻辑所在
       # 在这个示例中，我们只是返回一个简单的事件
       yield Event(author=self.name, content="Task finished successfully.")
 
# Define individual agents with proper initialization
# LlmAgent requires a model to be specified.
# 定义具有明确功能的智能体
# LlmAgent 需要指定使用的模型
greeter = LlmAgent(
   name="Greeter",
   model="gemini-2.0-flash-exp",
   instruction="You are a friendly greeter."
)
 
# Instantiate our concrete custom agent
# 实例化自定义智能体
task_doer = TaskExecutor()
 
# Create a parent agent and assign its sub-agents
# The parent agent's description and instructions should guide its delegation logic.
# 创建一个父智能体并声明其子智能体列表
# 父智能体的描述和指令应该能够指导其委派子智能体的逻辑
coordinator = LlmAgent(
   name="Coordinator",
   model="gemini-2.0-flash-exp",
   description="A coordinator that can greet users and execute tasks.",
   instruction="When asked to greet, delegate to the Greeter. When asked to perform a task, delegate to the TaskExecutor.",
   sub_agents=[
       greeter,
       task_doer
   ]
)
 
# The ADK framework automatically establishes the parent-child relationships.
# These assertions will pass if checked after initialization.
# ADK 框架会自动建立上述层级关系
# 如果在初始化后检查，这些断言将成立
assert greeter.parent_agent == coordinator
assert task_doer.parent_agent == coordinator
 
print("Agent hierarchy created successfully.")

import asyncio
from typing import AsyncGenerator
from google.adk.agents import LoopAgent, LlmAgent, BaseAgent
from google.adk.events import Event, EventActions
from google.adk.agents.invocation_context import InvocationContext
 
# Best Practice: Define custom agents as complete, self-describing classes.
# 最佳实践：将自定义智能体定义为完整的、自描述的类
class ConditionChecker(BaseAgent):
   """
   A custom agent that checks for a 'completed' status in the session state.
   一个自定义的智能体，用于检查会话是否为「completed」状态
   """
   name: str = "ConditionChecker"
   description: str = "Checks if a process is complete and signals the loop to stop."
 
   async def _run_async_impl(
       self, context: InvocationContext
   ) -> AsyncGenerator[Event, None]:
       """
       Checks state and yields an event to either continue or stop the loop.
       检查状态并产生事件以继续或停止循环
       """
       status = context.session.state.get("status", "pending")
       is_done = (status == "completed")
 
       if is_done:
           # Escalate to terminate the loop when the condition is met.
           # 满足条件时终止循环
           yield Event(author=self.name, actions=EventActions(escalate=True))
       else:
           # Yield a simple event to continue the loop.
           # 否则，发出简单事件以继续循环
           yield Event(author=self.name, content="Condition not met, continuing loop.")
 
# Correction: The LlmAgent must have a model and clear instructions.
# 说明：LlmAgent 必须指定使用的模型和清晰的指令
process_step = LlmAgent(
   name="ProcessingStep",
   model="gemini-2.0-flash-exp",
   instruction="You are a step in a longer process. Perform your task. If you are the final step, update session state by setting 'status' to 'completed'."
)
 
# The LoopAgent orchestrates the workflow.
# 使用 LoopAgent 编排工作流
poller = LoopAgent(
   name="StatusPoller",
   max_iterations=10,
   sub_agents=[
       process_step,
       ConditionChecker() # Instantiating the well-defined custom agent.
   ]
)
 
# This poller will now execute 'process_step'
# and then 'ConditionChecker'
# repeatedly until the status is 'completed' or 10 iterations
# have passed.
# 此轮询器现在将执行 'process_step' 智能体
# 然后执行 'ConditionChecker' 智能体
# 重复执行直到状态为 'completed' 或达到 10 次循环

from google.adk.agents import SequentialAgent, Agent
 
# This agent's output will be saved to session.state["data"]
# 该智能体的输出将保存到 session.state["data"]
step1 = Agent(name="Step1_Fetch", output_key="data")
 
# This agent will use the data from the previous step.
# We instruct it on how to find and use this data.
# 该智能体会使用上一步的输出
# 通过指令告诉它如何查找和使用上一步的输出
step2 = Agent(
   name="Step2_Process",
   instruction="Analyze the information found in state['data'] and provide a summary."
)
 
pipeline = SequentialAgent(
   name="MyPipeline",
   sub_agents=[step1, step2]
)
 
# When the pipeline is run with an initial input, Step1 will execute,
# its response will be stored in session.state["data"], and then
# Step2 will execute, using the information from the state as instructed.
# 当使用原始输入运行 pipeline 时，Step1 先执行，
# 它的响应会保存到 session.state["data"] 中，然后
# Step2 再执行，按照指令使用状态中的信息

from google.adk.agents import Agent, ParallelAgent
 
# It's better to define the fetching logic as tools for the agents
# For simplicity in this example, we'll embed the logic in the agent's instruction.
# In a real-world scenario, you would use tools.
# 在实际场景中最好将查询数据的逻辑定义为智能体的工具
# 为了简单起见，在这个示例中，我们将查询数据的逻辑放在了智能体的指令中
 
# Define the individual agents that will run in parallel
# 定义两个并行运行的子智能体：weather_fetcher 和 news_fetcher
weather_fetcher = Agent(
   name="weather_fetcher",
   model="gemini-2.0-flash-exp",
   instruction="Fetch the weather for the given location and return only the weather report.",
   output_key="weather_data"  # The result will be stored in session.state["weather_data"]
)
 
news_fetcher = Agent(
   name="news_fetcher",
   model="gemini-2.0-flash-exp",
   instruction="Fetch the top news story for the given topic and return only that story.",
   output_key="news_data"      # The result will be stored in session.state["news_data"]
)
 
# Create the ParallelAgent to orchestrate the sub-agents
# 使用 ParallelAgent 编排子智能体并行执行
data_gatherer = ParallelAgent(
   name="data_gatherer",
   sub_agents=[
       weather_fetcher,
       news_fetcher
   ]
)

from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool
from google.genai import types
 
# 1. A simple function tool for the core capability.
# This follows the best practice of separating actions from reasoning.
# 1. 一个简单的函数工具
# 遵循分离操作与推理的最佳实践
def generate_image(prompt: str) -> dict:
   """
   Generates an image based on a textual prompt.
   基于文本提示生成图像
 
   Args:
       prompt: A detailed description of the image to generate.
       要生成的图像的详细描述
   Returns:
       A dictionary with the status and the generated image bytes.
       包含状态和生成的图像字节的字典
   """
 
   print(f"TOOL: Generating image for prompt: '{prompt}'")
   # In a real implementation, this would call an image generation API.
   # For this example, we return mock image data.
   # 在实际实现中，这里将调用真实的图像生成 API
   # 但是为了简便起见，在这个示例中，我们仅返回模拟的图像数据
   mock_image_bytes = b"mock_image_data_for_a_cat_wearing_a_hat"
   return {
       "status": "success",
       # The tool returns the raw bytes, the agent will handle the Part creation.
       # 工具返回原始字节，智能体将处理 Part 创建
       "image_bytes": mock_image_bytes,
       "mime_type": "image/png"
   }
 
# 2. Refactor the ImageGeneratorAgent into an LlmAgent.
# It now correctly uses the input passed to it.
# 2. 将 ImageGeneratorAgent 重构为 LlmAgent
# 它现在能接收并使用传给它的参数了
image_generator_agent = LlmAgent(
   name="ImageGen",
   model="gemini-2.0-flash",
   description="Generates an image based on a detailed text prompt.",
   instruction=(
       "You are an image generation specialist. Your task is to take the user's request "
       "and use the `generate_image` tool to create the image. "
       "The user's entire request should be used as the 'prompt' argument for the tool. "
       "After the tool returns the image bytes, you MUST output the image."
   ),
   tools=[generate_image]
)
 
# 3. Wrap the corrected agent in an AgentTool.
# The description here is what the parent agent sees.
# 3. 将智能体封装在 AgentTool 中
# 这里的描述对父智能体是可见的
image_tool = agent_tool.AgentTool(
   agent=image_generator_agent,
   description="Use this tool to generate an image. The input should be a descriptive prompt of the desired image."
)
 
# 4. The parent agent remains unchanged. Its logic was correct.
# 4. 父智能体保持不变。
artist_agent = LlmAgent(
   name="Artist",
   model="gemini-2.0-flash",
   instruction=(
       "You are a creative artist. First, invent a creative and descriptive prompt for an image. "
       "Then, use the `ImageGen` tool to generate the image using your prompt."
   ),
   tools=[image_tool]
)