第 3 章：并行化

 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

import os
import asyncio
from typing import Optional

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import Runnable, RunnableParallel, RunnablePassthrough

# --- 配置 ---
# 确保环境变量已设置 API key（如 OPENAI_API_KEY）
try:
   llm: Optional[ChatOpenAI] = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
  
except Exception as e:
   print(f"初始化语言模型出错：{e}")
   llm = None

# --- 定义独立链 ---
# 三个链分别执行不同任务，可并行运行

summarize_chain: Runnable = (
   ChatPromptTemplate.from_messages([
       ("system", "请简明扼要地总结以下主题："),
       ("user", "{topic}")
   ])
   | llm
   | StrOutputParser()
)

questions_chain: Runnable = (
   ChatPromptTemplate.from_messages([
       ("system", "请针对以下主题生成三个有趣的问题："),
       ("user", "{topic}")
   ])
   | llm
   | StrOutputParser()
)

terms_chain: Runnable = (
   ChatPromptTemplate.from_messages([
       ("system", "请从以下主题中提取 5-10 个关键词，用逗号分隔："),
       ("user", "{topic}")
   ])
   | llm
   | StrOutputParser()
)

# --- 构建并行 + 汇总链 ---

# 1. 定义并行任务块，结果与原始 topic 一起传递到下一步
map_chain = RunnableParallel(
   {
       "summary": summarize_chain,
       "questions": questions_chain,
       "key_terms": terms_chain,
       "topic": RunnablePassthrough(),  # 传递原始 topic
   }
)

# 2. 定义最终汇总 prompt，整合并行结果
synthesis_prompt = ChatPromptTemplate.from_messages([
   ("system", """根据以下信息：
    摘要：{summary}
    相关问题：{questions}
    关键词：{key_terms}
    请综合生成完整答案。"""),
   ("user", "原始主题：{topic}")
])

# 3. 构建完整链，将并行结果直接传递给汇总 prompt，再由 LLM 和输出解析器处理
full_parallel_chain = map_chain | synthesis_prompt | llm | StrOutputParser()

# --- 运行链 ---
async def run_parallel_example(topic: str) -> None:
   """
   异步调用并行处理链，输出综合结果。

   Args:
       topic: 传递给 LangChain 的主题输入
   """
   if not llm:
       print("LLM 未初始化，无法运行示例。")
       return

   print(f"\n--- 并行 LangChain 示例，主题：'{topic}' ---")
   try:
       # `ainvoke` 的输入是单个 topic 字符串，
       # 会传递给 map_chain 中的每个 runnable
       response = await full_parallel_chain.ainvoke(topic)
       print("\n--- 最终响应 ---")
       print(response)
   except Exception as e:
       print(f"\n链执行出错：{e}")

if __name__ == "__main__":
   test_topic = "太空探索的历史"
   # Python 3.7+ 推荐用 asyncio.run 执行异步函数
   asyncio.run(run_parallel_example(test_topic))

 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

from google.adk.agents import LlmAgent, ParallelAgent, SequentialAgent
from google.adk.tools import google_search
GEMINI_MODEL="gemini-2.0-flash"

# --- 1. 定义并行运行的调研子智能体---

# 调研员 1：可再生能源
researcher_agent_1 = LlmAgent(
    name="RenewableEnergyResearcher",
    model=GEMINI_MODEL,
    instruction="""你是一名专注于能源领域的 AI 调研助手。
调研“可再生能源最新进展”，使用 Google Search 工具。
请简明总结关键发现（1-2 句），只输出摘要。
""",
    description="调研可再生能源。",
    tools=[google_search],
    output_key="renewable_energy_result"
)

# 调研员 2：电动汽车
researcher_agent_2 = LlmAgent(
    name="EVResearcher",
    model=GEMINI_MODEL,
    instruction="""你是一名专注于交通领域的 AI 调研助手。
调研“电动汽车技术最新进展”，使用 Google Search 工具。
请简明总结关键发现（1-2 句），只输出摘要。
""",
    description="调研电动汽车技术。",
    tools=[google_search],
    output_key="ev_technology_result"
)

# 调研员 3：碳捕集
researcher_agent_3 = LlmAgent(
    name="CarbonCaptureResearcher",
    model=GEMINI_MODEL,
    instruction="""你是一名专注于气候解决方案的 AI 调研助手。
调研“碳捕集方法现状”，使用 Google Search 工具。
请简明总结关键发现（1-2 句），只输出摘要。
""",
    description="调研碳捕集方法。",
    tools=[google_search],
    output_key="carbon_capture_result"
)

# --- 2. 创建并行智能体（并发运行调研员）---
parallel_research_agent = ParallelAgent(
    name="ParallelWebResearchAgent",
    sub_agents=[researcher_agent_1, researcher_agent_2, researcher_agent_3],
    description="并行运行多个调研智能体，收集信息。"
)

# --- 3. 定义合并智能体（并行智能体完成后运行）---
merger_agent = LlmAgent(
    name="SynthesisAgent",
    model=GEMINI_MODEL,
    instruction="""你是一名负责整合调研结果的 AI 助手。
你的任务是将以下调研摘要合成为结构化报告，并明确归属。每个主题用标题分段，确保内容连贯，仅整合输入摘要。

**注意：你的全部回答必须严格基于下方“输入摘要”，不得添加任何外部知识或细节。**

**输入摘要：**

*   **可再生能源：**
    {renewable_energy_result}
*   **电动汽车：**
    {ev_technology_result}
*   **碳捕集：**
    {carbon_capture_result}

**输出格式：**

## 可持续技术最新进展摘要

### 可再生能源发现
（基于 RenewableEnergyResearcher 的摘要，仅整合上述内容）

### 电动汽车发现
（基于 EVResearcher 的摘要，仅整合上述内容）

### 碳捕集发现
（基于 CarbonCaptureResearcher 的摘要，仅整合上述内容）

### 总结
（仅基于上述内容，简要总结 1-2 句）

只输出结构化报告，严格按上述格式，不加其他说明。
""",
    description="整合并行智能体的调研结果，生成结构化报告，仅基于输入内容。",
)

# --- 4. 创建串行智能体（总流程控制）---
sequential_pipeline_agent = SequentialAgent(
    name="ResearchAndSynthesisPipeline",
    sub_agents=[parallel_research_agent, merger_agent],
    description="协调并行调研与结果整合。"
)
root_agent = sequential_pipeline_agent