Agentic Design Patterns - Chapter 8_ Memory Management

第 8 章：记忆管理

有效的记忆管理对于智能 Agent 保留信息至关重要。与人类类似，Agent 需要不同类型的记忆才能高效运行。本章深入探讨记忆管理，特别关注 Agent 的即时（短期）和持久（长期）记忆需求。

在 Agent 系统中，记忆指代 Agent 保留并利用过去交互、观察和学习经验中信息的能力。这种能力使 Agent 能够做出明智决策、维护对话上下文并随时间持续改进。Agent 记忆通常分为两大主要类型：

  • 短期记忆（上下文记忆）： 类似于工作记忆，保存当前处理或最近访问的信息。对于使用大语言模型（LLM）的 Agent，短期记忆主要存在于上下文窗口中。该窗口包含最近消息、Agent 回复、工具使用结果以及当前交互中的 Agent 反思，所有这些都为 LLM 的后续响应和操作提供信息支撑。上下文窗口容量有限，制约了 Agent 可直接访问的近期信息量。高效的短期记忆管理涉及在有限空间内保留最相关信息，可能通过总结旧对话片段或突出关键细节等技术实现。具有”长上下文”窗口的模型出现仅扩展了这种短期记忆的容量，允许单次交互中保存更多信息。然而，这种上下文仍是临时的，会话结束后即丢失，且每次处理都可能成本高昂、效率低下。因此，Agent 需要独立的记忆类型来实现真正持久性、从过往交互中调用信息并建立持久知识库。
  • 长期记忆（持久记忆）： 作为 Agent 跨各种交互、任务或延长期间所需信息的存储库，类似于长期知识库。数据通常存储在 Agent 即时处理环境之外，常见于数据库、知识图谱或向量数据库中。在向量数据库中，信息被转换为数字向量存储，使 Agent 能够基于语义相似性而非精确关键字匹配检索数据，此过程称为语义搜索。当 Agent 需要长期记忆中的信息时，会查询外部存储、检索相关数据并将其集成到短期上下文中供即时使用，从而将先前知识与当前交互结合。

实际应用与用例

记忆管理对于 Agent 跟踪信息并随时间智能执行至关重要。这是 Agent 超越基本问答能力的必要条件。主要应用包括：

  • 聊天机器人与对话式 AI： 维护对话流程依赖短期记忆。聊天机器人需记住先前用户输入以提供连贯响应。长期记忆使聊天机器人能回忆用户偏好、过往问题或先前讨论，提供个性化和持续交互
  • 面向任务的 Agent： 管理多步骤任务的 Agent 需要短期记忆跟踪先前步骤、当前进度和总体目标。此类信息可能驻留于任务上下文或临时存储中。长期记忆对于访问不在即时上下文中的特定用户相关数据至关重要
  • 个性化体验： 提供定制交互的 Agent 利用长期记忆存储和检索用户偏好、过往行为和个人信息。这使 Agent 能调整其响应和建议
  • 学习与改进： Agent 可通过从过去交互中学习来提升性能。成功策略、错误和新信息存储于长期记忆中，促进未来适应。强化学习 Agent 以此方式存储学习策略或知识
  • 信息检索（RAG）： 设计用于回答问题的 Agent 访问知识库（即其长期记忆），通常在检索增强生成（RAG）中实现。Agent 检索相关文档或数据以指导其响应
  • 自主系统： 机器人或自动驾驶汽车需要记忆存储地图、路线、对象位置和学习行为。这涉及用于即时环境的短期记忆和用于通用环境知识的长期记忆

记忆使 Agent 能够维护历史、学习、个性化交互并管理复杂的时间依赖问题

实操代码：Google Agent Developer Kit (ADK) 中的记忆管理

Google Agent Developer Kit（ADK）提供了结构化的上下文和记忆管理方法，包含实际应用组件。深入理解 ADK 的 Session、State 和 Memory 对于构建需要保留信息的 Agent 至关重要。

与人类交互类似，Agent 需要能够回忆先前交流才能进行连贯自然的对话。ADK 通过三个核心概念及其关联服务简化了上下文管理。

与 Agent 的每次交互均可视为独特对话线程。Agent 可能需要访问早期交互数据。ADK 将其结构化如下：

  • Session（会话）： 独立聊天线程，记录该特定交互的消息和操作（Events），同时存储与该对话相关的临时数据（State）
  • State（状态）（session.state）： 存储在 Session 中的数据，包含仅与当前活动聊天线程相关的信息
  • Memory（记忆）： 来自各种过往聊天或外部来源信息的可搜索存储库，作为超出即时对话范围的数据检索资源

ADK 为构建复杂、有状态和上下文感知 Agent 所需的关键组件提供专用服务。SessionService 通过处理 Session 对象的启动、记录和终止来管理聊天线程，而 MemoryService 监督长期知识（Memory）的存储和检索

SessionService 和 MemoryService 均提供多种配置选项，允许用户根据应用需求选择存储方法。记忆选项适用于测试目的，但数据不会在重启后保留。对于持久存储和可扩展性，ADK 还支持数据库和基于云的服务

Session：跟踪每次聊天

ADK 中的 Session 对象旨在跟踪和管理单个聊天线程。在与 Agent 开始对话时，SessionService 生成一个 Session 对象，表示为 google.adk.sessions.Session。此对象封装了与特定对话线程相关的所有数据，包括唯一标识符（id、app_name、user_id）、作为 Event 对象的事件的时间顺序记录、用于会话特定临时数据的存储区域（称为 state）以及指示最后更新的时间戳（last_update_time）。开发人员通常通过 SessionService 间接与 Session 对象交互。SessionService 负责管理对话会话的生命周期，包括启动新会话、恢复先前的会话、记录会话活动（包括状态更新）、识别活动会话以及管理会话数据的删除。ADK 提供几种具有不同存储机制的 SessionService 实现，用于会话历史和临时数据，例如 InMemorySessionService，适合测试但不提供跨应用程序重启的数据持久性。

1
2
3
4
5
## 示例：使用 InMemorySessionService
## 这适用于不需要跨应用程序重启的数据持久性的本地开发和测试。
from google.adk.sessions import InMemorySessionService

session_service = InMemorySessionService()

然后是 DatabaseSessionService，如果您想可靠地保存到您管理的数据库。

1
2
3
4
5
6
7
8
9
## 示例：使用 DatabaseSessionService
## 这适用于需要持久存储的生产或开发。
## 您需要配置数据库 URL（例如，用于 SQLite、PostgreSQL 等）。
## 需要：pip install google-adk[sqlalchemy] 和数据库驱动程序（例如，PostgreSQL 的 psycopg2）
from google.adk.sessions import DatabaseSessionService

## 示例使用本地 SQLite 文件：
db_url = "sqlite:///./my_agent_data.db"
session_service = DatabaseSessionService(db_url=db_url)

此外，还有 VertexAiSessionService，它使用 Vertex AI 基础设施在 Google Cloud 上进行可扩展的生产。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
## 示例：使用 VertexAiSessionService
## 这适用于 Google Cloud Platform 上的可扩展生产，利用
## Vertex AI 基础设施进行会话管理。
## 需要：pip install google-adk[vertexai] 和 GCP 设置/身份验证
from google.adk.sessions import VertexAiSessionService

PROJECT_ID = "your-gcp-project-id"  # 替换为您的 GCP 项目 ID
LOCATION = "us-central1"  # 替换为您想要的 GCP 位置

## 与此服务一起使用的 app_name 应对应于 Reasoning Engine ID 或名称
REASONING_ENGINE_APP_NAME = "projects/your-gcp-project-id/locations/us-central1/reasoningEngines/your-engine-id"  # 替换为您的 Reasoning Engine 资源名称

session_service = VertexAiSessionService(project=PROJECT_ID, location=LOCATION)

## 使用此服务时，将 REASONING_ENGINE_APP_NAME 传递给服务方法：
## session_service.create_session(app_name=REASONING_ENGINE_APP_NAME, ...)
## session_service.get_session(app_name=REASONING_ENGINE_APP_NAME, ...)
## session_service.append_event(session, event, app_name=REASONING_ENGINE_APP_NAME)
## session_service.delete_session(app_name=REASONING_ENGINE_APP_NAME, ...)

选择适当的 SessionService 至关重要，因为它决定了 Agent 的交互历史和临时数据的存储方式及其持久性。

每次消息交换都涉及一个循环过程：接收消息，Runner 使用 SessionService 检索或建立 Session，Agent 使用 Session 的上下文（状态和历史交互）处理消息，Agent 生成响应并可能更新状态，Runner 将其封装为 Event，session_service.append_event 方法记录新事件并更新存储中的状态。然后 Session 等待下一条消息。理想情况下，当交互结束时使用 delete_session 方法终止会话。此过程说明了 SessionService 如何通过管理特定于 Session 的历史和临时数据来维护连续性。

State：Session 的草稿本

在 ADK 中，每个代表聊天线程的 Session 都包含 state 组件，类似于 Agent 在特定对话期间的临时工作记忆。虽然 session.events 记录整个聊天历史，但 session.state 存储和更新与活动聊天相关的动态数据点

从根本上讲，session.state 作为字典运行，以键值对形式存储数据。其核心功能是使 Agent 能够保留和管理对连贯对话至关重要的详细信息，如用户偏好、任务进度、增量数据收集或影响后续 Agent 操作的条件标志

状态结构包含字符串键与可序列化 Python 类型值的配对，包括字符串、数字、布尔值、列表及包含这些基本类型的字典。State 是动态的，在整个对话过程中不断演变。这些更改的持久性取决于配置的 SessionService

可使用键前缀定义数据范围和持久性来组织状态。无前缀的键是会话特定的

  • user: 前缀将数据与跨所有会话的用户 ID 关联
  • app: 前缀指定应用程序所有用户间共享的数据
  • temp: 前缀指示仅对当前处理轮次有效且不持久存储的数据

Agent 通过单个 session.state 字典访问所有状态数据。SessionService 处理数据检索、合并和持久性。应在通过 session_service.append_event() 向会话历史添加 Event 时更新状态。这确保了准确跟踪、在持久服务中的正确保存以及状态更改的安全处理

  1. 简单方法：使用 output_key（用于 Agent 文本回复）： 如果您只想将 Agent 的最终文本响应直接保存到状态中，这是最简单的方法。设置 LlmAgent 时，只需告诉它要使用的 output_key。Runner 看到这一点并在追加事件时自动创建必要的操作以将响应保存到状态。让我们看一个通过 output_key 演示状态更新的代码示例。
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
## 从 Google Agent Developer Kit (ADK) 导入必要的类
from google.adk.agents import LlmAgent
from google.adk.sessions import InMemorySessionService, Session
from google.adk.runners import Runner
from google.genai.types import Content, Part

## 定义带有 output_key 的 LlmAgent。
greeting_agent = LlmAgent(
    name="Greeter",
    model="gemini-2.0-flash",
    instruction="生成一个简短、友好的问候语。",
    output_key="last_greeting"
)

## --- 设置 Runner 和 Session ---
app_name, user_id, session_id = "state_app", "user1", "session1"
session_service = InMemorySessionService()
runner = Runner(
    agent=greeting_agent,
    app_name=app_name,
    session_service=session_service
)

session = session_service.create_session(
    app_name=app_name,
    user_id=user_id,
    session_id=session_id
)

print(f"初始状态：{session.state}")

## --- 运行 Agent ---
user_message = Content(parts=[Part(text="你好")])
print("\n--- 运行 agent ---")
for event in runner.run(
    user_id=user_id,
    session_id=session_id,
    new_message=user_message
):
    if event.is_final_response():
        print("Agent 已响应。")

## --- 检查更新的状态 ---
## 在 runner 完成处理所有事件*之后*正确检查状态。
updated_session = session_service.get_session(app_name, user_id, session_id)
print(f"\nAgent 运行后的状态：{updated_session.state}")

在幕后，Runner 看到您的 output_key，并在调用 append_event 时自动创建具有 state_delta 的必要操作。

  1. 标准方法：使用 EventActions.state_delta（用于更复杂的更新）： 对于需要执行更复杂操作的时候——例如一次更新多个键、保存不只是文本的内容、针对特定范围（如 user: 或 app:）或进行与 Agent 最终文本回复无关的更新——您将手动构建状态更改的字典（state_delta）并将其包含在要追加的 Event 的 EventActions 中。让我们看一个例子：
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
import time
from google.adk.tools.tool_context import ToolContext
from google.adk.sessions import InMemorySessionService

## --- 定义推荐的基于工具的方法 ---
def log_user_login(tool_context: ToolContext) -> dict:
    """
    在用户登录事件时更新会话状态。
    此工具封装了与用户登录相关的所有状态更改。
    参数：
        tool_context：由 ADK 自动提供，提供对会话状态的访问。
    返回：
        确认操作成功的字典。
    """
    # 通过提供的上下文直接访问状态。
    state = tool_context.state

    # 获取当前值或默认值，然后更新状态。
    # 这更加清晰并将逻辑共置。
    login_count = state.get("user:login_count", 0) + 1
    state["user:login_count"] = login_count
    state["task_status"] = "active"
    state["user:last_login_ts"] = time.time()
    state["temp:validation_needed"] = True

    print("从 `log_user_login` 工具内部更新了状态。")

    return {
        "status": "success",
        "message": f"已跟踪用户登录。总登录次数：{login_count}。"
    }

## --- 使用演示 ---
## 在真实应用程序中，LLM Agent 会决定调用此工具。
## 在这里，我们模拟直接调用以进行演示。

## 1. 设置
session_service = InMemorySessionService()
app_name, user_id, session_id = "state_app_tool", "user3", "session3"
session = session_service.create_session(
    app_name=app_name,
    user_id=user_id,
    session_id=session_id,
    state={"user:login_count": 0, "task_status": "idle"}
)

print(f"初始状态：{session.state}")

## 2. 模拟工具调用（在真实应用中，ADK Runner 会执行此操作）
## 我们仅为此独立示例手动创建 ToolContext。
from google.adk.tools.tool_context import InvocationContext
mock_context = ToolContext(
    invocation_context=InvocationContext(
        app_name=app_name, user_id=user_id, session_id=session_id,
        session=session, session_service=session_service
    )
)

## 3. 执行工具
log_user_login(mock_context)

## 4. 检查更新的状态
updated_session = session_service.get_session(app_name, user_id, session_id)
print(f"工具执行后的状态：{updated_session.state}")

## 预期输出将显示与"之前"情况相同的状态更改，
## 但代码组织明显更清晰
## 且更健壮。

此代码演示了一种基于工具的方法来管理应用程序中的用户会话状态。它定义了一个函数 log_user_login，充当工具。此工具负责在用户登录时更新会话状态。
该函数接受由 ADK 提供的 ToolContext 对象，以访问和修改会话的状态字典。在工具内部，它递增 *user:login_count，将 *task_status 设置为”active”，记录 user:last_login_ts（时间戳），并添加临时标志 temp:validation_needed。

代码的演示部分模拟了如何使用此工具。它设置了内存会话服务并创建了具有一些预定义状态的初始会话。然后手动创建 ToolContext 以模拟 ADK Runner 执行工具的环境。使用此模拟上下文调用 log_user_login 函数。最后，代码再次检索会话以显示状态已通过工具的执行更新。目标是展示将状态更改封装在工具中如何使代码比直接在工具外部操作状态更清晰、更有组织。

请注意，强烈不建议在检索会话后直接修改 session.state 字典，因为它会绕过标准事件处理机制。这种直接更改不会记录在会话的事件历史中，可能不会被选定的 SessionService 持久化，可能导致并发问题，并且不会更新时间戳等基本元数据。更新会话状态的推荐方法是在 LlmAgent 上使用 output_key 参数（专门用于 Agent 的最终文本响应）或在通过 session_service.append_event() 追加事件时在 EventActions.state_delta 中包含状态更改。session.state 应主要用于读取现有数据。

总而言之，在设计状态时，保持简单，使用基本数据类型，为键提供清晰的名称并正确使用前缀，避免深度嵌套，并始终使用 append_event 过程更新状态。

Memory：使用 MemoryService 的长期知识

在 Agent 系统中，Session 组件维护当前聊天历史（events）和特定于单个对话的临时数据（state）的记录。然而，为使 Agent 在多次交互中保留信息或访问外部数据，需要长期知识管理。这由 MemoryService 促进实现

1
2
3
4
5
6
## 示例：使用 InMemoryMemoryService
## 这适用于不需要跨应用程序重启数据持久性的本地开发和测试
## 应用停止时内存内容会丢失
from google.adk.memory import InMemoryMemoryService

memory_service = InMemoryMemoryService()

Session 和 State 可概念化为单个聊天会话的短期记忆，而由 MemoryService 管理的长期知识则充当持久且可搜索的存储库。此存储库可能包含来自多个过往交互或外部来源的信息。由 BaseMemoryService 接口定义的 MemoryService 为管理这种可搜索的长期知识建立了标准。其主要功能包括添加信息（涉及从会话中提取内容并使用 add_session_to_memory 方法存储）以及检索信息（允许 Agent 查询存储并使用 search_memory 方法接收相关数据）

ADK 提供了几种实现来创建这种长期知识存储。InMemoryMemoryService 提供适合测试目的的临时存储解决方案，但数据不会在应用重启后保留。对于生产环境，通常使用 VertexAiRagMemoryService。此服务利用 Google Cloud 的检索增强生成（RAG）服务，实现可扩展、持久和语义搜索功能（另请参阅第 14 章关于 RAG）

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
## 示例：使用 VertexAiRagMemoryService
## 这适用于 GCP 上的可扩展生产，利用
## Vertex AI RAG（检索增强生成）实现持久的、
## 可搜索的内存。
## 需要：pip install google-adk[vertexai]，GCP
## 设置/身份验证和 Vertex AI RAG Corpus。
from google.adk.memory import VertexAiRagMemoryService

## 您的 Vertex AI RAG Corpus 的资源名称
RAG_CORPUS_RESOURCE_NAME = "projects/your-gcp-project-id/locations/us-central1/ragCorpora/your-corpus-id"  # 替换为您的 Corpus 资源名称

## 检索行为的可选配置
SIMILARITY_TOP_K = 5  # 要检索的顶部结果数
VECTOR_DISTANCE_THRESHOLD = 0.7  # 向量相似度阈值

memory_service = VertexAiRagMemoryService(
    rag_corpus=RAG_CORPUS_RESOURCE_NAME,
    similarity_top_k=SIMILARITY_TOP_K,
    vector_distance_threshold=VECTOR_DISTANCE_THRESHOLD
)

## 使用此服务时，像 add_session_to_memory
## 和 search_memory 这样的方法将与指定的 Vertex AI
## RAG Corpus 交互。

实操代码：LangChain 和 LangGraph 中的记忆管理

在 LangChain 和 LangGraph 中，Memory 是创建智能自然对话应用的关键组件。它使 AI Agent 能够记住过去交互信息、从反馈中学习并适应用户偏好。LangChain 的记忆功能通过引用存储历史来丰富当前提示，然后记录最新交换供将来使用，从而提供此基础。随着 Agent 处理更复杂任务，这种能力对效率和用户满意度变得至关重要

短期记忆： 这是线程范围的，意味着它跟踪单个会话或线程内正在进行的对话。它提供即时上下文，但完整历史可能挑战 LLM 的上下文窗口，导致错误或性能下降。LangGraph 将短期记忆作为 Agent 状态的一部分管理，该状态通过检查点器持久化，允许随时恢复线程

长期记忆： 存储跨会话的用户特定或应用级数据，在对话线程间共享。它保存在自定义”命名空间”中，可在任何线程的任何时间调用。LangGraph 提供存储来保存和调用长期记忆，使 Agent 能无限期保留知识

LangChain 提供了多种工具管理对话历史，从手动控制到链内自动集成

ChatMessageHistory：手动记忆管理 对于在正式链外直接简单控制对话历史，ChatMessageHistory 类是理想选择。它允许手动跟踪对话交换

1
2
3
4
5
6
7
8
9
10
11
from langchain.memory import ChatMessageHistory

## 初始化历史对象
history = ChatMessageHistory()

## 添加用户和 AI 消息
history.add_user_message("我下周要去纽约。")
history.add_ai_message("太好了！这是一个很棒的城市。")

## 访问消息列表
print(history.messages)

ConversationBufferMemory：链的自动化记忆。要将记忆直接集成到链中，ConversationBufferMemory 是一个常见的选择。它保存对话的缓冲区并使其可用于您的提示词。其行为可以通过两个关键参数自定义：

  • memory_key：指定提示词中将保存聊天历史的变量名的字符串。默认为”history”。
  • return_messages：决定历史格式的布尔值。
    • 如果为 False（默认），它返回单个格式化字符串，这对于标准 LLM 是理想的。
    • 如果为 True，它返回消息对象列表，这是聊天模型的推荐格式。
1
2
3
4
5
6
7
8
9
10
from langchain.memory import ConversationBufferMemory

## 初始化内存
memory = ConversationBufferMemory()

## 保存对话轮次
memory.save_context({"input": "天气怎么样？"}, {"output": "今天是晴天。"})

## 将内存作为字符串加载
print(memory.load_memory_variables({}))

将此内存集成到 LLMChain 中允许模型访问对话的历史并提供上下文相关的响应

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
from langchain_openai import OpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate
from langchain.memory import ConversationBufferMemory

## 1. 定义 LLM 和提示词
llm = OpenAI(temperature=0)
template = """你是一个有帮助的旅行代理。
之前的对话：
{history}
新问题：{question}
响应："""
prompt = PromptTemplate.from_template(template)

## 2. 配置内存
## memory_key "history" 与提示词中的变量匹配
memory = ConversationBufferMemory(memory_key="history")

## 3. 构建链
conversation = LLMChain(llm=llm, prompt=prompt, memory=memory)

## 4. 运行对话
response = conversation.predict(question="我想预订航班。")
print(response)
response = conversation.predict(question="顺便说一下，我叫 Sam。")
print(response)
response = conversation.predict(question="我的名字是什么？")
print(response)

为了提高聊天模型的有效性，建议通过设置 return_messages=True 使用消息对象的结构化列表。

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
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.memory import ConversationBufferMemory
from langchain_core.prompts import (
    ChatPromptTemplate,
    MessagesPlaceholder,
    SystemMessagePromptTemplate,
    HumanMessagePromptTemplate,
)

## 1. 定义聊天模型和提示词
llm = ChatOpenAI()
prompt = ChatPromptTemplate(
    messages=[
        SystemMessagePromptTemplate.from_template("你是一个友好的助手。"),
        MessagesPlaceholder(variable_name="chat_history"),
        HumanMessagePromptTemplate.from_template("{question}")
    ]
)

## 2. 配置内存
## return_messages=True 对于聊天模型是必不可少的
memory = ConversationBufferMemory(memory_key="chat_history", return_messages=True)

## 3. 构建链
conversation = LLMChain(llm=llm, prompt=prompt, memory=memory)

## 4. 运行对话
response = conversation.predict(question="嗨，我是 Jane。")
print(response)
response = conversation.predict(question="你记得我的名字吗？")
print(response)

长期记忆类型：长期记忆允许系统在不同对话中保留信息，提供更深层次上下文和个性化。可分解为类似人类记忆的三种类型：

  • 语义记忆：记住事实 涉及保留特定事实和概念，如用户偏好或领域知识。用于基础 Agent 的响应，带来更个性化和相关的交互。此信息可作为持续更新的用户”配置文件”（JSON 文档）或作为单个事实文档的”集合”管理
  • 情景记忆：记住经历 涉及回忆过去事件或行动。对于 AI Agent，情景记忆通常用于记住如何完成任务。实践中常通过少样本示例提示实现，Agent 从过去成功交互序列中学习以正确执行任务
  • 程序记忆：记住规则 关于如何执行任务的记忆——Agent 的核心指令和行为，通常包含在其系统提示中。Agent 修改自身提示以适应和改进很常见。有效技术是”反思”，其中 Agent 被提示其当前指令和最近交互，然后被要求改进自身指令

下面是演示 Agent 如何使用反思来更新存储在 LangGraph BaseStore 中的程序记忆的伪代码

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
## 更新 agent 指令的节点
def update_instructions(state: State, store: BaseStore):
    namespace = ("instructions",)
    # 从存储中获取当前指令
    current_instructions = store.search(namespace)[0]

    # 创建提示词要求 LLM 反思对话
    # 并生成新的、改进的指令
    prompt = prompt_template.format(
        instructions=current_instructions.value["instructions"],
        conversation=state["messages"]
    )

    # 从 LLM 获取新指令
    output = llm.invoke(prompt)
    new_instructions = output['new_instructions']

    # 将更新的指令保存回存储
    store.put(("agent_instructions",), "agent_a", {"instructions": new_instructions})

## 使用指令生成响应的节点
def call_model(state: State, store: BaseStore):
    namespace = ("agent_instructions", )
    # 从存储中检索最新指令
    instructions = store.get(namespace, key="agent_a")[0]

    # 使用检索到的指令格式化提示词
    prompt = prompt_template.format(instructions=instructions.value["instructions"])
    # ... 应用程序逻辑继续

LangGraph 将长期记忆存储为存储中的 JSON 文档。每个记忆都在自定义命名空间（如文件夹）和不同的键（如文件名）下组织。这种分层结构允许轻松组织和检索信息。以下代码演示了如何使用 InMemoryStore 放置、获取和搜索记忆。

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
from langgraph.store.memory import InMemoryStore

## 真实嵌入函数的占位符
def embed(texts: list[str]) -> list[list[float]]:
    # 在真实应用程序中，使用适当的嵌入模型
    return [[1.0, 2.0] for _ in texts]

## 初始化内存存储。对于生产，使用数据库支持的存储。
store = InMemoryStore(index={"embed": embed, "dims": 2})

## 为特定用户和应用程序上下文定义命名空间
user_id = "my-user"
application_context = "chitchat"
namespace = (user_id, application_context)

## 1. 将记忆放入存储
store.put(
    namespace,
    "a-memory",  # 此记忆的键
    {
        "rules": [
            "用户喜欢简短、直接的语言",
            "用户只说英语和 python",
        ],
        "my-key": "my-value",
    },
)

## 2. 通过其命名空间和键获取记忆
item = store.get(namespace, "a-memory")
print("检索的项目：", item)

## 3. 在命名空间内搜索记忆，按内容过滤
## 并按与查询的向量相似度排序。
items = store.search(
    namespace,
    filter={"my-key": "my-value"},
    query="语言偏好"
)
print("搜索结果：", items)

Vertex Memory Bank

Memory Bank 是 Vertex AI Agent Engine 中的托管服务，为 Agent 提供持久长期内存。该服务使用 Gemini 模型异步分析对话历史以提取关键事实和用户偏好

此信息持久存储，按定义范围（如用户 ID）组织，并智能更新以整合新数据并解决矛盾。开始新会话时，Agent 通过完全数据调用或使用嵌入的相似性搜索检索相关记忆。此过程允许 Agent 跨会话维护连续性并根据回忆信息个性化响应

Agent 的 runner 与 VertexAiMemoryBankService 交互，后者首先初始化。此服务处理 Agent 对话期间生成记忆的自动存储。每个记忆标记有唯一 USER_ID 和 APP_NAME，确保将来准确检索

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
from google.adk.memory import VertexAiMemoryBankService

agent_engine_id = agent_engine.api_resource.name.split("/")[-1]
memory_service = VertexAiMemoryBankService(
    project="PROJECT_ID",
    location="LOCATION",
    agent_engine_id=agent_engine_id
)

session = await session_service.get_session(
    app_name=app_name,
    user_id="USER_ID",
    session_id=session.id
)

await memory_service.add_session_to_memory(session)

Memory Bank 提供与 Google ADK 的无缝集成，提供即开即用体验。对于使用其他 Agent 框架（如 LangGraph 和 CrewAI）的用户，Memory Bank 还通过直接 API 调用提供支持。演示这些集成的在线代码示例可供感兴趣读者使用

概览

是什么：：Agent 系统需要记住过去交互信息以执行复杂任务并提供连贯体验。没有记忆机制，Agent 是无状态的，无法维护对话上下文、从经验中学习或为用户个性化响应。这从根本上将其限制在简单一次性交互中，无法处理多步骤过程或不断变化的用户需求。核心问题是如何有效管理单个对话的即时临时信息和随时间收集的大量持久知识

为什么： 标准化解决方案是实现区分短期和长期存储的双组件记忆系统。短期的上下文记忆在 LLM 的上下文窗口内保存最近交互数据以维护对话流程。对于必须持久的信息，长期记忆解决方案使用外部数据库（通常是向量存储）实现高效语义检索。像 Google ADK 这样的 Agent 框架提供特定组件来管理此过程，如用于对话线程的 Session 和用于其临时数据的 State。专用的 MemoryService 用于与长期知识库交互，允许 Agent 检索相关的过去信息并将其纳入当前上下文

经验法则： 当 Agent 需要做的不仅仅是回答单个问题时使用此模式。对于必须在整个对话中维护上下文、跟踪多步骤任务进度或通过回忆用户偏好和历史来个性化交互的 Agent，这是必不可少的。每当 Agent 预期基于过去成功、失败或新获得信息进行学习或适应时，实现记忆管理

视觉摘要

图 1：内存管理设计模式

关键要点

快速回顾内存管理的核心要点：

  • 内存对于 Agent 跟踪事物、学习和个性化交互至关重要
  • 对话式 AI 依赖单个聊天中即时上下文的短期内存和跨多个会话持久知识的长期内存
  • 短期内存（即时信息）是临时的，通常受 LLM 上下文窗口或框架传递上下文方式的限制
  • 长期内存（持久信息）使用向量数据库等外部存储跨不同聊天保存信息，并通过搜索访问
  • 像 ADK 这样的框架具有特定部分，如 Session（聊天线程）、State（临时聊天数据）和 MemoryService（可搜索的长期知识）来管理内存
  • ADK 的 SessionService 处理聊天会话的整个生命周期，包括其历史（events）和临时数据（state）
  • ADK 的 session.state 是临时聊天数据的字典。前缀（user:、app:、temp:）指示数据归属位置及是否持久
  • 在 ADK 中，应通过在添加事件时使用 EventActions.state_delta 或 output_key 更新状态，而非直接更改状态字典
  • ADK 的 MemoryService 用于将信息放入长期存储并让 Agent 搜索，通常使用工具
  • LangChain 提供如 ConversationBufferMemory 的实用工具，自动将单个对话历史注入提示，使 Agent 能回忆即时上下文
  • LangGraph 通过使用存储来保存和检索跨不同用户会话的语义事实、情景经验甚至可更新程序规则，实现高级长期内存
  • Memory Bank 是托管服务，通过自动提取、存储和调用用户特定信息为 Agent 提供持久长期内存，在 Google 的 ADK、LangGraph 和 CrewAI 等框架中实现个性化持续对话

结论

本章深入探讨了 Agent 系统内存管理的重要工作，展示了短暂上下文与长期持久知识间的区别。我们讨论了这些内存类型的设置方式以及在构建更智能 Agent 时的应用场景。我们详细了解了 Google ADK 如何提供 Session、State 和 MemoryService 等特定组件来处理此过程。既然已介绍 Agent 如何记住事物（短期和长期），我们可以继续探讨它们如何学习和适应。下一个模式”学习和适应”关注 Agent 根据新经验或数据改变其思考、行动或所知的方式

参考文献

  1. ADK Memory, https://google.github.io/adk-docs/sessions/memory/
  2. LangGraph Memory, https://langchain-ai.github.io/langgraph/concepts/memory/
  3. Vertex AI Agent Engine Memory Bank, https://cloud.google.com/blog/products/ai-machine-learning/vertex-ai-memory-bank-in-public-preview

展开全文 >>

Agentic Design Patterns - Glossary

术语表

基础概念

提示词（Prompt）： 提示词是用户向 AI 模型提供的输入，通常以问题、指令或陈述形式呈现，旨在激发模型生成相应输出。提示词的质量与结构深度影响模型响应效果，使提示工程成为高效运用 AI 的核心技能。

上下文窗口（Context Window）： 上下文窗口指 AI 模型单次处理的最大 token 容量，涵盖输入内容与生成输出。此固定尺寸构成关键限制——超出窗口范围的信息将被忽略，而更大窗口则支持更复杂对话交互与文档分析能力。

上下文学习（In-Context Learning）： 上下文学习是 AI 通过提示中直接嵌入的示例掌握新任务的能力，无需额外训练过程。此强大特性使单一通用模型可即时适配海量特定任务场景。

零样本、单样本与少样本提示（Zero-Shot, One-Shot, & Few-Shot Prompting）： 此类提示技术通过提供零个、单个或少量任务示例引导模型生成响应。增加示例数量通常能提升模型对用户意图的把握精度，增强特定任务表现。

多模态（Multimodality）： 多模态指 AI 理解并处理跨数据类型信息（如文本、图像、音频）的能力。该特性支持更丰富类人化交互，例如图像描述或语音问答场景。

事实锚定（Grounding）： 事实锚定是将模型输出关联至可验证真实信息源的过程，旨在确保事实准确性并降低幻觉发生。常通过 RAG 等技术实现，提升 AI 系统可信度。

核心 AI 模型架构

Transformer： Transformer 构成多数现代 LLM 的基石神经网络架构。其核心创新自注意力机制能高效处理长文本序列，精准捕捉词汇间复杂关联。

循环神经网络（Recurrent Neural Network, RNN）： 循环神经网络是早于 Transformer 的基础架构。RNN 采用序列化信息处理方式，通过循环结构维持对历史输入的”记忆”，曾广泛应用于文本与语音处理任务。

专家混合（Mixture of Experts, MoE）： 专家混合为高效模型架构，通过”路由”网络动态筛选少量”专家”网络处理特定输入。该设计使模型在保持海量参数规模的同时，有效控制计算开销。

扩散模型（Diffusion Models）： 扩散模型是专精高质量图像生成的生成式模型。其工作原理是通过向数据注入随机噪声，继而训练模型精确逆转该过程，从而实现从随机初始状态生成新颖数据。

Mamba： Mamba 是采用选择性状态空间模型（Selective State Space Model, SSM）的新兴 AI 架构，具备超长上下文序列的高效处理能力。其选择性机制可聚焦相关信息同时滤除噪声，被视为 Transformer 的潜在替代方案。

LLM 开发生命周期

强大语言模型的开发遵循明确演进路径。初始阶段为预训练（Pre-training）——通过海量通用互联网文本数据集训练构建大规模基础模型，掌握语言规律、推理能力与世界知识。随后进入微调（Fine-tuning）阶段，通用模型在较小规模任务特定数据集上进一步训练，实现面向特定场景的能力适配。最终阶段为对齐（Alignment），调整专业化模型行为模式，确保输出兼具实用性、安全性且符合人类价值导向。

预训练技术（Pre-training Techniques）： 预训练是模型从海量数据中汲取通用知识的奠基阶段。该领域顶尖技术涵盖多样化的学习目标设定。最普遍的是因果语言建模（Causal Language Modeling, CLM），模型通过预测句中下一词汇进行学习。另一主流方法是掩码语言建模（Masked Language Modeling, MLM），模型负责还原文本中被刻意遮蔽的词汇。其他关键方法论包括：去噪目标（Denoising Objectives）——模型学习将受损输入修复至原始状态；对比学习（Contrastive Learning）——模型习得区分相似与相异数据片段的能力；以及下句预测（Next Sentence Prediction, NSP）——模型判断两句间是否存在逻辑连贯性。

微调技术（Fine-tuning Techniques）： 微调是使用专业化小规模数据集将通用预训练模型适配至特定任务的过程。最常用方法为监督微调（Supervised Fine-Tuning, SFT），模型在标注准确的输入输出配对示例上进行训练。指令微调（Instruction Tuning）作为流行变体，重点提升模型遵循用户指令的精准度。为提升过程效率，普遍采用参数高效微调（Parameter-Efficient Fine-Tuning, PEFT）方法，其中代表性技术包括 LoRA（低秩适应）——仅更新少量参数，及其内存优化版本 QLoRA。检索增强生成（Retrieval-Augmented Generation, RAG）作为补充技术，通过在微调或推理阶段连接外部知识源增强模型能力。

对齐与安全技术（Alignment & Safety Techniques）： 对齐是确保 AI 模型行为契合人类价值观与期望的过程，使其输出兼具帮助性与安全性。最突出的技术是基于人类反馈的强化学习（Reinforcement Learning from Human Feedback, RLHF），通过基于人类偏好训练的”奖励模型”指导 AI 学习过程，常辅以近端策略优化（Proximal Policy Optimization, PPO）等算法保障训练稳定性。近年涌现的简化替代方案包括直接偏好优化（Direct Preference Optimization, DPO）——规避独立奖励模型需求，以及卡尼曼-特沃斯基优化（Kahneman-Tversky Optimization, KTO）——进一步简化数据收集流程。为确保部署安全，护栏（Guardrails）作为终态安全层被广泛应用，实时过滤输出并阻断有害行为。

增强 AI Agent 能力

AI Agent 是能感知环境并采取自主行动以实现目标的智能系统。其效能通过强大的推理框架显著提升。

思维链（Chain of Thought, CoT）： 该提示技术引导模型在给出最终答案前逐步展示推理过程。此类”显式思考”机制常在复杂推理任务中产生更精确结果。

思维树（Tree of Thoughts, ToT）： 思维树是进阶推理框架，Agent 同步探索多条推理路径（如树枝分叉）。该框架支持 Agent 自评估不同思路质量，择优推进最 promising 路径，极大增强复杂问题解决效能。

ReAct（推理-行动框架，Reason and Act）： ReAct 是将推理与工具使用融合于循环流程的 Agent 框架。Agent 先进行”思考”确定行动方向，随后通过工具执行”行动”，并依据执行结果”观察”调整后续思考，该机制在复杂任务解决中表现卓越。

规划（Planning）： 指 Agent 将高层目标分解为系列可管理子任务的能力。Agent 据此制定执行计划并按序推进步骤，从而胜任复杂的多阶段任务。

深度研究（Deep Research）： 深度研究指 Agent 通过迭代式信息检索、发现整合与新问题识别，实现对主题的自主深入探索能力。这使得 Agent 能构建远超单次查询范围的 comprehensive 主题认知。

评审模型（Critique Model）： 评审模型是经专门训练的 AI 模型，用于审查、评估并对其他 AI 模型输出提供反馈。其扮演自动化评审者角色，协助识别错误、优化推理流程并确保最终输出符合预期质量标准。

展开全文 >>

Agentic Design Patterns - Chapter 7_ Multi-Agent Collaboration

第 7 章：多 Agent 协作

虽然单体 Agent 架构对于定义明确的问题可能是有效的，但在面对复杂的多领域任务时，其能力往往受到限制。多 Agent 协作模式通过将系统构建为由不同专门化 Agent 组成的协作集合来解决这些限制。这种方法基于任务分解原则，其中高级目标被分解为离散的子问题。然后将每个子问题分配给拥有最适合该任务的特定工具、数据访问或推理能力的 Agent。

例如，一个复杂的研究查询可能被分解并分配给研究 Agent 进行信息检索、数据分析 Agent 进行统计处理，以及综合 Agent 生成最终报告。这种系统的效能不仅仅源于劳动分工，而是关键依赖于 Agent 间通信的机制。这需要标准化的通信协议和共享本体，允许 Agent 交换数据、委托子任务并协调其行动以确保最终输出的连贯性。

这种分布式架构提供了几个优势，包括增强的模块化、可扩展性和稳健性，因为单个 Agent 的故障不一定会导致整个系统故障。协作允许产生协同结果，其中多 Agent 系统的集体性能超过集合内任何单个 Agent 的潜在能力。

多 Agent 协作模式概述

多 Agent 协作模式涉及设计系统，其中多个独立或半独立的 Agent 协同工作以实现共同目标。每个 Agent 通常具有定义的角色、与总体目标一致的特定目标，并且可能访问不同的工具或知识库。此模式的力量在于这些 Agent 之间的交互和协同作用。

协作可以采取各种形式：

  • 顺序交接： 一个 Agent 完成任务并将其输出传递给另一个 Agent 以进行管道中的下一步（类似于规划模式，但明确涉及不同的 Agent）。

  • 并行处理： 多个 Agent 同时处理问题的不同部分，然后它们的结果稍后被组合。

  • 辩论和共识： 多 Agent 协作，其中具有不同观点和信息来源的 Agent 进行讨论以评估选项，最终达成共识或更明智的决策。

  • 层次结构： 管理者 Agent 可能根据其工具访问或插件能力动态地将任务委托给工作 Agent，并综合其结果。每个 Agent 还可以处理相关的工具组，而不是单个 Agent 处理所有工具。

  • 专家团队： 在不同领域具有专业知识的 Agent（例如，研究员、作家、编辑）协作产生复杂输出。

  • 批评者-审查者： Agent 创建初始输出，如计划、草稿或答案。第二组 Agent 然后批判性地评估此输出是否符合政策、安全性、合规性、正确性、质量以及与组织目标的一致性。原始创建者或最终 Agent 根据此反馈修订输出。此模式对于代码生成、研究写作、逻辑检查和确保道德一致性特别有效。这种方法的优势包括增强的稳健性、改进的质量以及减少幻觉或错误的可能性。

多 Agent 系统（见图 1）从根本上包括 Agent 角色和职责的界定、Agent 通过其交换信息的通信渠道的建立，以及指导其协作努力的任务流或交互协议的制定。

图 1：多 Agent 系统示例

像 CrewAI 和 Google ADK 这样的框架旨在通过提供 Agent、任务及其交互程序的规范结构来促进这种范式。这种方法对于需要各种专业知识、包含多个离散阶段或利用并发处理和跨 Agent 信息确认优势的挑战特别有效。

实际应用与用例

多 Agent 协作是一种适用于众多领域的强大模式：

  • 复杂研究和分析： 一组 Agent 可以协作完成研究项目。一个 Agent 可能专门搜索学术数据库，另一个总结发现，第三个识别趋势，第四个将信息综合成报告。这反映了人类研究团队可能如何运作。
  • 软件开发： 想象 Agent 协作构建软件。一个 Agent 可以是需求分析师，另一个是代码生成器，第三个是测试员，第四个是文档编写者。他们可以在彼此之间传递输出以构建和验证组件。
  • 创意内容生成： 创建营销活动可能涉及市场研究 Agent、文案撰写 Agent、图形设计 Agent（使用图像生成工具）和社交媒体调度 Agent，所有这些都在一起工作。
  • 财务分析： 多 Agent 系统可以分析金融市场。Agent 可能专门获取股票数据、分析新闻情绪、执行技术分析和生成投资建议。
  • 客户支持升级： 前线支持 Agent 可以处理初始查询，在需要时将复杂问题升级给专家 Agent（例如，技术专家或计费专家），展示基于问题复杂性的顺序交接。
  • 供应链优化： Agent 可以代表供应链中的不同节点（供应商、制造商、分销商）并协作优化库存水平、物流和调度以响应需求变化或中断。
  • 网络分析与修复：自主操作从 Agent 架构中受益匪浅，特别是在故障定位方面。多个 Agent 可以协作分类和修复问题，建议最佳行动。这些 Agent 还可以与传统机器学习模型和工具集成，利用现有系统，同时提供生成式 AI 的优势。

界定专门 Agent 并细致编排其相互关系的能力使开发人员能够构建展现增强模块化、可扩展性以及处理单个集成 Agent 无法解决的复杂性的系统。

多 Agent 协作：探索相互关系和通信结构

理解 Agent 交互和通信的复杂方式对于设计有效的多 Agent 系统至关重要。如图 2 所示，存在一系列相互关系和通信模型，从最简单的单 Agent 场景到复杂的、定制设计的协作框架。每个模型都呈现独特的优势和挑战，影响多 Agent 系统的整体效率、稳健性和适应性。

1. 单 Agent： 在最基本的层面上，”单 Agent”在没有与其他实体直接交互或通信的情况下自主运行。虽然此模型易于实现和管理，但其能力固有地受到单个 Agent 范围和资源的限制。它适用于可分解为独立子问题的任务，每个子问题都可由单个自给自足的 Agent 解决。

2. 网络： “网络”模型代表了向协作迈出的重要一步，其中多个 Agent 以去中心化方式直接相互交互。通信通常以点对点方式进行，允许共享信息、资源甚至任务。此模型促进弹性，因为一个 Agent 的故障不一定会使整个系统瘫痪。然而，管理通信开销并确保大型非结构化网络中的连贯决策可能具有挑战性。

3. 监督者： 在”监督者”模型中，专用 Agent（”监督者”）监督和协调一组下属 Agent 的活动。监督者充当通信、任务分配和冲突解决的中心枢纽。这种层次结构提供了清晰的权限线，可以简化管理和控制。然而，它引入了单点故障（监督者），如果监督者被大量下属或复杂任务压倒，可能会成为瓶颈。

4. 监督者作为工具： 此模型是”监督者”概念的细微扩展，其中监督者的角色不太关乎直接命令和控制，而更多关乎向其他 Agent 提供资源、指导或分析支持。监督者可能提供工具、数据或计算服务，使其他 Agent 能够更有效地执行其任务，而不必规定其每一个行动。这种方法旨在利用监督者的能力，而不施加严格的自上而下控制。

5. 层次化： “层次化”模型扩展了监督者概念，创建了多层组织结构。这涉及多个监督者级别，高级监督者监督低级监督者，最终在最低层有一组操作 Agent。此结构非常适合可分解为子问题的复杂问题，每个子问题由层次结构的特定层管理。它提供了一种结构化的可扩展性和复杂性管理方法，允许在定义的边界内进行分布式决策。

图 2：Agent 以各种方式进行通信和交互。

6. 自定义： “自定义”模型代表了多 Agent 系统设计的终极灵活性。它允许创建根据给定问题或应用程序的特定要求精确定制的独特相互关系和通信结构。这可能涉及结合前述模型元素的混合方法，或从环境的独特约束和机会中产生的全新设计。自定义模型通常源于需要针对特定性能指标进行优化、处理高度动态的环境或将特定领域知识纳入系统架构。设计和实现自定义模型通常需要对多 Agent 系统原理有深入理解，并仔细考虑通信协议、协调机制和涌现行为。

总之，为多 Agent 系统选择相互关系和通信模型是关键的设计决策。每个模型提供不同的优势和劣势，最佳选择取决于诸如任务复杂性、Agent 数量、期望的自主程度、对稳健性的需求以及可接受的通信开销等因素。多 Agent 系统的未来进展可能会继续探索和完善这些模型，以及开发协作智能的新范式。

实操代码（Crew AI）

此 Python 代码使用 CrewAI 框架定义了一个 AI 驱动的团队来生成关于 AI 趋势的博客文章。它首先设置环境，从 .env 文件加载 API 密钥。应用程序的核心涉及定义两个 Agent：一个研究员用于查找和总结 AI 趋势，一个作家用于基于研究创建博客文章。

相应地定义了两个任务：一个用于研究趋势，另一个用于撰写博客文章，写作任务依赖于研究任务的输出。然后将这些 Agent 和任务组装成一个团队，指定顺序流程，其中任务按顺序执行。团队使用 Agent、任务和语言模型（特别是”gemini-2.0-flash”模型）初始化。主函数使用 kickoff() 方法执行此团队，编排 Agent 之间的协作以产生所需的输出。最后，代码打印团队执行的最终结果，即生成的博客文章。

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
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from langchain_google_genai import ChatGoogleGenerativeAI

def setup_environment():
    """加载环境变量并检查所需的 API 密钥。"""
    load_dotenv()
    if not os.getenv("GOOGLE_API_KEY"):
        raise ValueError("未找到 GOOGLE_API_KEY。请在您的 .env 文件中设置它。")

def main():
    """
    使用最新的 Gemini 模型初始化并运行用于内容创建的 AI 团队。
    """
    setup_environment()
    
    # 定义要使用的语言模型。
    # 更新为 Gemini 2.0 系列中的模型以获得更好的性能和功能。
    # 对于尖端（预览）功能，您可以使用 "gemini-2.5-flash"。
    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash")
    
    # 定义具有特定角色和目标的 Agent
    researcher = Agent(
        role='高级研究分析师',
        goal='查找并总结 AI 的最新趋势。',
        backstory="你是一位经验丰富的研究分析师，擅长识别关键趋势和综合信息。",
        verbose=True,
        allow_delegation=False,
    )
    
    writer = Agent(
        role='技术内容作家',
        goal='基于研究发现撰写清晰且引人入胜的博客文章。',
        backstory="你是一位熟练的作家，可以将复杂的技术主题转化为易于理解的内容。",
        verbose=True,
        allow_delegation=False,
    )
    
    # 为 Agent 定义任务
    research_task = Task(
        description="研究 2024-2025 年人工智能中出现的前 3 个趋势。重点关注实际应用和潜在影响。",
        expected_output="前 3 个 AI 趋势的详细摘要，包括关键点和来源。",
        agent=researcher,
    )
    
    writing_task = Task(
        description="基于研究发现撰写一篇 500 字的博客文章。文章应该引人入胜且易于普通读者理解。",
        expected_output="一篇关于最新 AI 趋势的完整 500 字博客文章。",
        agent=writer,
        context=[research_task],
    )
    
    # 创建团队
    blog_creation_crew = Crew(
        agents=[researcher, writer],
        tasks=[research_task, writing_task],
        process=Process.sequential,
        llm=llm,
        verbose=2 # 为详细的团队执行日志设置详细程度
    )
    
    # 执行团队
    print("## 使用 Gemini 2.0 Flash 运行博客创建团队... ##")
    try:
        result = blog_creation_crew.kickoff()
        print("\n------------------\n")
        print("## 团队最终输出 ##")
        print(result)
    except Exception as e:
        print(f"\n发生意外错误：{e}")

if __name__ == "__main__":
    main()

我们现在将深入研究 Google ADK 框架中的更多示例，特别强调层次化、并行和顺序协调范式，以及将 Agent 实现为操作工具。

实操代码（Google ADK）

以下代码示例演示了通过创建父子关系在 Google ADK 中建立层次化 Agent 结构。代码定义了两种类型的 Agent：LlmAgent 和从 BaseAgent 派生的自定义 TaskExecutor Agent。TaskExecutor 专为特定的非 LLM 任务而设计，在此示例中，它只是生成”任务成功完成”事件。使用指定的模型和指令初始化名为 greeter 的 LlmAgent，以充当友好的欢迎者。自定义 TaskExecutor 实例化为 task_doer。创建名为 coordinator 的父 LlmAgent，也带有模型和指令。coordinator 的指令引导它将欢迎委托给 greeter，将任务执行委托给 task_doer。greeter 和 task_doer 作为子 Agent 添加到 coordinator，建立父子关系。然后代码断言此关系设置正确。最后，它打印一条消息，指示已成功创建 Agent 层次结构。

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
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

## 通过扩展 BaseAgent 正确实现自定义 Agent
class TaskExecutor(BaseAgent):
    """具有自定义非 LLM 行为的专门 Agent。"""
    name: str = "TaskExecutor"
    description: str = "执行预定义的任务。"
    
    async def _run_async_impl(self, context: InvocationContext) -> AsyncGenerator[Event, None]:
        """任务的自定义实现逻辑。"""
        # 这是您的自定义逻辑所在的地方。
        # 对于此示例，我们只会生成一个简单的事件。
        yield Event(author=self.name, content="任务成功完成。")

## 使用适当的初始化定义单个 Agent
## LlmAgent 需要指定模型。
greeter = LlmAgent(
    name="Greeter",
    model="gemini-2.0-flash-exp",
    instruction="你是一个友好的欢迎者。"
)

task_doer = TaskExecutor() # 实例化我们的具体自定义 Agent

## 创建父 Agent 并分配其子 Agent
## 父 Agent 的描述和指令应该引导其委托逻辑。
coordinator = LlmAgent(
    name="Coordinator",
    model="gemini-2.0-flash-exp",
    description="可以欢迎用户并执行任务的协调者。",
    instruction="当被要求欢迎时，委托给 Greeter。当被要求执行任务时，委托给 TaskExecutor。",
    sub_agents=[
        greeter,
        task_doer
    ]
)

## ADK 框架自动建立父子关系。
## 如果在初始化后检查，这些断言将通过。
assert greeter.parent_agent == coordinator
assert task_doer.parent_agent == coordinator

print("Agent 层次结构创建成功。")

此代码摘录说明了在 Google ADK 框架中使用 LoopAgent 建立迭代工作流。代码定义了两个 Agent：ConditionChecker 和 ProcessingStep。ConditionChecker 是一个自定义 Agent，检查会话状态中的”status”值。如果”status”为”completed”，ConditionChecker 升级事件以停止循环。否则，它生成事件以继续循环。ProcessingStep 是使用”gemini-2.0-flash-exp”模型的 LlmAgent。其指令是执行任务，如果是最后一步则将会话”status”设置为”completed”。创建名为 StatusPoller 的 LoopAgent。StatusPoller 配置为 max_iterations=10。StatusPoller 包括 ProcessingStep 和 ConditionChecker 实例作为子 Agent。LoopAgent 将顺序执行子 Agent 最多 10 次迭代，如果 ConditionChecker 发现状态为”completed”则停止。

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
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

## 最佳实践：将自定义 Agent 定义为完整的、自描述的类。
class ConditionChecker(BaseAgent):
    """检查会话状态中"completed"状态的自定义 Agent。"""
    name: str = "ConditionChecker"
    description: str = "检查流程是否完成并向循环发出停止信号。"
    
    async def _run_async_impl(
        self, context: InvocationContext
    ) -> AsyncGenerator[Event, None]:
        """检查状态并生成事件以继续或停止循环。"""
        status = context.session.state.get("status", "pending")
        is_done = (status == "completed")
        
        if is_done:
            # 在满足条件时升级以终止循环。
            yield Event(author=self.name, actions=EventActions(escalate=True))
        else:
            # 生成简单事件以继续循环。
            yield Event(author=self.name, content="条件未满足，继续循环。")

## 更正：LlmAgent 必须有模型和清晰的指令。
process_step = LlmAgent(
    name="ProcessingStep",
    model="gemini-2.0-flash-exp",
    instruction="你是较长流程中的一步。执行你的任务。如果你是最后一步，通过将 'status' 设置为 'completed' 来更新会话状态。"
)

## LoopAgent 编排工作流。
poller = LoopAgent(
    name="StatusPoller",
    max_iterations=10,
    sub_agents=[
        process_step,
        ConditionChecker() # 实例化定义良好的自定义 Agent。
    ]
)

## 此轮询器现在将执行 'process_step'
## 然后执行 'ConditionChecker'
## 重复直到状态为 'completed' 或已经过
## 10 次迭代。

此代码摘录阐明了 Google ADK 中的 SequentialAgent 模式，专为构建线性工作流而设计。此代码使用 google.adk.agents 库定义顺序 Agent 管道。管道由两个 Agent 组成，step1 和 step2。step1 命名为”Step1_Fetch”，其输出将存储在会话状态中的键”data”下。step2 命名为”Step2_Process”，并被指示分析存储在 session.state[“data”] 中的信息并提供摘要。名为”MyPipeline”的 SequentialAgent 编排这些子 Agent 的执行。当使用初始输入运行管道时，step1 将首先执行。来自 step1 的响应将保存到键”data”下的会话状态中。随后，step2 将执行，根据其指令利用 step1 放入状态的信息。此结构允许构建工作流，其中一个 Agent 的输出成为下一个 Agent 的输入。这是创建多步 AI 或数据处理管道的常见模式。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
from google.adk.agents import SequentialAgent, LlmAgent # Changed Agent to LlmAgent

## 此 Agent 的输出将保存到 session.state["data"]
step1 = LlmAgent(name="Step1_Fetch", output_key="data", model="gemini-2.0-flash-exp") # Added model for LlmAgent

## 此 Agent 将使用前一步的数据。
## 我们指示它如何查找和使用此数据。
step2 = LlmAgent( # Changed Agent to LlmAgent
    name="Step2_Process",
    instruction="分析在 state['data'] 中找到的信息并提供摘要。",
    model="gemini-2.0-flash-exp" # Added model for LlmAgent
)

pipeline = SequentialAgent(
    name="MyPipeline",
    sub_agents=[step1, step2]
)

## 当使用初始输入运行管道时，Step1 将执行，
## 其响应将存储在 session.state["data"] 中，然后
## Step2 将执行，按指示使用来自状态的信息。

以下代码示例说明了 Google ADK 中的 ParallelAgent 模式，它促进多个 Agent 任务的并发执行。data_gatherer 设计为并发运行两个子 Agent：weather_fetcher 和 news_fetcher。weather_fetcher Agent 被指示获取给定位置的天气并将结果存储在 session.state[“weather_data”] 中。同样，news_fetcher Agent 被指示检索给定主题的头条新闻故事并将其存储在 session.state[“news_data”] 中。每个子 Agent 都配置为使用”gemini-2.0-flash-exp”模型。ParallelAgent 编排这些子 Agent 的执行，允许它们并行工作。来自 weather_fetcher 和 news_fetcher 的结果将被收集并存储在会话状态中。最后，示例展示了如何在 Agent 执行完成后从 final_state 访问收集的天气和新闻数据。

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
from google.adk.agents import LlmAgent, ParallelAgent # Changed Agent to LlmAgent

## 最好将获取逻辑定义为 Agent 的工具
## 为了简化此示例，我们将逻辑嵌入 Agent 的指令中。
## 在实际场景中，您将使用工具。

## 定义将并行运行的各个 Agent
weather_fetcher = LlmAgent( # Changed Agent to LlmAgent
    name="weather_fetcher",
    model="gemini-2.0-flash-exp",
    instruction="获取给定位置的天气并仅返回天气报告。",
    output_key="weather_data"  # 结果将存储在 session.state["weather_data"] 中
)

news_fetcher = LlmAgent( # Changed Agent to LlmAgent
    name="news_fetcher",
    model="gemini-2.0-flash-exp",
    instruction="获取给定主题的头条新闻故事并仅返回该故事。",
    output_key="news_data"      # 结果将存储在 session.state["news_data"] 中
)

## 创建 ParallelAgent 以编排子 Agent
data_gatherer = ParallelAgent(
    name="data_gatherer",
    sub_agents=[
        weather_fetcher,
        news_fetcher
    ]
)

提供的代码段示例了 Google ADK 中的”Agent 作为工具”范式，使 Agent 能够以类似于函数调用的方式利用另一个 Agent 的能力。具体来说，代码使用 Google 的 LlmAgent 和 AgentTool 类定义了一个图像生成系统。它由两个 Agent 组成：父 artist_agent 和子 Agent image_generator_agent。generate_image 函数是一个简单的工具，模拟图像创建，返回模拟图像数据。image_generator_agent 负责根据其接收的文本提示词使用此工具。artist_agent 的角色是首先发明一个创意图像提示词。然后它通过 AgentTool 包装器调用 image_generator_agent。AgentTool 充当桥梁，允许一个 Agent 将另一个 Agent 用作工具。当 artist_agent 调用 image_tool 时，AgentTool 使用艺术家发明的提示词调用 image_generator_agent。image_generator_agent 然后使用该提示词使用 generate_image 函数。最后，生成的图像（或模拟数据）通过 Agent 返回。此架构演示了一个分层 Agent 系统，其中更高级别的 Agent 编排较低级别的专门 Agent 以执行任务。

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
from google.adk.agents import LlmAgent
from google.adk.tools import agent_tool
from google.genai import types

## 1. 核心能力的简单函数工具。
## 这遵循将操作与推理分离的最佳实践。
def generate_image(prompt: str) -> dict:
    """
    基于文本提示词生成图像。
    参数：
        prompt：要生成的图像的详细描述。
    返回：
        包含状态和生成的图像字节的字典。
    """
    print(f"工具：为提示词生成图像：'{prompt}'")
    # 在真实实现中，这将调用图像生成 API。
    # 对于此示例，我们返回模拟图像数据。
    mock_image_bytes = b"mock_image_data_for_a_cat_wearing_a_hat"
    return {
        "status": "success",
        # 工具返回原始字节，Agent 将处理 Part 创建。
        "image_bytes": mock_image_bytes,
        "mime_type": "image/png"
    }

## 2. 将 ImageGeneratorAgent 重构为 LlmAgent。
## 它现在正确使用传递给它的输入。
image_generator_agent = LlmAgent(
    name="ImageGen",
    model="gemini-2.0-flash",
    description="基于详细的文本提示词生成图像。",
    instruction=(
        "你是图像生成专家。你的任务是接受用户的请求 "
        "并使用 `generate_image` 工具创建图像。"
        "用户的整个请求应用作工具的 'prompt' 参数。"
        "工具返回图像字节后，你必须输出图像。"
    ),
    tools=[generate_image]
)

## 3. 将更正的 Agent 包装在 AgentTool 中。
## 这里的描述是父 Agent 看到的。
image_tool = agent_tool.AgentTool(
    agent=image_generator_agent,
    description="使用此工具生成图像。输入应该是所需图像的描述性提示词。"
)

## 4. 父 Agent 保持不变。其逻辑是正确的。
artist_agent = LlmAgent(
    name="Artist",
    model="gemini-2.0-flash",
    instruction=(
        "你是一位富有创造力的艺术家。首先，为图像发明一个创意且描述性的提示词。"
        "然后，使用 `ImageGen` 工具使用你的提示词生成图像。"
    ),
    tools=[image_tool]
)

概览

是什么： 复杂问题通常超出单个单体基于 LLM 的 Agent 的能力。单个 Agent 可能缺乏解决多方面任务所有部分所需的多样化专业技能或对特定工具的访问。此限制造成瓶颈，降低系统的整体效率和可扩展性。因此，处理复杂的多领域目标变得低效，并可能导致不完整或次优的结果。

为什么： 多 Agent 协作模式通过创建多个协作 Agent 的系统提供了标准化解决方案。复杂问题被分解为更小的更易于管理的子问题。然后将每个子问题分配给具有解决它所需的精确工具和能力的专门 Agent。这些 Agent 通过定义的通信协议和交互模型（如顺序交接、并行工作流或层次化委托）协同工作。这种 Agent 化的分布式方法创造了协同效应，使团队能够实现任何单个 Agent 都无法实现的结果。

经验法则： 当任务对于单个 Agent 来说太复杂并且可以分解为需要专业技能或工具的不同子任务时，使用此模式。它非常适合受益于多样化专业知识、并行处理或具有多个阶段的结构化工作流的问题，例如复杂的研究和分析、软件开发或创意内容生成。

视觉摘要

图 3：多 Agent 设计模式

关键要点

  • 多 Agent 协作涉及多个 Agent 协同工作以实现共同目标。
  • 此模式利用专业角色、分布式任务和 Agent 间通信。
  • 协作可以采取顺序交接、并行处理、辩论或层次结构等形式。
  • 此模式非常适合需要多样化专业知识或多个不同阶段的复杂问题。

结论

本章探讨了多 Agent 协作模式，展示了在系统内编排多个专门 Agent 的好处。我们研究了各种协作模型，强调该模式在跨不同领域解决复杂多方面问题中的关键作用。理解 Agent 协作自然会引出对其与外部环境交互的探究。

参考文献

  1. Multi-Agent Collaboration Mechanisms: A Survey of LLMs, https://arxiv.org/abs/2501.06322
  2. Multi-Agent System — The Power of Collaboration, https://aravindakumar.medium.com/introducing-multi-agent-frameworks-the-power-of-collaboration-e9db31bba1b6

展开全文 >>

Agentic Design Patterns - Chapter 6_ Planning

第 6 章：规划

智能行为通常不仅仅涉及对即时输入做出反应。它需要远见、将复杂任务分解为更小的可管理步骤，以及制定实现期望结果的策略。这就是规划模式发挥作用的地方。规划的核心是 Agent 或 Agent 系统制定一系列行动以从初始状态向目标状态移动的能力。

规划模式概述

在 AI 的背景下，将规划 Agent 视为您委托复杂目标的专家是有帮助的。当您要求它”组织团队外出活动”时，您定义的是”什么”——目标及其约束——而不是”如何”。Agent 的核心任务是自主规划实现该目标的路线。它必须首先理解初始状态（例如，预算、参与者人数、期望日期）和目标状态（成功预订的外出活动），然后发现连接它们的最佳行动序列。计划并不是预先知道的；它是响应请求而创建的。

这个过程的标志是适应性。初始计划仅仅是一个起点，而不是僵化的脚本。Agent 的真正力量在于其整合新信息并引导项目绕过障碍的能力。例如，如果首选场地变得不可用或选择的餐饮服务商已完全预订，一个有能力的 Agent 不会简单地失败。它会适应：记录新的约束，重新评估选项，并制定新计划，也许通过建议替代场地或日期。

然而，认识到灵活性和可预测性之间的权衡至关重要。动态规划是一个特定的工具，而不是通用解决方案。当问题的解决方案已经被充分理解且可重复时，将 Agent 限制为预定的固定工作流更有效。这种方法限制 Agent 的自主性以减少不确定性和不可预测行为的风险，保证可靠和一致的结果。因此，使用规划 Agent 与简单任务执行 Agent 的决定取决于一个问题：是否需要发现”如何”，还是已经知道？

实际应用与用例

规划模式是自主系统中的核心计算过程，使 Agent 能够综合一系列行动以实现指定目标，特别是在动态或复杂环境中。这个过程将高级目标转换为由离散可执行步骤组成的结构化计划。

在过程任务自动化等领域，规划用于编排复杂的工作流。例如，像新员工入职这样的业务流程可以分解为定向的子任务序列，例如创建系统帐户、分配培训模块和与不同部门协调。Agent 生成一个计划以逻辑顺序执行这些步骤，调用必要的工具或与各种系统交互以管理依赖关系。

在机器人和自主导航中，规划对于状态空间遍历是基础性的。一个系统，无论是物理机器人还是虚拟实体，都必须生成路径或行动序列以从初始状态转换到目标状态。这涉及优化时间或能源消耗等指标，同时遵守环境约束，如避开障碍物或遵守交通规则。

此模式对于结构化信息综合也至关重要。当被要求生成像研究报告这样的复杂输出时，Agent 可以制定一个包括信息收集、数据总结、内容结构化和迭代完善的不同阶段的计划。同样，在涉及多步问题解决的客户支持场景中，Agent 可以创建并遵循诊断、解决方案实施和升级的系统计划。

从本质上讲，规划模式允许 Agent 从简单的反应性行动转向目标导向的行为。它提供了解决需要一系列相互依赖操作的问题所必需的逻辑框架。

实操代码（Crew AI）

以下部分将演示使用 Crew AI 框架实现规划模式。此模式涉及一个 Agent，它首先制定多步计划以解决复杂查询，然后顺序执行该计划。

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
import os
from dotenv import load_dotenv
from crewai import Agent, Task, Crew, Process
from langchain_openai import ChatOpenAI

## 从 .env 文件加载环境变量以确保安全
load_dotenv()

## 1. 为清晰起见，明确定义语言模型
llm = ChatOpenAI(model="gpt-4-turbo")

## 2. 定义一个清晰且专注的 Agent
planner_writer_agent = Agent(
    role='文章规划者和撰写者',
    goal='规划然后撰写关于指定主题的简洁、引人入胜的摘要。',
    backstory=(
        '你是一位专业的技术作家和内容策略师。'
        '你的优势在于在写作之前创建清晰、可操作的计划，'
        '确保最终摘要既信息丰富又易于理解。'
    ),
    verbose=True,
    allow_delegation=False,
    llm=llm # 将特定 LLM 分配给 Agent
)

## 3. 定义具有更结构化和具体的预期输出的任务
topic = "强化学习在 AI 中的重要性"
high_level_task = Task(
    description=(
        f"1. 为主题'{topic}'的摘要创建要点计划。\n"
        f"2. 根据您的计划撰写摘要，保持在 200 字左右。"
    ),
    expected_output=(
        "包含两个不同部分的最终报告：\n\n"
        "### 计划\n"
        "- 概述摘要要点的项目符号列表。\n\n"
        "### 摘要\n"
        "- 主题的简洁且结构良好的摘要。"
    ),
    agent=planner_writer_agent,
)

## 使用清晰的流程创建团队
crew = Crew(
    agents=[planner_writer_agent],
    tasks=[high_level_task],
    process=Process.sequential,
)

## 执行任务
print("## 运行规划和写作任务 ##")
result = crew.kickoff()
print("\n\n---\n## 任务结果 ##\n---")
print(result)

此代码使用 CrewAI 库创建一个 AI Agent，该 Agent 规划并撰写关于给定主题的摘要。它首先导入必要的库，包括 CrewAI 和 langchain_openai，并从 .env 文件加载环境变量。明确定义了一个 ChatOpenAI 语言模型供 Agent 使用。创建了一个名为 planner_writer_agent 的 Agent，具有特定的角色和目标：规划然后撰写简洁的摘要。Agent 的背景故事强调其在规划和技术写作方面的专业知识。定义了一个任务，明确描述首先创建计划，然后撰写关于”强化学习在 AI 中的重要性”主题的摘要，并为预期输出指定了特定格式。组建了一个包含 Agent 和任务的团队，设置为顺序处理它们。最后，调用 crew.kickoff() 方法执行定义的任务并打印结果。

Google DeepResearch

Google Gemini DeepResearch（见图 1）是一个基于 Agent 的系统，设计用于自主信息检索和综合。它通过多步 Agent 管道运行，动态和迭代地查询 Google 搜索以系统地探索复杂主题。该系统设计用于处理大量基于网络的资源，评估收集的数据的相关性和知识差距，并执行后续搜索以解决它们。最终输出将经过审核的信息整合成一个结构化的多页摘要，并附有原始来源的引用。

进一步扩展，该系统的运行不是单个查询-响应事件，而是一个受管理的长时间运行过程。它首先将用户的提示词分解为多点研究计划（见图 1），然后呈现给用户审查和修改。这允许在执行之前协作塑造研究轨迹。一旦计划获得批准，Agent 管道就会启动其迭代搜索和分析循环。这不仅仅是执行一系列预定义的搜索；Agent 根据收集的信息动态制定和完善其查询，主动识别知识差距、确认数据点并解决差异。


图 1：Google Deep Research Agent 生成使用 Google 搜索作为工具的执行计划。

关键的架构组件是系统异步管理此过程的能力。这种设计确保了调查（可能涉及分析数百个来源）对单点故障具有弹性，并允许用户脱离并在完成时收到通知。系统还可以集成用户提供的文档，将来自私有来源的信息与其基于网络的研究相结合。最终输出不仅仅是发现的串联列表，而是一个结构化的多页报告。在综合阶段，模型对收集的信息进行批判性评估，识别主要主题并将内容组织成具有逻辑部分的连贯叙述。该报告设计为交互式的，通常包括音频概述、图表和指向原始引用来源的链接等功能，允许用户验证和进一步探索。除了综合结果外，模型还明确返回它搜索和咨询的完整来源列表（见图 2）。这些作为引用呈现，提供完全的透明度和对主要信息的直接访问。整个过程将简单的查询转换为全面的、综合的知识体。


图 2：Deep Research 计划执行的示例，导致使用 Google 搜索作为工具搜索各种网络来源。

通过减轻手动数据获取和综合所需的大量时间和资源投资，Gemini DeepResearch 为信息发现提供了更结构化和详尽的方法。该系统的价值在跨各个领域的复杂、多方面研究任务中尤为明显。

例如，在竞争分析中，可以指示 Agent 系统地收集和整理关于市场趋势、竞争对手产品规格、来自各种在线来源的公众情绪和营销策略的数据。这个自动化过程取代了手动跟踪多个竞争对手的繁重任务，使分析师能够专注于更高层次的战略解释而不是数据收集（见图 3）。


图 3：Google Deep Research Agent 生成的最终输出，代表我们分析使用 Google 搜索作为工具获得的来源。

同样，在学术探索中，该系统作为进行广泛文献综述的强大工具。它可以识别和总结基础论文，追踪概念在众多出版物中的发展，并绘制特定领域内新兴研究前沿的地图，从而加速学术探究的初始和最耗时的阶段。

这种方法的效率源于迭代搜索和过滤周期的自动化，这是手动研究的核心瓶颈。通过系统处理比人类研究人员在可比时间框架内通常可行的更大量和更多样化的信息来源来实现全面性。这种更广泛的分析范围有助于减少选择偏差的潜力，并增加发现不太明显但可能至关重要的信息的可能性，从而导致对主题更稳健和充分支持的理解。

OpenAI Deep Research API

OpenAI Deep Research API 是一个专门设计用于自动化复杂研究任务的工具。它利用一个先进的 Agent 模型，可以独立推理、规划和从现实世界来源综合信息。与简单的问答模型不同，它接受高级查询并自主将其分解为子问题，使用其内置工具执行网络搜索，并提供结构化的、富含引用的最终报告。该 API 提供对整个过程的直接程序化访问，在撰写本文时使用 o3-deep-research-2025-06-26 等模型进行高质量综合，以及更快的 o4-mini-deep-research-2025-06-26 用于延迟敏感的应用程序。

Deep Research API 很有用，因为它自动化了原本需要数小时的手动研究，提供适合为业务战略、投资决策或政策建议提供信息的专业级、数据驱动的报告。其主要好处包括：

  • 结构化、引用的输出： 它产生组织良好的报告，带有链接到来源元数据的内联引用，确保声明可验证且有数据支持。
  • 透明度： 与 ChatGPT 中的抽象过程不同，API 公开所有中间步骤，包括 Agent 的推理、它执行的特定网络搜索查询以及它运行的任何代码。这允许详细的调试、分析以及更深入地理解最终答案是如何构建的。
  • 可扩展性： 它支持模型上下文协议（MCP），使开发人员能够将 Agent 连接到私有知识库和内部数据源，将公共网络研究与专有信息混合。

要使用 API，您向 client.responses.create 端点发送请求，指定模型、输入提示词和 Agent 可以使用的工具。输入通常包括定义 Agent 角色和期望输出格式的 system_message，以及 user_query。您还必须包括 web_search_preview 工具，并可以选择添加其他工具，如 code_interpreter 或自定义 MCP 工具（见第 10 章）用于内部数据。

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
from openai import OpenAI

## 使用您的 API 密钥初始化客户端
client = OpenAI(api_key="YOUR_OPENAI_API_KEY")

## 定义 Agent 的角色和用户的研究问题
system_message = """你是一名准备结构化、数据驱动报告的专业研究员。专注于数据丰富的见解，使用可靠的来源，并包括内联引用。"""

user_query = "研究司美格鲁肽对全球医疗保健系统的经济影响。"

## 创建 Deep Research API 调用
response = client.responses.create(
    model="o3-deep-research-2025-06-26",
    input=[
        {
            "role": "developer",
            "content": [{"type": "input_text", "text": system_message}]
        },
        {
            "role": "user",
            "content": [{"type": "input_text", "text": user_query}]
        }
    ],
    reasoning={"summary": "auto"},
    tools=[{"type": "web_search_preview"}]
)

## 从响应中访问并打印最终报告
final_report = response.output[-1].content[0].text
print(final_report)

## --- 访问内联引用和元数据 ---
print("--- 引用 ---")
annotations = response.output[-1].content[0].annotations

if not annotations:
    print("报告中未找到注释。")
else:
    for i, citation in enumerate(annotations):
        # 引用所指的文本范围
        cited_text = final_report[citation.start_index:citation.end_index]
        print(f"引用 {i+1}:")
        print(f"  引用文本：{cited_text}")
        print(f"  标题：{citation.title}")
        print(f"  URL：{citation.url}")
        print(f"  位置：字符 {citation.start_index}–{citation.end_index}")

print("\n" + "="*50 + "\n")

## --- 检查中间步骤 ---
print("--- 中间步骤 ---")

## 1. 推理步骤：模型生成的内部计划和摘要。
try:
    reasoning_step = next(item for item in response.output if item.type == "reasoning")
    print("\n[找到推理步骤]")
    for summary_part in reasoning_step.summary:
        print(f"  - {summary_part.text}")
except StopIteration:
    print("\n未找到推理步骤。")

## 2. 网络搜索调用：Agent 执行的确切搜索查询。
try:
    search_step = next(item for item in response.output if item.type == "web_search_call")
    print("\n[找到网络搜索调用]")
    print(f"  执行的查询：'{search_step.action['query']}'")
    print(f"  状态：{search_step.status}")
except StopIteration:
    print("\n未找到网络搜索步骤。")

## 3. 代码执行：Agent 使用代码解释器运行的任何代码。
try:
    code_step = next(item for item in response.output if item.type == "code_interpreter_call")
    print("\n[找到代码执行步骤]")
    print("  代码输入：")
    print(f"  ```python\n{code_step.input}\n  ```")
    print("  代码输出：")
    print(f"  {code_step.output}")
except StopIteration:
    print("\n未找到代码执行步骤。")

此代码片段利用 OpenAI API 执行”深度研究”任务。它首先使用您的 API 密钥初始化 OpenAI 客户端，这对于身份验证至关重要。然后，它将 AI Agent 的角色定义为专业研究员，并设置用户关于司美格鲁肽经济影响的研究问题。代码构造对 o3-deep-research-2025-06-26 模型的 API 调用，提供定义的系统消息和用户查询作为输入。它还请求推理的自动摘要并启用网络搜索功能。进行 API 调用后，它提取并打印最终生成的报告。

随后，它尝试访问并显示报告注释中的内联引用和元数据，包括引用文本、标题、URL 和报告中的位置。最后，它检查并打印模型采取的中间步骤的详细信息，例如推理步骤、网络搜索调用（包括执行的查询）以及如果使用了代码解释器的任何代码执行步骤。

概览

是什么： 复杂问题通常无法通过单个操作解决，需要远见才能实现期望的结果。没有结构化的方法，Agent 系统难以处理涉及多个步骤和依赖关系的多方面请求。这使得将高级目标分解为一系列可管理的更小可执行任务变得困难。因此，系统无法有效地制定策略，在面对复杂目标时导致不完整或不正确的结果。

为什么： 规划模式通过让 Agent 系统首先创建一个连贯的计划来解决目标提供了标准化解决方案。它涉及将高级目标分解为一系列更小的可操作步骤或子目标。这允许系统管理复杂的工作流、编排各种工具并以逻辑顺序处理依赖关系。LLM 特别适合这一点，因为它们可以基于其庞大的训练数据生成合理且有效的计划。这种结构化方法将简单的反应性 Agent 转变为战略执行者，可以主动朝着复杂目标工作，甚至在必要时调整其计划。

经验法则： 当用户的请求太复杂而无法通过单个操作或工具处理时使用此模式。它非常适合自动化多步流程，例如生成详细的研究报告、新员工入职或执行竞争分析。每当任务需要一系列相互依赖的操作以达到最终的综合结果时，应用规划模式。

视觉摘要

图 4；规划设计模式

关键要点

  • 规划使 Agent 能够将复杂目标分解为可操作的顺序步骤。
  • 它对于处理多步任务、工作流自动化和导航复杂环境至关重要。
  • LLM 可以通过基于任务描述生成逐步方法来执行规划。
  • 明确提示或设计任务以要求规划步骤会在 Agent 框架中鼓励这种行为。
  • Google Deep Research 是一个代表我们分析使用 Google 搜索作为工具获得的来源的 Agent。它反思、规划并执行。

结论

总之，规划模式是将 Agent 系统从简单的反应性响应者提升为战略性、目标导向的执行者的基础组件。现代大型语言模型为此提供了核心能力，自主地将高级目标分解为连贯的可操作步骤。此模式从简单的顺序任务执行（如 CrewAI Agent 创建并遵循写作计划所演示的）扩展到更复杂和动态的系统。Google DeepResearch Agent 体现了这种高级应用，创建基于持续信息收集而适应和演化的迭代研究计划。最终，规划为复杂问题的人类意图和自动化执行之间提供了必要的桥梁。通过构建问题解决方法，此模式使 Agent 能够管理复杂的工作流并提供全面的综合结果。

参考文献

  1. Google DeepResearch (Gemini Feature): gemini.google.com
  2. OpenAI ,Introducing deep research https://openai.com/index/introducing-deep-research/
  3. Perplexity, Introducing Perplexity Deep Research, https://www.perplexity.ai/hub/blog/introducing-perplexity-deep-research

展开全文 >>

Agentic Design Patterns - Chapter 5_ Tool Use

第 5 章：工具使用（函数调用）

工具使用模式概述

到目前为止，我们讨论的 Agent 模式主要涉及编排语言模型之间的交互和管理 Agent 内部工作流中的信息流（链式、路由、并行化、反思）。然而，要使 Agent 真正有用并与现实世界或外部系统交互，它们需要使用工具的能力。

工具使用模式通常通过一种称为函数调用的机制实现，使 Agent 能够与外部 API、数据库、服务交互，甚至执行代码。它允许位于 Agent 核心的大型语言模型（LLM）根据用户请求或任务当前状态，决定何时以及如何使用特定的外部函数。

该过程通常包括：

  1. 工具定义： 外部函数或能力被定义并描述给 LLM。此描述包括函数的目的、名称以及它接受的参数及其类型和描述。
  2. LLM 决策： LLM 接收用户的请求和可用的工具定义。基于其对请求和工具的理解，LLM 决定是否需要调用一个或多个工具来满足请求。
  3. 函数调用生成： 如果 LLM 决定使用工具，它会生成一个结构化输出（通常是 JSON 对象），指定要调用的工具名称和要传递给它的参数（参数），这些参数从用户的请求中提取。
  4. 工具执行： Agent 框架或编排层拦截此结构化输出。它识别请求的工具并使用提供的参数执行实际的外部函数。
  5. 观察/结果： 工具执行的输出或结果返回给 Agent。
  6. LLM 处理（可选但常见）： LLM 接收工具的输出作为上下文，并使用它向用户制定最终响应或决定工作流中的下一步（可能涉及调用另一个工具、反思或提供最终答案）。

此模式是基础性的，因为它打破了 LLM 训练数据的限制，允许它访问最新信息、执行它内部无法完成的计算、与用户特定数据交互或触发现实世界的行动。函数调用是连接 LLM 推理能力与可用的大量外部功能之间差距的技术机制。

虽然”函数调用”准确描述了调用特定预定义代码函数的过程，但考虑更广泛的”工具调用”概念会更有益。这个更广泛的术语承认 Agent 的能力可以远远超出简单的函数执行。”工具”可以是传统函数，也可以是复杂的 API 端点、数据库查询请求，甚至是针对其他专门 Agent 的指令。这种视角使我们能够设想更复杂的系统，例如，主 Agent 可能将复杂的数据分析任务委托给专用的”分析师 Agent”，或通过 API 查询外部知识库。以”工具调用”的思维方式，能更好地捕捉 Agent 作为跨数字资源和其他智能实体生态系统的编排者的全部潜力。

像 LangChain、LangGraph 和 Google Agent Developer Kit (ADK) 这样的框架为定义工具并将它们集成到 Agent 工作流中提供了强大的支持，通常利用现代 LLM（如 Gemini 或 OpenAI 系列中的那些）的原生函数调用能力。在这些框架的”画布”上，您定义工具，然后配置 Agent（通常是 LLM Agent）以意识到并能够使用这些工具。

工具使用是构建强大、交互式和外部感知 Agent 的基石模式。

实际应用与用例

工具使用模式几乎适用于 Agent 需要超越生成文本来执行操作或检索特定动态信息的任何场景：

  1. 从外部源检索信息：
    访问 LLM 训练数据中不存在的实时数据或信息。
  • 用例： 天气 Agent。
    • 工具： 接受位置并返回当前天气状况的天气 API。
    • Agent 流程： 用户问”伦敦的天气如何？”，LLM 识别需要天气工具，用”伦敦”调用工具，工具返回数据，LLM 将数据格式化为用户友好的响应。
  1. 与数据库和 API 交互：
    对结构化数据执行查询、更新或其他操作。
  • 用例： 电子商务 Agent。
    • 工具： API 调用以检查产品库存、获取订单状态或处理付款。
    • Agent 流程： 用户问”产品 X 有库存吗？”，LLM 调用库存 API，工具返回库存数量，LLM 告诉用户库存状态。
  1. 执行计算和数据分析：
    使用外部计算器、数据分析库或统计工具。
  • 用例： 金融 Agent。
    • 工具： 计算器函数、股票市场数据 API、电子表格工具。
    • Agent 流程： 用户问”AAPL 的当前价格是多少，如果我以 150 美元购买 100 股，计算潜在利润？”，LLM 调用股票 API，获取当前价格，然后调用计算器工具，获取结果，格式化响应。
  1. 发送通信：
    发送电子邮件、消息或对外部通信服务进行 API 调用。
  • 用例： 个人助理 Agent。
    • 工具： 电子邮件发送 API。
    • Agent 流程： 用户说”给 John 发一封关于明天会议的电子邮件。”，LLM 使用从请求中提取的收件人、主题和正文调用电子邮件工具。
  1. 执行代码：
    在安全环境中运行代码片段以执行特定任务。
  • 用例： 编码助手 Agent。
    • 工具： 代码解释器。
    • Agent 流程： 用户提供 Python 代码片段并问”这段代码做什么？”，LLM 使用解释器工具运行代码并分析其输出。
  1. 控制其他系统或设备：
    与智能家居设备、物联网平台或其他连接系统交互。
  • 用例： 智能家居 Agent。
    • 工具： 控制智能灯的 API。
    • Agent 流程： 用户说”关闭客厅的灯。”LLM 使用命令和目标设备调用智能家居工具。

工具使用将语言模型从文本生成器转变为能够在数字或物理世界中感知、推理和行动的 Agent（见图 1）

图 1：Agent 使用工具的一些示例

实操代码示例（LangChain）

在 LangChain 框架中实现工具使用是一个两阶段过程。首先，定义一个或多个工具，通常通过封装现有的 Python 函数或其他可运行组件。随后，将这些工具绑定到语言模型，从而赋予模型在确定需要外部函数调用以满足用户查询时生成结构化工具使用请求的能力。

以下实现将通过定义一个简单的函数来模拟信息检索工具，以此演示此原理。随后，将构建并配置一个 Agent 以利用此工具响应用户输入。执行此示例需要安装核心 LangChain 库和特定于模型的提供程序包。此外，使用所选语言模型服务的适当身份验证（通常通过在本地环境中配置的 API 密钥）是必要的先决条件。

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
import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
import logging

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool as langchain_tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

## 取消注释
## 安全地提示用户并将 API 密钥设置为环境变量
os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter your Google API key: ")
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

try:
    # 需要具有函数/工具调用能力的模型。
    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
    print(f"✅ 语言模型已初始化：{llm.model}")
except Exception as e:
    print(f"🛑 初始化语言模型时出错：{e}")
    llm = None

## --- 定义工具 ---
@langchain_tool
def search_information(query: str) -> str:
    """
    提供有关给定主题的事实信息。使用此工具查找诸如"法国首都"或"伦敦的天气？"等短语的答案。
    """
    print(f"\n--- 🛠️ 工具调用：search_information，查询：'{query}' ---")
    
    # 使用预定义结果字典模拟搜索工具。
    simulated_results = {
        "weather in london": "伦敦目前多云，温度为 15°C。",
        "capital of france": "法国的首都是巴黎。",
        "population of earth": "地球的估计人口约为 80 亿人。",
        "tallest mountain": "珠穆朗玛峰是海拔最高的山峰。",
        "default": f"'{query}' 的模拟搜索结果：未找到特定信息，但该主题似乎很有趣。"
    }
    
    result = simulated_results.get(query.lower(), simulated_results["default"])
    print(f"--- 工具结果：{result} ---")
    return result

tools = [search_information]

## --- 创建工具调用 Agent ---
if llm:
    # 此提示词模板需要一个 `agent_scratchpad` 占位符用于 Agent 的内部步骤。
    agent_prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个有用的助手。"),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])
    
    # 创建 Agent，将 LLM、工具和提示词绑定在一起。
    agent = create_tool_calling_agent(llm, tools, agent_prompt)
    
    # AgentExecutor 是调用 Agent 并执行所选工具的运行时。
    # 这里不需要 'tools' 参数，因为它们已经绑定到 Agent。
    agent_executor = AgentExecutor(agent=agent, verbose=True, tools=tools)

async def run_agent_with_tool(query: str):
    """使用查询调用 Agent 执行器并打印最终响应。"""
    print(f"\n--- 🏃 使用查询运行 Agent：'{query}' ---")
    try:
        response = await agent_executor.ainvoke({"input": query})
        print("\n--- ✅ 最终 Agent 响应 ---")
        print(response["output"])
    except Exception as e:
        print(f"\n🛑 Agent 执行期间发生错误：{e}")

async def main():
    """并发运行所有 Agent 查询。"""
    tasks = [
        run_agent_with_tool("法国的首都是什么？"),
        run_agent_with_tool("伦敦的天气怎么样？"),
        run_agent_with_tool("告诉我一些关于狗的事情。") # 应该触发默认工具响应
    ]
    await asyncio.gather(*tasks)

nest_asyncio.apply()
asyncio.run(main())

代码使用 LangChain 库和 Google Gemini 模型设置了一个工具调用 Agent。它定义了一个 search_information 工具，模拟为特定查询提供事实答案。该工具对”weather in london”、”capital of france”和”population of earth”有预定义的响应，以及其他查询的默认响应。初始化了一个 ChatGoogleGenerativeAI 模型，确保其具有工具调用能力。创建了一个 ChatPromptTemplate 来指导 Agent 的交互。使用 create_tool_calling_agent 函数将语言模型、工具和提示词组合成一个 Agent。然后设置一个 AgentExecutor 来管理 Agent 的执行和工具调用。定义了 run_agent_with_tool 异步函数以使用给定查询调用 Agent 并打印结果。main 异步函数准备多个要并发运行的查询。这些查询旨在测试 search_information 工具的特定和默认响应。最后，asyncio.run(main()) 调用执行所有 Agent 任务。代码包括在继续 Agent 设置和执行之前成功 LLM 初始化的检查。

实操代码示例（CrewAI）

此代码提供了如何在 CrewAI 框架中实现函数调用（工具）的实际示例。它设置了一个简单的场景，其中 Agent 配备了查找信息的工具。该示例专门演示了使用此 Agent 和工具获取模拟股票价格。

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
## pip install crewai langchain-openai
import os
from crewai import Agent, Task, Crew
from crewai.tools import tool
import logging

## --- 最佳实践：配置日志 ---
## 基本日志设置有助于调试和跟踪团队的执行。
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

## --- 设置您的 API 密钥 ---
## 对于生产环境，建议使用更安全的密钥管理方法
## 例如在运行时加载的环境变量或秘密管理器。
#
## 为您选择的 LLM 提供商设置环境变量（例如，OPENAI_API_KEY）
## os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
## os.environ["OPENAI_MODEL_NAME"] = "gpt-4o"

## --- 1. 重构的工具：返回干净的数据 ---
## 该工具现在返回原始数据（浮点数）或引发标准 Python 错误。
## 这使其更可重用，并强制 Agent 正确处理结果。
@tool("Stock Price Lookup Tool")
def get_stock_price(ticker: str) -> float:
    """
    获取给定股票代码符号的最新模拟股票价格。
    以浮点数形式返回价格。如果未找到代码，则引发 ValueError。
    """
    logging.info(f"工具调用：get_stock_price，代码为 '{ticker}'")
    
    simulated_prices = {
        "AAPL": 178.15,
        "GOOGL": 1750.30,
        "MSFT": 425.50,
    }
    
    price = simulated_prices.get(ticker.upper())
    if price is not None:
        return price
    else:
        # 引发特定错误比返回字符串更好。
        # Agent 配备了处理异常的能力，可以决定下一步行动。
        raise ValueError(f"未找到代码 '{ticker.upper()}' 的模拟价格。")

## --- 2. 定义 Agent ---
## Agent 定义保持不变，但它现在将利用改进的工具。
financial_analyst_agent = Agent(
    role='高级财务分析师',
    goal='使用提供的工具分析股票数据并报告关键价格。',
    backstory="你是一位经验丰富的财务分析师，擅长使用数据源查找股票信息。你提供清晰、直接的答案。",
    verbose=True,
    tools=[get_stock_price],
    # 允许委托可能很有用，但对于这个简单任务不是必需的。
    allow_delegation=False,
)

## --- 3. 精炼任务：更清晰的说明和错误处理 ---
## 任务描述更具体，并指导 Agent 如何应对
## 成功的数据检索和潜在错误。
analyze_aapl_task = Task(
    description=(
        "Apple（代码：AAPL）的当前模拟股票价格是多少？"
        "使用 'Stock Price Lookup Tool' 查找它。"
        "如果未找到代码，你必须报告无法检索价格。"
    ),
    expected_output=(
        "一个清晰的句子，说明 AAPL 的模拟股票价格。"
        "例如：'AAPL 的模拟股票价格是 $178.15。'"
        "如果无法找到价格，请明确说明。"
    ),
    agent=financial_analyst_agent,
)

## --- 4. 组建团队 ---
## 团队协调 Agent 和任务如何协同工作。
financial_crew = Crew(
    agents=[financial_analyst_agent],
    tasks=[analyze_aapl_task],
    verbose=True # 在生产环境中设置为 False 以获得较少的详细日志
)

## --- 5. 在主执行块中运行团队 ---
## 使用 __name__ == "__main__": 块是标准 Python 最佳实践。
def main():
    """运行团队的主函数。"""
    # 在启动前检查 API 密钥以避免运行时错误。
    if not os.environ.get("OPENAI_API_KEY"):
        print("错误：未设置 OPENAI_API_KEY 环境变量。")
        print("请在运行脚本之前设置它。")
        return
    
    print("\n## 启动财务团队...")
    print("---------------------------------")
    
    # kickoff 方法启动执行。
    result = financial_crew.kickoff()
    
    print("\n---------------------------------")
    print("## 团队执行完成。")
    print("\n最终结果：\n", result)

if __name__ == "__main__":
    main()

此代码演示了使用 CrewAI 库模拟财务分析任务的简单应用程序。它定义了一个自定义工具 get_stock_price，模拟查找预定义股票代码的价格。该工具设计为对有效代码返回浮点数，或对无效代码引发 ValueError。创建了一个名为 financial_analyst_agent 的 CrewAI Agent，角色为高级财务分析师。该 Agent 被赋予 get_stock_price 工具进行交互。定义了一个任务 analyze_aapl_task，专门指示 Agent 使用工具查找 AAPL 的模拟股票价格。任务描述包括关于使用工具时如何处理成功和失败情况的明确说明。组建了一个团队，包含 financial_analyst_agent 和 analyze_aapl_task。为 Agent 和团队启用了详细设置，以在执行期间提供详细日志。脚本的主要部分在标准 if name == “main“: 块中使用 kickoff() 方法运行团队的任务。在启动团队之前，它会检查是否设置了 OPENAI_API_KEY 环境变量，这是 Agent 运行所必需的。然后将团队执行的结果（即任务的输出）打印到控制台。代码还包括基本日志配置，以更好地跟踪团队的行动和工具调用。它使用环境变量进行 API 密钥管理，尽管它指出对于生产环境建议使用更安全的方法。简而言之，核心逻辑展示了如何定义工具、Agent 和任务，以在 CrewAI 中创建协作工作流。

实操代码（Google ADK）

Google Agent Developer Kit (ADK) 包括一个原生集成工具库，可以直接集成到 Agent 的能力中。

Google 搜索： 此类组件的主要示例是 Google 搜索工具。此工具作为 Google 搜索引擎的直接接口，为 Agent 提供执行网络搜索和检索外部信息的功能。

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
from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types
import nest_asyncio
import asyncio

## 定义会话设置和 Agent 执行所需的变量
APP_NAME="Google Search_agent"
USER_ID="user1234"
SESSION_ID="1234"

## 使用搜索工具定义 Agent
root_agent = ADKAgent(
  name="basic_search_agent",
  model="gemini-2.0-flash-exp",
  description="使用 Google 搜索回答问题的 Agent。",
  instruction="我可以通过搜索互联网回答您的问题。随便问我什么！",
  tools=[google_search] # Google 搜索是执行 Google 搜索的预构建工具。
)

## Agent 交互
async def call_agent(query):
  """
  使用查询调用 Agent 的辅助函数。
  """
  # 会话和运行器
  session_service = InMemorySessionService()
  session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
  runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)
  
  content = types.Content(role='user', parts=[types.Part(text=query)])
  events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
  
  for event in events:
      if event.is_final_response():
          final_response = event.content.parts[0].text
          print("Agent 响应：", final_response)

nest_asyncio.apply()
asyncio.run(call_agent("最新的 AI 新闻是什么？"))

此代码演示了如何使用 Python 的 Google ADK 创建和使用由 Google ADK 驱动的基本 Agent。该 Agent 设计为通过利用 Google 搜索作为工具来回答问题。首先，从 IPython、google.adk 和 google.genai 导入必要的库。定义了应用程序名称、用户 ID 和会话 ID 的常量。创建了一个名为”basic_search_agent”的 Agent 实例，带有描述和说明指示其目的。它被配置为使用 Google 搜索工具，这是 ADK 提供的预构建工具。初始化 InMemorySessionService（见第 8 章）以管理 Agent 的会话。为指定的应用程序、用户和会话 ID 创建新会话。实例化 Runner，将创建的 Agent 与会话服务链接。此运行器负责在会话中执行 Agent 的交互。定义了辅助函数 call_agent 以简化向 Agent 发送查询和处理响应的过程。在 call_agent 内部，用户的查询被格式化为具有角色”user”的 types.Content 对象。使用用户 ID、会话 ID 和新消息内容调用 runner.run 方法。runner.run 方法返回表示 Agent 操作和响应的事件列表。代码遍历这些事件以查找最终响应。如果事件被识别为最终响应，则提取该响应的文本内容。提取的 Agent 响应然后打印到控制台。最后，使用查询”最新的 AI 新闻是什么？”调用 call_agent 函数以演示 Agent 的运行情况。

代码执行： Google ADK 具有用于专门任务的集成组件，包括用于动态代码执行的环境。built_in_code_execution 工具为 Agent 提供沙盒 Python 解释器。这允许模型编写和运行代码以执行计算任务、操作数据结构和执行程序脚本。这种功能对于解决需要确定性逻辑和精确计算的问题至关重要，这超出了概率语言生成本身的范围。

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
import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
import logging

from google.adk.agents import LlmAgent # Removed ADKAgent as it's deprecated
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.adk.code_executors import BuiltInCodeExecutor
from google.genai import types

## 定义会话设置和 Agent 执行所需的变量
APP_NAME = "calculator"
USER_ID = "user1234"
SESSION_ID = "session_code_exec_async"

## Agent 定义
code_agent = LlmAgent(
    name="calculator_agent",
    model="gemini-2.0-flash",
    code_executor=BuiltInCodeExecutor(),
    instruction="""你是一个计算器 Agent。
    当给定数学表达式时，编写并执行 Python 代码来计算结果。
    仅返回最终的数值结果作为纯文本，不带 markdown 或代码块。
    """,
    description="执行 Python 代码以进行计算。",
)

## Agent 交互（异步）
async def call_agent_async(query):
    # 会话和运行器
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    runner = Runner(agent=code_agent, app_name=APP_NAME, session_service=session_service)
    
    content = types.Content(role='user', parts=[types.Part(text=query)])
    
    print(f"\n--- 运行查询：{query} ---")
    final_response_text = "未捕获最终文本响应。"
    
    try:
        # 使用 run_async
        async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
            print(f"事件 ID：{event.id}，作者：{event.author}")
            
            # --- 首先检查特定部分 ---
            # has_specific_part = False
            if event.content and event.content.parts and event.is_final_response():
                for part in event.content.parts: # 遍历所有部分
                    if part.executable_code:
                        # 通过 .code 访问实际代码字符串
                        print(f"  调试：Agent 生成的代码：\n```python\n{part.executable_code.code}\n```")
                        # has_specific_part = True # Removed as it's not used
                    elif part.code_execution_result:
                        # 正确访问结果和输出
                        print(f"  调试：代码执行结果：{part.code_execution_result.outcome} - 输出：\n{part.code_execution_result.output}")
                        # has_specific_part = True # Removed as it's not used
                    # 也打印在任何事件中找到的任何文本部分以进行调试
                    elif part.text and not part.text.isspace():
                        print(f"  文本：'{part.text.strip()}'")
                        # 不要在这里设置 has_specific_part=True，因为我们希望下面的最终响应逻辑
                
                # --- 在特定部分之后检查最终响应 ---
                text_parts = [part.text for part in event.content.parts if part.text]
                final_result = "".join(text_parts)
                print(f"==> 最终 Agent 响应：{final_result}")
    except Exception as e:
        print(f"Agent 运行期间出错：{e}")
    
    print("-" * 30)

## 运行示例的主异步函数
async def main():
    await call_agent_async("计算 (5 + 7) * 3 的值")
    await call_agent_async("10 的阶乘是多少？")

## 执行主异步函数
try:
    nest_asyncio.apply()
    asyncio.run(main())
except RuntimeError as e:
    # 在已运行的循环中运行 asyncio.run 时处理特定错误（如 Jupyter/Colab）
    if "cannot be called from a running event loop" in str(e):
        print("\n在现有事件循环中运行（如 Colab/Jupyter）。")
        print("请在笔记本单元格中运行 `await main()` 代替。")
        # 如果在交互式环境（如笔记本）中，您可能需要运行：
        # await main()
    else:
        raise e # 重新引发其他运行时错误

此脚本使用 Google 的 Agent Development Kit (ADK) 创建一个通过编写和执行 Python 代码解决数学问题的 Agent。它定义了一个 LlmAgent，专门指示其充当计算器，为其配备 built_in_code_execution 工具。主要逻辑位于 call_agent_async 函数中，该函数向 Agent 的运行器发送用户查询并处理结果事件。在此函数内部，异步循环遍历事件，打印生成的 Python 代码及其执行结果以进行调试。代码仔细区分这些中间步骤和包含数值答案的最终事件。最后，main 函数使用两个不同的数学表达式运行 Agent，以演示其执行计算的能力。

企业搜索： 此代码使用 Python 中的 google.adk 库定义了一个 Google ADK 应用程序。它专门使用 VSearchAgent，该 Agent 旨在通过搜索指定的 Vertex AI 搜索数据存储来回答问题。代码初始化一个名为”q2_strategy_vsearch_agent”的 VSearchAgent，提供描述、要使用的模型（”gemini-2.0-flash-exp”）和 Vertex AI 搜索数据存储的 ID。DATASTORE_ID 预期设置为环境变量。然后为 Agent 设置 Runner，使用 InMemorySessionService 管理对话历史。定义了异步函数 call_vsearch_agent_async 以与 Agent 交互。此函数接受查询，构造消息内容对象，并调用运行器的 run_async 方法将查询发送到 Agent。然后该函数将 Agent 的响应流式传输回控制台。它还打印有关最终响应的信息，包括来自数据存储的任何源归因。包含错误处理以捕获 Agent 执行期间的异常，提供有关潜在问题（如数据存储 ID 不正确或缺少权限）的信息性消息。提供了另一个异步函数 run_vsearch_example 以演示如何使用示例查询调用 Agent。主执行块检查 DATASTORE_ID 是否已设置，然后使用 asyncio.run 运行示例。它包括检查以处理在已有运行事件循环的环境（如 Jupyter 笔记本）中运行代码的情况。

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
import asyncio
from google.genai import types
from google.adk import agents
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
import os

## --- 配置 ---
## 确保您已设置 GOOGLE_API_KEY 和 DATASTORE_ID 环境变量
## 例如：
## os.environ["GOOGLE_API_KEY"] = "YOUR_API_KEY"
## os.environ["DATASTORE_ID"] = "YOUR_DATASTORE_ID"

DATASTORE_ID = os.environ.get("DATASTORE_ID")

## --- 应用程序常量 ---
APP_NAME = "vsearch_app"
USER_ID = "user_123"  # 示例用户 ID
SESSION_ID = "session_456" # 示例会话 ID

## --- Agent 定义（根据指南的示例更新） ---
vsearch_agent = agents.VSearchAgent(
    name="q2_strategy_vsearch_agent",
    description="使用 Vertex AI 搜索回答有关 Q2 战略文档的问题。",
    model="gemini-2.0-flash-exp", # 根据指南示例更新模型
    datastore_id=DATASTORE_ID,
    model_parameters={"temperature": 0.0}
)

## --- 运行器和会话初始化 ---
runner = Runner(
    agent=vsearch_agent,
    app_name=APP_NAME,
    session_service=InMemorySessionService(),
)

## --- Agent 调用逻辑 ---
async def call_vsearch_agent_async(query: str):
    """初始化会话并流式传输 Agent 的响应。"""
    print(f"用户：{query}")
    print("Agent：", end="", flush=True)
    
    try:
        # 正确构造消息内容
        content = types.Content(role='user', parts=[types.Part(text=query)])
        
        # 从异步运行器处理到达的事件
        async for event in runner.run_async(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
        ):
            # 用于响应文本的逐令牌流式传输
            if hasattr(event, 'content_part_delta') and event.content_part_delta:
                print(event.content_part_delta.text, end="", flush=True)
            
            # 处理最终响应及其关联的元数据
            if event.is_final_response():
                print() # 流式响应后换行
                if event.grounding_metadata:
                    print(f"  (来源归因：找到 {len(event.grounding_metadata.grounding_attributions)} 个来源)")
                else:
                    print("  (未找到基础元数据)")
                print("-" * 30)
    except Exception as e:
        print(f"\n发生错误：{e}")
        print("请确保您的数据存储 ID 正确，并且服务帐户具有必要的权限。")
        print("-" * 30)

## --- 运行示例 ---
async def run_vsearch_example():
    # 替换为与您的数据存储内容相关的问题
    await call_vsearch_agent_async("总结 Q2 战略文档的要点。")
    await call_vsearch_agent_async("实验室 X 提到了哪些安全程序？")

## --- 执行 ---
if __name__ == "__main__":
    if not DATASTORE_ID:
        print("错误：未设置 DATASTORE_ID 环境变量。")
    else:
        try:
            asyncio.run(run_vsearch_example())
        except RuntimeError as e:
            # 这处理在已有运行事件循环的环境中调用 asyncio.run 的情况
            # （如 Jupyter 笔记本）。
            if "cannot be called from a running event loop" in str(e):
                print("在运行事件循环中跳过执行。请直接运行此脚本。")
            else:
                raise e

总的来说，此代码为构建利用 Vertex AI 搜索根据存储在数据存储中的信息回答问题的对话式 AI 应用程序提供了基本框架。它演示了如何定义 Agent、设置运行器以及在流式传输响应的同时异步与 Agent 交互。重点是从特定数据存储检索和综合信息以回答用户查询。

Vertex Extensions： Vertex AI 扩展是一个结构化的 API 包装器，使模型能够连接到外部 API 以进行实时数据处理和操作执行。扩展提供企业级安全性、数据隐私和性能保证。它们可用于生成和运行代码、查询网站以及分析来自私有数据存储的信息等任务。Google 为常见用例提供预构建扩展，如代码解释器和 Vertex AI 搜索，并可选择创建自定义扩展。扩展的主要好处包括强大的企业控制和与其他 Google 产品的无缝集成。扩展和函数调用之间的关键区别在于它们的执行：Vertex AI 自动执行扩展，而函数调用需要用户或客户端手动执行。

概览

是什么： 大型语言模型（LLM）是强大的文本生成器，但它们基本上与外部世界断开连接。它们的知识是静态的，仅限于训练数据，并且缺乏执行操作或检索实时信息的能力。这种固有的限制阻止它们完成需要与外部 API、数据库或服务交互的任务。没有通往这些外部系统的桥梁，它们解决现实世界问题的效用受到严重限制。

为什么： 工具使用模式（通常通过函数调用实现）为此问题提供了标准化解决方案。它的工作原理是以 LLM 可以理解的方式向其描述可用的外部函数或”工具”。基于用户的请求，Agent LLM 可以决定是否需要工具，并生成指定要调用哪个函数以及使用什么参数的结构化数据对象（如 JSON）。编排层执行此函数调用，检索结果，并将其反馈给 LLM。这允许 LLM 将最新的外部信息或操作结果合并到其最终响应中，有效地赋予其行动能力。

经验法则： 当 Agent 需要突破 LLM 的内部知识并与外部世界交互时，使用工具使用模式。这对于需要实时数据（例如，检查天气、股票价格）、访问私有或专有信息（例如，查询公司数据库）、执行精确计算、执行代码或触发其他系统中的操作（例如，发送电子邮件、控制智能设备）的任务至关重要。

视觉摘要：

图 2：工具使用设计模式

关键要点

  • 工具使用（函数调用）允许 Agent 与外部系统交互并访问动态信息。
  • 它涉及定义具有 LLM 可以理解的清晰描述和参数的工具。
  • LLM 决定何时使用工具并生成结构化函数调用。
  • Agent 框架执行实际的工具调用并将结果返回给 LLM。
  • 工具使用对于构建可以执行现实世界操作并提供最新信息的 Agent 至关重要。
  • LangChain 使用 @tool 装饰器简化工具定义，并提供 create_tool_calling_agent 和 AgentExecutor 用于构建工具使用 Agent。
  • Google ADK 有许多非常有用的预构建工具，如 Google 搜索、代码执行和 Vertex AI 搜索工具。

结论

工具使用模式是将大型语言模型的功能范围扩展到其固有文本生成能力之外的关键架构原则。通过为模型配备与外部软件和数据源交互的能力，此范式允许 Agent 执行操作、进行计算并从其他系统检索信息。此过程涉及模型在确定满足用户查询需要时生成调用外部工具的结构化请求。LangChain、Google ADK 和 CrewAI 等框架提供结构化抽象和组件，促进这些外部工具的集成。这些框架管理向模型公开工具规范并解析其后续工具使用请求的过程。这简化了可以与外部数字环境交互并在其中采取行动的复杂 Agent 系统的开发。

参考文献

  1. LangChain Documentation (Tools): https://python.langchain.com/docs/integrations/tools/
  2. Google Agent Developer Kit (ADK) Documentation (Tools): https://google.github.io/adk-docs/tools/
  3. OpenAI Function Calling Documentation: https://platform.openai.com/docs/guides/function-calling
  4. CrewAI Documentation (Tools): https://docs.crewai.com/concepts/tools

展开全文 >>

Agentic Design Patterns - Chapter 4_ Reflection

第 4 章：反思

反思模式概述

在前面的章节中，我们探讨了基础的 Agent 模式：顺序执行的链式、动态路径选择的路由以及并发任务执行的并行化。这些模式使 Agent 能够更高效、更灵活地执行复杂任务。然而，即使采用复杂的工作流，Agent 的初始输出或计划也可能并非最优、准确或完整。这正是反思模式发挥关键作用之处。

反思模式涉及 Agent 评估其自身工作、输出或内部状态，并利用该评估来提升性能或优化响应。这是一种自我纠正或自我改进机制，允许 Agent 基于反馈、内部评审或与预期标准的对比，迭代优化其输出或调整策略。反思有时可由专门的 Agent 来促进，其特定职责是分析初始 Agent 的输出。

与简单顺序链（输出直接传递至下一步）或路径选择的路由不同，反思引入了反馈循环。Agent 不仅产生输出，还会检查该输出（或其生成过程），识别潜在问题或改进空间，并运用这些洞察生成更好的版本或调整后续行动。

该过程通常包含以下步骤：

  1. 执行： Agent 执行任务或生成初始输出
  2. 评估/评审： Agent（通常通过另一个 LLM 调用或规则集）分析上一步结果。此评估可能涉及事实准确性、连贯性、风格、完整性、指令遵循度或其他相关标准
  3. 反思/优化： 基于评审意见，Agent 确定改进方向。这可能包括生成优化后的输出、调整后续步骤参数，甚至修改整体计划
  4. 迭代（可选但常见）： 优化后的输出或调整后的方法可继续执行，反思过程可重复进行，直至获得满意结果或达到停止条件

反思模式的一个关键且高效的实现是将流程分离为两个不同的逻辑角色：生产者和评审者。这通常被称为”生成器-评审者”或”生产者-审查者”模型。虽然单个 Agent 可执行自我反思，但使用两个专门 Agent（或两个具有不同系统提示的独立 LLM 调用）通常能产生更稳健和客观的结果。

  1. 生产者 Agent：此 Agent 的主要职责是执行任务的初始工作。它完全专注于内容生成，无论是编写代码、起草博客文章还是制定计划。它接收初始提示并生成输出的第一版

  2. 评审者 Agent：此 Agent 的唯一目的是评估生产者生成的输出。它被赋予不同的指令集，通常承担特定角色（如”高级软件工程师”、”严谨的事实核查员”）。评审者的指令引导其根据特定标准分析生产者工作，包括事实准确性、代码质量、风格要求或完整性。其设计目标是发现缺陷、提出改进建议并提供结构化反馈

这种关注点分离非常有效，因为它避免了 Agent 审查自身工作时可能产生的”认知偏差”。评审者 Agent 以全新视角处理输出，完全专注于发现错误和改进空间。评审者的反馈随后传回生产者 Agent，生产者 Agent 以此为指南生成新的优化版本。提供的 LangChain 和 ADK 代码示例均实现了这种双 Agent 模型：LangChain 示例使用特定的”reflector_prompt”创建评审者角色，而 ADK 示例明确定义了生产者和审查者 Agent。

实现反思通常需要在 Agent 工作流中构建这些反馈循环。这可通过代码中的迭代循环实现，或使用支持基于评估结果进行状态管理和条件转换的框架。虽然单步评估和优化可在 LangChain/LangGraph、ADK 或 Crew.AI 链中实现，但真正的迭代反思通常涉及更复杂的编排。

反思模式对于构建能够产出高质量输出、处理精细任务并展现一定自我意识和适应性的 Agent 至关重要。它推动 Agent 从单纯执行指令转向更复杂的问题解决和内容生成形式。

值得注意的是反思与目标设定和监控（见第 11 章）的交叉点。目标为 Agent 的自我评估提供最终基准，而监控跟踪其进展。在许多实际案例中，反思可能充当纠正引擎，利用监控反馈分析偏差并调整策略。这种协同作用将 Agent 从被动执行者转变为有目的地自适应工作以实现目标。

此外，当 LLM 保持对话记忆（见第 8 章）时，反思模式的有效性显著增强。对话历史为评估阶段提供关键上下文，使 Agent 不仅能孤立评估输出，还能在先前交互、用户反馈和演进目标的背景下进行评估。这使 Agent 能从过去的评审中学习并避免重复错误。没有记忆时，每次反思都是独立事件；有了记忆，反思成为累积过程，每个周期都基于前一个周期，从而实现更智能和上下文感知的优化。

实际应用与用例

反思模式在输出质量、准确性或复杂约束遵循度至关重要的场景中极具价值：

  1. 创意写作与内容生成：
    优化生成的文本、故事、诗歌或营销文案
  • 用例： 博客文章撰写 Agent
    • 反思： 生成草稿，评审其流畅度、语气和清晰度，然后基于评审重写。重复直至内容达到质量标准
    • 优势： 产出更精炼有效的内容
  1. 代码生成与调试：
    编写代码、识别错误并进行修复
  • 用例： Python 函数编写 Agent
    • 反思： 编写初始代码，运行测试或静态分析，识别错误或低效环节，然后基于发现修改代码
    • 优势： 生成更健壮和功能完善的代码
  1. 复杂问题解决：
    评估多步推理任务中的中间步骤或建议方案
  • 用例： 逻辑谜题求解 Agent
    • 反思： 提出步骤，评估其是否接近解决方案或引入矛盾，必要时回溯或选择不同步骤
    • 优势： 提升 Agent 在复杂问题空间中的导航能力
  1. 摘要与信息综合：
    优化摘要的准确性、完整性和简洁性
  • 用例： 长文档摘要 Agent
    • 反思： 生成初始摘要，与原始文档关键点对比，优化摘要以补全缺失信息或提升准确性
    • 优势： 创建更准确全面的摘要
  1. 规划与策略：
    评估提议计划并识别潜在缺陷或改进点
  • 用例： 目标达成行动规划 Agent
    • 反思： 生成计划，模拟执行或根据约束评估可行性，基于评估修订计划
    • 优势： 制定更有效和现实的计划
  1. 对话 Agent：
    审查对话历史轮次以保持上下文、纠正误解或提升响应质量
  • 用例： 客户支持聊天机器人
    • 反思： 用户响应后，审查对话历史和最后生成消息，确保连贯性并准确回应用户最新输入
    • 优势： 促成更自然有效的对话

反思为 Agent 系统增加了元认知层，使其能从自身输出和过程中学习，从而产生更智能、可靠和高质量的结果

实操代码示例（LangChain）

实现完整的迭代反思过程需要状态管理和循环执行机制。虽然这些在基于图的框架（如 LangGraph）中内置处理或通过自定义程序代码实现，但单个反思周期的基本原理可通过 LCEL（LangChain 表达式语言）的组合语法有效演示。

此示例使用 Langchain 库和 OpenAI 的 GPT-4o 模型实现反思循环，迭代生成并优化计算数字阶乘的 Python 函数。该过程从任务提示开始，生成初始代码，然后基于模拟高级软件工程师角色的评审反复反思代码，在每次迭代中优化代码，直至评审阶段确认代码完美或达到最大迭代次数。最后打印优化后的代码。

首先确保安装必要库：

1
pip install langchain langchain-community langchain-openai

您还需要使用所选语言模型的 API 密钥配置环境（如 OpenAI、Google Gemini、Anthropic）

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
import os
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.messages import SystemMessage, HumanMessage

## --- 配置 ---
## 从 .env 文件加载环境变量（用于 OPENAI_API_KEY）
load_dotenv()

## 检查是否设置了 API 密钥
if not os.getenv("OPENAI_API_KEY"):
    raise ValueError("在 .env 文件中未找到 OPENAI_API_KEY。请添加它。")

## 初始化聊天 LLM。我们使用 gpt-4o 以获得更好的推理。
## 使用较低的温度以获得更确定性的输出。
llm = ChatOpenAI(model="gpt-4o", temperature=0.1)

def run_reflection_loop():
    """
    演示多步 AI 反思循环以逐步改进 Python 函数。
    """
    # --- 核心任务 ---
    task_prompt = """
    你的任务是创建一个名为 `calculate_factorial` 的 Python 函数。
    此函数应执行以下操作：
    1. 接受单个整数 `n` 作为输入。
    2. 计算其阶乘 (n!)。
    3. 包含清楚解释函数功能的文档字符串。
    4. 处理边缘情况：0 的阶乘是 1。
    5. 处理无效输入：如果输入是负数，则引发 ValueError。
    """
    
    # --- 反思循环 ---
    max_iterations = 3
    current_code = ""
    
    # 我们将构建对话历史以在每一步中提供上下文。
    message_history = [HumanMessage(content=task_prompt)]
    
    for i in range(max_iterations):
        print("\n" + "="*25 + f" 反思循环：迭代 {i + 1} " + "="*25)
        
        # --- 1. 生成/完善阶段 ---
        # 在第一次迭代中，它生成。在后续迭代中，它完善。
        if i == 0:
            print("\n>>> 阶段 1：生成初始代码...")
            # 第一条消息只是任务提示词。
            response = llm.invoke(message_history)
            current_code = response.content
        else:
            print("\n>>> 阶段 1：基于先前批评完善代码...")
            # 消息历史现在包含任务、
            # 最后一个代码和最后一个批评。
            # 我们指示模型应用批评。
            message_history.append(HumanMessage(content="请使用提供的批评完善代码。"))
            response = llm.invoke(message_history)
            current_code = response.content
        
        print("\n--- 生成的代码 (v" + str(i + 1) + ") ---\n" + current_code)
        message_history.append(response) # 将生成的代码添加到历史记录
        
        # --- 2. 反思阶段 ---
        print("\n>>> 阶段 2：对生成的代码进行反思...")
        # 为反思 Agent 创建特定提示词。
        # 这要求模型充当高级代码审查员。
        reflector_prompt = [
            SystemMessage(content="""
                你是一名高级软件工程师和 Python 专家。
                你的角色是执行细致的代码审查。
                根据原始任务要求批判性地评估提供的 Python 代码。
                查找错误、风格问题、缺失的边缘情况和改进领域。
                如果代码完美并满足所有要求，
                用单一短语 'CODE_IS_PERFECT' 响应。
                否则，提供批评的项目符号列表。
            """),
            HumanMessage(content=f"原始任务：\n{task_prompt}\n\n要审查的代码：\n{current_code}")
        ]
        
        critique_response = llm.invoke(reflector_prompt)
        critique = critique_response.content
        
        # --- 3. 停止条件 ---
        if "CODE_IS_PERFECT" in critique:
            print("\n--- 批评 ---\n未发现进一步批评。代码令人满意。")
            break
        
        print("\n--- 批评 ---\n" + critique)
        
        # 将批评添加到历史记录以用于下一个完善循环。
        message_history.append(HumanMessage(content=f"对先前代码的批评：\n{critique}"))
    
    print("\n" + "="*30 + " 最终结果 " + "="*30)
    print("\n反思过程后的最终精炼代码：\n")
    print(current_code)

if __name__ == "__main__":
    run_reflection_loop()

代码首先设置环境，加载 API 密钥，并使用低温度初始化强大的语言模型（如 GPT-4o）以获得专注输出。核心任务由提示定义，要求创建计算数字阶乘的 Python 函数，包括文档字符串、边界情况（0 的阶乘）和负输入错误处理的特定要求。run_reflection_loop 函数协调迭代优化过程。在循环中，第一次迭代时语言模型根据任务提示生成初始代码，后续迭代中则基于前一步的评审优化代码。单独的”反思者”角色（同样由语言模型扮演但使用不同系统提示）充当高级软件工程师，根据原始任务要求评审生成的代码。此评审以问题项目符号列表或短语 ‘CODE_IS_PERFECT’（如无问题）形式提供。循环持续至评审指示代码完美或达到最大迭代次数。对话历史被维护并在每一步传递给语言模型，为生成/优化和反思阶段提供上下文。最后，脚本在循环结束后打印最终生成的代码版本

实操代码示例（ADK）

现在让我们看一个使用 Google ADK 实现的概念性代码示例。具体而言，代码通过采用生成器-评审者结构来展示，其中一个组件（生成器）产生初始结果或计划，另一个组件（评审者）提供批判性反馈或评审，引导生成器朝向更优化或准确的最终输出

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
from google.adk.agents import SequentialAgent, LlmAgent

## 第一个 Agent 生成初始草稿。
generator = LlmAgent(
    name="DraftWriter",
    description="生成关于给定主题的初始草稿内容。",
    instruction="撰写关于用户主题的简短、信息丰富的段落。",
    output_key="draft_text" # 输出保存到此状态键。
)

## 第二个 Agent 评审第一个 Agent 的草稿。
reviewer = LlmAgent(
    name="FactChecker",
    description="审查给定文本的事实准确性并提供结构化评审。",
    instruction="""
    你是一个细致的事实核查员。
    1. 阅读状态键 'draft_text' 中提供的文本。
    2. 仔细验证所有声明的事实准确性。
    3. 你的最终输出必须是包含两个键的字典：
       - "status": 字符串，"ACCURATE" 或 "INACCURATE"。
       - "reasoning": 字符串，提供对你的状态的清楚解释，如果发现任何问题则引用具体问题。
    """,
    output_key="review_output" # 结构化字典保存在这里。
)

## SequentialAgent 确保生成器在审查者之前运行。
review_pipeline = SequentialAgent(
    name="WriteAndReview_Pipeline",
    sub_agents=[generator, reviewer]
)

## 执行流程：
## 1. generator 运行 -> 将其段落保存到 state['draft_text']。
## 2. reviewer 运行 -> 读取 state['draft_text'] 并将其字典输出保存到 state['review_output']。

此代码演示了在 Google ADK 中使用顺序 Agent 管道生成和审查文本。它定义了两个 LlmAgent 实例：generator 和 reviewer。generator Agent 设计用于创建关于给定主题的初始草稿段落，被指示撰写简短信息丰富的文章，并将其输出保存至状态键 draft_text。reviewer Agent 作为生成器产出文本的事实核查员，被指示从 draft_text 读取文本并验证其事实准确性。评审者的输出是包含两个键的结构化字典：status 和 reasoning。status 指示文本为”ACCURATE”或”INACCURATE”，reasoning 则提供状态解释。此字典保存至状态键 review_output。创建名为 review_pipeline 的 SequentialAgent 来管理两个 Agent 的执行顺序，确保生成器先运行，然后是评审者。整体执行流程为：生成器产出文本并保存至状态，随后评审者从状态读取文本，执行事实核查，并将其发现（状态和推理）保存回状态。此管道允许使用独立 Agent 进行结构化内容创建和审查过程。注意： 对于感兴趣者，还提供了利用 ADK 的 LoopAgent 的替代实现。

结束前需考虑，虽然反思模式显著提升输出质量，但也带来重要权衡。迭代过程虽强大，但可能导致更高成本和延迟，因为每个优化循环可能需要新的 LLM 调用，使其对时间敏感应用并非最优选择。此外，该模式内存密集；随着每次迭代，对话历史扩展，包含初始输出、评审和后续优化

概览

是什么： Agent 的初始输出通常次优，存在不准确、不完整或未满足复杂要求的问题。基础 Agent 工作流缺乏 Agent 识别和修复自身错误的内置流程。这通过让 Agent 评估自身工作，或更稳健地引入独立逻辑 Agent 充当评审者来解决，防止无论质量如何初始响应都成为最终结果

为什么： 反思模式通过引入自我纠正和优化机制提供解决方案。它建立反馈循环，其中”生产者” Agent 生成输出，然后”评审者” Agent（或生产者自身）根据预定义标准进行评估。随后使用此评审生成改进版本。这种生成、评估和优化的迭代过程逐步提升最终结果质量，从而产生更准确、连贯和可靠的结果

经验法则： 当最终输出的质量、准确性和细节比速度和成本更重要时使用反思模式。它对生成精炼长篇内容、编写调试代码以及创建详细计划等任务特别有效。当任务需要通用生产者 Agent 可能遗漏的高客观性或专门评估时，使用独立评审者 Agent

视觉摘要

图 1：反思设计模式，自我反思

图 2：反思设计模式，生产者和评审者 Agent

关键要点

  • 反思模式的主要优势在于其迭代自我纠正和优化输出的能力，带来显著更高的质量、准确性和复杂指令遵循度
  • 它涉及执行、评估/评审和优化的反馈循环。反思对需要高质量、准确或精细输出的任务至关重要
  • 一个强大的实现是生产者-评审者模型，其中独立 Agent（或提示角色）评估初始输出。这种关注点分离增强客观性，并支持更专业、结构化的反馈
  • 然而，这些优势以增加的延迟和计算成本为代价，同时伴随超出模型上下文窗口或被 API 服务限制的更高风险
  • 虽然完整迭代反思通常需要状态化工作流（如 LangGraph），但单个反思步骤可在 LangChain 中使用 LCEL 实现，以传递输出进行评审和后续优化
  • Google ADK 可通过顺序工作流促进反思，其中一个 Agent 的输出被另一个 Agent 评审，允许后续优化步骤
  • 此模式使 Agent 能够执行自我纠正并随时间提升性能

结论

反思模式为 Agent 工作流中的自我纠正提供关键机制，实现超越单次执行的迭代改进。这是通过创建循环实现的：系统生成输出，根据特定标准评估，然后使用该评估产生优化结果。此评估可由 Agent 自身（自我反思）执行，或通常更有效地由不同评审者 Agent 执行，这代表了模式内的关键架构选择

虽然完全自主的多步反思过程需要强大的状态管理架构，但其核心原理可在单个生成-评审-优化周期中有效演示。作为控制结构，反思可与其他基础模式集成，以构建更健壮和功能更复杂的 Agent 系统

参考文献

以下是有关反思模式和相关概念的一些进一步阅读资源：

  1. Training Language Models to Self-Correct via Reinforcement Learning, https://arxiv.org/abs/2409.12917
  2. LangChain Expression Language (LCEL) Documentation: https://python.langchain.com/docs/introduction/
  3. LangGraph Documentation:https://www.langchain.com/langgraph
  4. Google Agent Developer Kit (ADK) Documentation (Multi-Agent Systems): https://google.github.io/adk-docs/agents/multi-agents/

展开全文 >>

Agentic Design Patterns - Chapter 3_ Parallelization

第 3 章：并行化

并行化模式概述

在前面的章节中，我们探讨了用于顺序工作流的提示链和用于动态决策及不同路径间转换的路由。虽然这些模式不可或缺，但许多复杂的 Agent 任务涉及多个可同时执行而非顺序执行的子任务。这正是并行化模式变得至关重要的场景。

并行化涉及并发执行多个组件，包括 LLM 调用、工具使用乃至整个子 Agent（见图 1）。与等待一个步骤完成再开始下一个步骤不同，并行执行允许独立任务同时运行，从而显著缩短可分解为独立部分的任务的总执行时间。

设想一个旨在研究主题并总结其发现的 Agent。顺序方法可能如下：

  1. 搜索来源 A
  2. 总结来源 A
  3. 搜索来源 B
  4. 总结来源 B
  5. 基于摘要 A 和 B 合成最终答案

而并行方法则可改为：

  1. 同时搜索来源 A 和来源 B
  2. 两个搜索完成后，同时总结来源 A 和来源 B
  3. 基于摘要 A 和 B 合成最终答案（此步骤通常为顺序执行，需等待并行步骤完成）

其核心思想在于识别工作流中不依赖其他部分输出的环节并并行执行。这在处理具有延迟的外部服务（如 API 或数据库）时尤其有效，因为您可以并发发出多个请求。

实现并行化通常需要支持异步执行或多线程/多进程的框架。现代 Agent 框架在设计时已充分考虑异步操作，使您能够轻松定义可并行运行的步骤。

图 1. 带有子 Agent 的并行化示例

像 LangChain、LangGraph 和 Google ADK 这样的框架提供了并行执行机制。在 LangChain 表达式语言（LCEL）中，您可以通过使用运算符（如 | 用于顺序）组合可运行对象来实现并行执行，并通过构建支持分支并发执行的链或图。LangGraph 凭借其图结构，允许您定义可从单个状态转换执行的多个节点，从而在工作流中有效启用并行分支。Google ADK 提供了强大的原生机制来促进和管理 Agent 的并行执行，显著提升了复杂多 Agent 系统的效率和可扩展性。ADK 框架的这一固有能力使开发人员能够设计并实现多个 Agent 并发而非顺序操作的解决方案。

并行化模式对于提升 Agent 系统的效率和响应能力至关重要，特别是在处理涉及多个独立查询、计算或与外部服务交互的任务时。这是优化复杂 Agent 工作流性能的关键技术。

实际应用与用例

并行化是优化 Agent 性能的强大模式，适用于多种应用场景：

  1. 信息收集与研究：
    同时从多个来源收集信息是经典用例。
  • 用例： 公司调研 Agent
    • 并行任务： 同步搜索新闻文章、获取股票数据、监控社交媒体提及和查询公司数据库
    • 优势： 相比顺序查询能更快获得全面视图
  1. 数据处理与分析：
    对数据的不同部分或应用不同分析技术进行并发处理
  • 用例： 客户反馈分析 Agent
    • 并行任务： 对一批反馈条目同时执行情感分析、关键词提取、反馈分类和紧急问题识别
    • 优势： 快速提供多维度分析结果
  1. 多 API 或工具交互：
    调用多个独立的 API 或工具以获取不同类型信息或执行不同操作
  • 用例： 旅行规划 Agent
    • 并行任务： 并发查询航班价格、酒店可用性、当地活动和餐厅推荐
    • 优势： 更快呈现完整的旅行方案
  1. 多组件内容生成：
    并行生成复杂内容的不同组成部分
  • 用例： 营销邮件创建 Agent
    • 并行任务： 同时生成邮件主题、起草正文内容、查找相关图片和创建行动号召按钮文本
    • 优势： 更高效地组装最终邮件
  1. 验证与核实：
    并发执行多个独立检查或验证操作
  • 用例： 用户输入验证 Agent
    • 并行任务： 同时验证邮箱格式、手机号码、地址信息数据库匹配和不当内容检测
    • 优势： 更快提供输入有效性反馈
  1. 多模态处理：
    并发处理同一输入的不同模态（文本、图像、音频）
  • 用例： 社交媒体帖子分析 Agent（包含文本和图像）
    • 并行任务： 同时分析文本情感与关键词并识别图像中的对象和场景描述
    • 优势： 更快速整合来自不同模态的洞察
  1. A/B 测试或多选项生成：
    并行生成多个响应或输出变体以选择最佳方案
  • 用例： 创意文本生成 Agent
    • 并行任务： 使用略有不同的提示或模型同时为文章生成三个不同标题
    • 优势： 便于快速比较并选择最佳选项

并行化是 Agent 设计中的基础优化技术，使开发人员能够通过利用独立任务的并发执行来构建更高性能和响应更快的应用程序。

实操代码示例（LangChain）

LangChain 框架中的并行执行由 LangChain 表达式语言（LCEL）实现。主要方法是在字典或列表结构中组织多个可运行组件。当该集合作为输入传递给链中的后续组件时，LCEL 运行时会并发执行其中包含的可运行对象。

在 LangGraph 中，这一原理应用于图的拓扑结构。并行工作流通过设计图结构来定义，使得多个无直接顺序依赖关系的节点可从单个公共节点启动。这些并行路径独立执行，其结果可在图中的后续汇聚点进行聚合。

以下实现展示了使用 LangChain 框架构建的并行处理工作流。该工作流设计用于响应单个用户查询并发执行多个独立操作。这些并行过程被实例化为不同的链或函数，各自的输出随后被聚合成统一结果。

此实现的先决条件包括安装必要的 Python 包，如 langchain、langchain-community 以及模型提供商库（如 langchain-openai）。此外，还需在本地环境中配置所选语言模型的有效 API 密钥以完成身份验证。

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
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 密钥环境变量（例如，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. 定义要并行运行的任务块。这些的结果，
##    以及原始主题，将被馈送到下一步。
map_chain = RunnableParallel(
    {
        "summary": summarize_chain,
        "questions": questions_chain,
        "key_terms": terms_chain,
        "topic": RunnablePassthrough(),  # 传递原始主题
    }
)

## 2. 定义将组合并行结果的最终综合提示词。
synthesis_prompt = ChatPromptTemplate.from_messages([
    ("system", """基于以下信息：
    摘要：{summary}
    相关问题：{questions}
    关键术语：{key_terms}
    综合一个全面的答案。"""),
    ("user", "原始主题：{topic}")
])

## 3. 通过将并行结果直接管道化
##    到综合提示词，然后是 LLM 和输出解析器，构建完整链。
full_parallel_chain = map_chain | synthesis_prompt | llm | StrOutputParser()

## --- 运行链 ---
async def run_parallel_example(topic: str) -> None:
    """
    异步调用具有特定主题的并行处理链
    并打印综合结果。
    参数：
        topic: 要由 LangChain 链处理的输入主题。
    """
    if not llm:
        print("LLM 未初始化。无法运行示例。")
        return
    
    print(f"\n--- 运行主题的并行 LangChain 示例：'{topic}' ---")
    try:
        # `ainvoke` 的输入是单个 'topic' 字符串，
        # 然后传递给 `map_chain` 中的每个可运行对象。
        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))

提供的 Python 代码实现了一个 LangChain 应用程序，旨在通过并行执行高效处理给定主题。请注意，asyncio 提供的是并发性而非真正的并行性。它通过事件循环在单线程上实现这一点，在任务空闲时（如等待网络请求）智能地在任务间切换。这创造了多个任务同时进行的效果，但代码实际上仍由单个线程执行，受 Python 全局解释器锁（GIL）限制。

代码首先从 langchain_openai 和 langchain_core 导入必要模块，包括语言模型、提示模板、输出解析器和可运行结构组件。代码尝试初始化 ChatOpenAI 实例，使用”gpt-4o-mini”模型并设置温度参数以控制创造性。语言模型初始化采用 try-except 块以确保健壮性。随后定义三个独立的 LangChain”链”，每个链设计用于对输入主题执行不同任务：第一个链负责简洁总结主题，第二个链生成与主题相关的三个有趣问题，第三个链从输入主题中识别 5-10 个关键术语并以逗号分隔。每个独立链都由针对特定任务定制的 ChatPromptTemplate、初始化的语言模型和用于字符串格式输出的 StrOutputParser 组成。

然后构建 RunnableParallel 块来捆绑这三个链以实现同时执行。此并行可运行对象还包含 RunnablePassthrough 以确保原始输入主题可用于后续步骤。为最终综合步骤定义了单独的 ChatPromptTemplate，接收摘要、问题、关键术语和原始主题作为输入以生成全面答案。完整的端到端处理链（full_parallel_chain）通过将 map_chain（并行块）连接到综合提示模板，再接入语言模型和输出解析器来创建。提供的异步函数 run_parallel_example 演示了如何调用此完整并行链，该函数接收主题作为输入并使用 invoke 运行异步链。最后，标准 Python if name == “main“: 块展示了如何使用示例主题”太空探索的历史”执行 run_parallel_example，通过 asyncio.run 管理异步执行。

本质上，此代码建立了一个工作流，对给定主题同时执行多个 LLM 调用（用于总结、生成问题和提取术语），最终通过另一个 LLM 调用整合结果。这展示了在 Agent 工作流中使用 LangChain 实现并行化的核心思想。

实操代码示例（Google ADK）

现在让我们关注 Google ADK 框架中说明这些概念的具体示例。我们将探讨如何应用 ADK 原语（如 ParallelAgent 和 SequentialAgent）来构建利用并发执行提升效率的 Agent 流程。

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
from google.adk.agents import LlmAgent, ParallelAgent, SequentialAgent
from google.adk.tools import google_search

GEMINI_MODEL = "gemini-2.0-flash"

## --- 1. 定义研究员子 Agent（并行运行）---
## 研究员 1：可再生能源
researcher_agent_1 = LlmAgent(
    name="RenewableEnergyResearcher",
    model=GEMINI_MODEL,
    instruction="""你是一名专门研究能源的 AI 研究助理。研究"可再生能源"的最新进展。使用提供的 Google 搜索工具。简洁地总结你的主要发现（1-2 句话）。*只*输出摘要。""",
    description="研究可再生能源。",
    tools=[google_search],
    # 将结果存储在状态中供合并 Agent 使用
    output_key="renewable_energy_result"
)

## 研究员 2：电动汽车
researcher_agent_2 = LlmAgent(
    name="EVResearcher",
    model=GEMINI_MODEL,
    instruction="""你是一名专门研究交通的 AI 研究助理。研究"电动汽车技术"的最新发展。使用提供的 Google 搜索工具。简洁地总结你的主要发现（1-2 句话）。*只*输出摘要。""",
    description="研究电动汽车技术。",
    tools=[google_search],
    # 将结果存储在状态中供合并 Agent 使用
    output_key="ev_technology_result"
)

## 研究员 3：碳捕获
researcher_agent_3 = LlmAgent(
    name="CarbonCaptureResearcher",
    model=GEMINI_MODEL,
    instruction="""你是一名专门研究气候解决方案的 AI 研究助理。研究"碳捕获方法"的当前状态。使用提供的 Google 搜索工具。简洁地总结你的主要发现（1-2 句话）。*只*输出摘要。""",
    description="研究碳捕获方法。",
    tools=[google_search],
    # 将结果存储在状态中供合并 Agent 使用
    output_key="carbon_capture_result"
)

## --- 2. 创建 ParallelAgent（并发运行研究员）---
## 此 Agent 协调研究员的并发执行。
## 一旦所有研究员完成并将结果存储在状态中，它就完成。
parallel_research_agent = ParallelAgent(
    name="ParallelWebResearchAgent",
    sub_agents=[researcher_agent_1, researcher_agent_2, researcher_agent_3],
    description="并行运行多个研究 Agent 以收集信息。"
)

## --- 3. 定义合并 Agent（在并行 Agent *之后*运行）---
## 此 Agent 获取并行 Agent 存储在会话状态中的结果
## 并将它们综合成一个带有归属的单一结构化响应。
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="将并行 Agent 的研究发现组合成结构化的、引用的报告，严格基于提供的输入。",
    # 合并不需要工具
    # 这里不需要 output_key，因为其直接响应是序列的最终输出
)

## --- 4. 创建 SequentialAgent（协调整体流程）---
## 这是将运行的主 Agent。它首先执行 ParallelAgent
## 以填充状态，然后执行 MergerAgent 以产生最终输出。
sequential_pipeline_agent = SequentialAgent(
    name="ResearchAndSynthesisPipeline",
    # 首先运行并行研究，然后合并
    sub_agents=[parallel_research_agent, merger_agent],
    description="协调并行研究并综合结果。"
)

root_agent = sequential_pipeline_agent

此代码定义了一个用于研究和综合可持续技术进展信息的多 Agent 系统。系统设置了三个 LlmAgent 实例作为专门的研究员：ResearcherAgent_1 专注于可再生能源，ResearcherAgent_2 研究电动汽车技术，ResearcherAgent_3 调查碳捕获方法。每个研究员 Agent 配置使用 GEMINI_MODEL 和 google_search 工具，被指示用 1-2 句话简洁总结发现，并通过 output_key 将摘要存储在会话状态中。

随后创建名为 ParallelWebResearchAgent 的 ParallelAgent 来并发运行这三个研究员 Agent，实现并行研究以节省时间。当所有子 Agent（研究员）完成并填充状态后，ParallelAgent 结束执行。

接下来定义 MergerAgent（同样是 LlmAgent）来综合研究结果。该 Agent 以并行研究员存储在会话状态中的摘要作为输入，其指令强调输出必须严格基于提供的输入摘要，禁止添加外部知识。MergerAgent 旨在将组合的发现结构化为带有各主题标题和简短总体结论的报告。

最后创建名为 ResearchAndSynthesisPipeline 的 SequentialAgent 来协调整个工作流。作为主控制器，该主 Agent 首先执行 ParallelAgent 进行研究，待其完成后执行 MergerAgent 综合收集的信息。sequential_pipeline_agent 被设置为 root_agent，作为运行此多 Agent 系统的入口点。整个过程旨在高效地从多个来源并行收集信息，并将其整合为单一结构化报告。

概览

是什么： 许多 Agent 工作流包含多个必须完成才能达成最终目标的子任务。纯顺序执行（每个任务等待前一个完成）通常低效且缓慢。当任务依赖外部 I/O 操作（如调用不同 API 或查询多个数据库）时，这种延迟成为主要瓶颈。若无并发执行机制，总处理时间等于所有单个任务持续时间之和，严重制约系统整体性能和响应能力。

为什么： 并行化模式通过启用独立任务的同时执行提供标准化解决方案。该模式通过识别工作流中不依赖彼此即时输出的组件（如工具使用或 LLM 调用）来实现。像 LangChain 和 Google ADK 这样的 Agent 框架提供内置构造来定义和管理这些并发操作。例如，主进程可调用多个并行运行的子任务，等待所有子任务完成后再继续下一步。通过同时而非顺序执行这些独立任务，该模式显著减少总执行时间。

经验法则： 当工作流包含多个可同时运行的独立操作时使用此模式，例如从多个 API 获取数据、处理不同数据块或生成多个内容片段供后续综合。

视觉摘要

图 2：并行化设计模式

关键要点

以下是本章的核心要点：

  • 并行化是一种通过并发执行独立任务来提高效率的模式
  • 在涉及等待外部资源（如 API 调用）的任务中特别有效
  • 采用并发或并行架构会引入显著复杂性和成本，影响设计、调试和系统日志等关键开发环节
  • 像 LangChain 和 Google ADK 这样的框架提供定义和管理并行执行的内置支持
  • 在 LangChain 表达式语言（LCEL）中，RunnableParallel 是并行运行多个可运行对象的关键构造
  • Google ADK 可通过 LLM 驱动的委托实现并行执行，协调器 Agent 的 LLM 识别独立子任务并触发专门子 Agent 的并发处理
  • 并行化有助于减少整体延迟，使 Agent 系统在处理复杂任务时更具响应性

结论

并行化模式是通过并发执行独立子任务来优化计算工作流的方法。该模式有效减少整体延迟，在涉及多个模型推理或对外部服务调用的复杂操作中尤为显著。

不同框架为此模式提供了不同的实现机制。在 LangChain 中，通过 RunnableParallel 等构造显式定义并同时执行多个处理链。而 Google Agent Developer Kit (ADK) 等框架则通过多 Agent 委托实现并行化，由主协调器模型将不同子任务分配给可并发操作的专门 Agent。

通过将并行处理与顺序（链式）和条件（路由）控制流相结合，可以构建能够高效管理各类复杂任务的复杂、高性能计算系统。

参考文献

以下是有关并行化模式和相关概念的一些进一步阅读资源：

  1. LangChain Expression Language (LCEL) Documentation (Parallelism): https://python.langchain.com/docs/concepts/lcel/
  2. Google Agent Developer Kit (ADK) Documentation (Multi-Agent Systems): https://google.github.io/adk-docs/agents/multi-agents/
  3. Python asyncio Documentation: https://docs.python.org/3/library/asyncio.html

展开全文 >>

Agentic Design Patterns - Chapter 2_ Routing

第 2 章：路由

路由模式概述

虽然通过提示词链进行顺序处理是执行确定性、线性工作流的基础技术，但其适用性在需要自适应响应的场景中受到限制。现实世界的 Agent 系统必须经常根据偶然因素在多个潜在行动之间进行仲裁，例如环境状态、用户输入或前一操作的结果。这种动态决策能力，控制流向不同的专门函数、工具或子流程，是通过一种称为路由的机制实现的。

路由将条件逻辑引入 Agent 的操作框架，使其能够从固定执行路径转变为这样一种模型：Agent 动态评估特定标准以从一组可能的后续行动中进行选择。这允许更灵活和上下文感知的系统行为。

例如，一个设计用于客户查询的 Agent，当配备路由功能时，可以首先对传入查询进行分类以确定用户的意图。基于此分类，它可以将查询定向到专门的 Agent 进行直接问答、用于帐户信息的数据库检索工具，或用于复杂问题的升级程序，而不是默认使用单一的预定响应路径。因此，使用路由的更复杂 Agent 可以：

  1. 分析用户的查询。
  2. 基于其意图路由查询：
    • 如果意图是”检查订单状态”，路由到与订单数据库交互的子 Agent 或工具链。
    • 如果意图是”产品信息”，路由到搜索产品目录的子 Agent 或链。
    • 如果意图是”技术支持”，路由到访问故障排除指南或升级到人工的不同链。
    • 如果意图不清楚，路由到澄清子 Agent 或提示词链。

路由模式的核心组件是执行评估并指导流程的机制。这种机制可以通过几种方式实现：

  • 基于 LLM 的路由： 语言模型本身可以被提示分析输入并输出指示下一步或目的地的特定标识符或指令。例如，提示词可能要求 LLM “分析以下用户查询并仅输出类别：’订单状态’、’产品信息’、’技术支持’或’其他’。” Agent 系统然后读取此输出并相应地指导工作流。
  • 基于嵌入的路由： 输入查询可以转换为向量嵌入（参见 RAG，第 14 章）。然后将此嵌入与代表不同路由或能力的嵌入进行比较。查询被路由到嵌入最相似的路由。这对于语义路由很有用，其中决策基于输入的含义而不仅仅是关键词。
  • 基于规则的路由： 这涉及使用基于关键词、模式或从输入中提取的结构化数据的预定义规则或逻辑（例如，if-else 语句、switch case）。这可以比基于 LLM 的路由更快、更确定，但在处理细微或新颖输入方面的灵活性较差。
  • 基于机器学习模型的路由： 它采用判别模型，例如分类器，该模型已经在小型标记数据语料库上专门训练以执行路由任务。虽然它与基于嵌入的方法在概念上有相似之处，但其关键特征是监督微调过程，该过程调整模型的参数以创建专门的路由功能。这种技术与基于 LLM 的路由不同，因为决策组件不是在推理时执行提示词的生成模型。相反，路由逻辑被编码在微调模型的学习权重中。虽然 LLM 可能在预处理步骤中用于生成合成数据以增强训练集，但它们不参与实时路由决策本身。

路由机制可以在 Agent 操作周期内的多个节点实现。它们可以在开始时应用以对主要任务进行分类，在处理链内的中间点应用以确定后续操作，或在子程序期间应用以从给定集合中选择最合适的工具。

诸如 LangChain、LangGraph 和 Google 的 Agent Developer Kit (ADK) 等计算框架提供了用于定义和管理这种条件逻辑的显式构造。凭借其基于状态的图架构，LangGraph 特别适合复杂的路由场景，其中决策取决于整个系统的累积状态。类似地，Google 的 ADK 提供了用于构建 Agent 能力和交互模型的基础组件，这些组件作为实现路由逻辑的基础。在这些框架提供的执行环境中，开发人员定义可能的操作路径以及决定计算图中节点之间转换的函数或基于模型的评估。

路由的实现使系统能够超越确定性顺序处理。它促进了更自适应的执行流的开发，可以动态且适当地响应更广泛的输入和状态变化。

实际应用与用例

路由模式是自适应 Agent 系统设计中的关键控制机制，使它们能够动态改变其执行路径以响应可变输入和内部状态。其效用通过提供必要的条件逻辑层跨越多个领域。

在人机交互中，例如虚拟助手或 AI 驱动的导师，路由用于解释用户意图。对自然语言查询的初始分析确定最合适的后续操作，无论是调用特定信息检索工具、升级到人工操作员，还是根据用户表现选择课程中的下一个模块。这使系统能够超越线性对话流并进行上下文响应。

在自动化数据和文档处理管道中，路由作为分类和分发功能。传入的数据，如电子邮件、支持票据或 API 有效载荷，根据内容、元数据或格式进行分析。然后系统将每个项目定向到相应的工作流，例如销售线索摄入流程、JSON 或 CSV 格式的特定数据转换函数，或紧急问题升级路径。

在涉及多个专门工具或 Agent 的复杂系统中，路由充当高级调度器。由用于搜索、总结和分析信息的不同 Agent 组成的研究系统将使用路由器根据当前目标将任务分配给最合适的 Agent。类似地，AI 编码助手使用路由来识别编程语言和用户意图——调试、解释或翻译——然后将代码片段传递给正确的专门工具。

最终，路由提供了创建功能多样化和上下文感知系统所必需的逻辑仲裁能力。它将 Agent 从预定义序列的静态执行器转变为可以在变化条件下就完成任务的最有效方法做出决策的动态系统。

实操代码示例（LangChain）

在代码中实现路由涉及定义可能的路径和决定采取哪条路径的逻辑。像 LangChain 和 LangGraph 这样的框架为此提供了特定的组件和结构。LangGraph 基于状态的图结构对于可视化和实现路由逻辑特别直观。

此代码演示了使用 LangChain 和 Google 的生成式 AI 的简单类 Agent 系统。它设置了一个”协调器”，根据请求的意图（预订、信息或不清楚）将用户请求路由到不同的模拟”子 Agent”处理程序。系统使用语言模型对请求进行分类，然后将其委托给适当的处理函数，模拟多 Agent 架构中常见的基本委托模式。

首先，确保您已安装必要的库：

1
pip install langchain langgraph google-cloud-aiplatform langchain-google-genai google-adk deprecated pydantic

您还需要使用您选择的语言模型的 API 密钥设置环境（例如，OpenAI、Google Gemini、Anthropic）。

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
## Copyright (c) 2025 Marco Fago
## https://www.linkedin.com/in/marco-fago/
#
## 此代码根据 MIT 许可证授权。
## 请参阅仓库中的 LICENSE 文件以获取完整许可文本。
from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableBranch

## --- 配置 ---
## 确保设置了您的 API 密钥环境变量（例如，GOOGLE_API_KEY）
try:
    llm = ChatGoogleGenerativeAI(model="gemini-2.5-flash", temperature=0)
    print(f"语言模型已初始化: {llm.model}")
except Exception as e:
    print(f"初始化语言模型时出错: {e}")
    llm = None

## --- 定义模拟子 Agent 处理程序（相当于 ADK 的 sub_agents）---
def booking_handler(request: str) -> str:
    """模拟预订 Agent 处理请求。"""
    print("\n--- 委托给预订处理程序 ---")
    return f"预订处理程序处理了请求：'{request}'。结果：模拟预订操作。"

def info_handler(request: str) -> str:
    """模拟信息 Agent 处理请求。"""
    print("\n--- 委托给信息处理程序 ---")
    return f"信息处理程序处理了请求：'{request}'。结果：模拟信息检索。"

def unclear_handler(request: str) -> str:
    """处理无法委托的请求。"""
    print("\n--- 处理不清楚的请求 ---")
    return f"协调器无法委托请求：'{request}'。请澄清。"

## --- 定义协调器路由链（相当于 ADK 协调器的指令）---
## 此链决定应委托给哪个处理程序。
coordinator_router_prompt = ChatPromptTemplate.from_messages([
    ("system", """分析用户的请求并确定哪个专家处理程序应处理它。
     - 如果请求与预订航班或酒店相关，
        输出 'booker'。
     - 对于所有其他一般信息问题，输出 'info'。
     - 如果请求不清楚或不适合任一类别，
        输出 'unclear'。
     只输出一个词：'booker'、'info' 或 'unclear'。"""),
    ("user", "{request}")
])

if llm:
    coordinator_router_chain = coordinator_router_prompt | llm | StrOutputParser()

## --- 定义委托逻辑（相当于 ADK 的基于 sub_agents 的自动流）---
## 使用 RunnableBranch 根据路由链的输出进行路由。
## 为 RunnableBranch 定义分支
branches = {
    "booker": RunnablePassthrough.assign(output=lambda x: booking_handler(x['request']['request'])),
    "info": RunnablePassthrough.assign(output=lambda x: info_handler(x['request']['request'])),
    "unclear": RunnablePassthrough.assign(output=lambda x: unclear_handler(x['request']['request'])),
}

## 创建 RunnableBranch。它接受路由链的输出
## 并将原始输入（'request'）路由到相应的处理程序。
delegation_branch = RunnableBranch(
    (lambda x: x['decision'].strip() == 'booker', branches["booker"]), # 添加了 .strip()
    (lambda x: x['decision'].strip() == 'info', branches["info"]),     # 添加了 .strip()
    branches["unclear"] # 'unclear' 或任何其他输出的默认分支
)

## 将路由链和委托分支组合成单个可运行对象
## 路由链的输出（'decision'）与原始输入（'request'）一起传递
## 到 delegation_branch。
coordinator_agent = {
    "decision": coordinator_router_chain,
    "request": RunnablePassthrough()
} | delegation_branch | (lambda x: x['output']) # 提取最终输出

## --- 示例用法 ---
def main():
    if not llm:
        print("\n由于 LLM 初始化失败，跳过执行。")
        return
    
    print("--- 运行预订请求 ---")
    request_a = "给我预订去伦敦的航班。"
    result_a = coordinator_agent.invoke({"request": request_a})
    print(f"最终结果 A: {result_a}")
    
    print("\n--- 运行信息请求 ---")
    request_b = "意大利的首都是什么？"
    result_b = coordinator_agent.invoke({"request": request_b})
    print(f"最终结果 B: {result_b}")
    
    print("\n--- 运行不清楚的请求 ---")
    request_c = "告诉我关于量子物理学的事。"
    result_c = coordinator_agent.invoke({"request": request_c})
    print(f"最终结果 C: {result_c}")

if __name__ == "__main__":
    main()

如上所述，这段 Python 代码使用 LangChain 库和 Google 的生成式 AI 模型（特别是 gemini-2.5-flash）构建了一个简单的类 Agent 系统。详细来说，它定义了三个模拟子 Agent 处理程序：booking_handler、info_handler 和 unclear_handler，每个都设计用于处理特定类型的请求。

核心组件是 coordinator_router_chain，它利用 ChatPromptTemplate 指示语言模型将传入的用户请求分类为三个类别之一：’booker’、’info’ 或 ‘unclear’。然后路由链的输出由 RunnableBranch 使用，将原始请求委托给相应的处理函数。RunnableBranch 检查来自语言模型的决策，并将请求数据定向到 booking_handler、info_handler 或 unclear_handler。coordinator_agent 结合了这些组件，首先路由请求以做出决策，然后将请求传递给选定的处理程序。最终输出从处理程序的响应中提取。

main 函数通过三个示例请求演示了系统的用法，展示了不同的输入如何被路由并由模拟 Agent 处理。包含了语言模型初始化的错误处理以确保健壮性。代码结构模仿了基本的多 Agent 框架，其中中央协调器根据意图将任务委托给专门的 Agent。

实操代码示例（Google ADK）

Agent Development Kit (ADK) 是一个用于工程化 Agent 系统的框架，为定义 Agent 的能力和行为提供了结构化环境。与基于显式计算图的架构相比，ADK 范式中的路由通常通过定义一组离散的”工具”来实现，这些工具代表 Agent 的功能。响应用户查询选择适当工具由框架的内部逻辑管理，该逻辑利用底层模型将用户意图与正确的功能处理程序匹配。

此 Python 代码演示了使用 Google ADK 库的 Agent Development Kit (ADK) 应用程序示例。它设置了一个”协调器” Agent，根据定义的指令将用户请求路由到专门的子 Agent（”Booker”用于预订，”Info”用于一般信息）。然后子 Agent 使用特定工具模拟处理请求，展示了 Agent 系统中的基本委托模式。

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
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
## Copyright (c) 2025 Marco Fago
#
## 此代码根据 MIT 许可证授权。
## 请参阅仓库中的 LICENSE 文件以获取完整许可文本。
import uuid
from typing import Dict, Any, Optional
from google.adk.agents import Agent
from google.adk.runners import InMemoryRunner
from google.adk.tools import FunctionTool
from google.genai import types
from google.adk.events import Event

## --- 定义工具函数 ---
## 这些函数模拟专家 Agent 的操作。
def booking_handler(request: str) -> str:
    """
    处理航班和酒店的预订请求。
    参数：
        request: 用户的预订请求。
    返回：
        确认预订已处理的消息。
    """
    print("-------------------------- 调用预订处理程序 ----------------------------")
    return f"已模拟对 '{request}' 的预订操作。"

def info_handler(request: str) -> str:
    """
    处理一般信息请求。
    参数：
        request: 用户的问题。
    返回：
        表示信息请求已处理的消息。
    """
    print("-------------------------- 调用信息处理程序 ----------------------------")
    return f"对 '{request}' 的信息请求。结果：模拟信息检索。"

def unclear_handler(request: str) -> str:
    """处理无法委托的请求。"""
    return f"协调器无法委托请求：'{request}'。请澄清。"

## --- 从函数创建工具 ---
booking_tool = FunctionTool(booking_handler)
info_tool = FunctionTool(info_handler)

## 定义配备各自工具的专门子 Agent
booking_agent = Agent(
    name="Booker",
    model="gemini-2.0-flash",
    description="一个专门的 Agent，通过调用预订工具处理所有航班和酒店预订请求。",
    tools=[booking_tool]
)

info_agent = Agent(
    name="Info",
    model="gemini-2.0-flash",
    description="一个专门的 Agent，通过调用信息工具提供一般信息并回答用户问题。",
    tools=[info_tool]
)

## 定义具有明确委托指令的父 Agent
coordinator = Agent(
    name="Coordinator",
    model="gemini-2.0-flash",
    instruction=(
        "你是主协调器。你唯一的任务是分析传入的用户请求"
        "并将它们委托给适当的专家 Agent。不要尝试直接回答用户。\n"
        "- 对于任何与预订航班或酒店相关的请求，委托给 'Booker' Agent。\n"
        "- 对于所有其他一般信息问题，委托给 'Info' Agent。"
    ),
    description="一个将用户请求路由到正确专家 Agent 的协调器。",
    # sub_agents 的存在默认启用 LLM 驱动的委托（自动流）。
    sub_agents=[booking_agent, info_agent]
)

## --- 执行逻辑 ---
async def run_coordinator(runner: InMemoryRunner, request: str):
    """使用给定请求运行协调器 Agent 并委托。"""
    print(f"\n--- 使用请求运行协调器: '{request}' ---")
    final_result = ""
    try:
        user_id = "user_123"
        session_id = str(uuid.uuid4())
        await runner.session_service.create_session(
            app_name=runner.app_name, user_id=user_id, session_id=session_id
        )
        
        for event in runner.run(
            user_id=user_id,
            session_id=session_id,
            new_message=types.Content(
                role='user',
                parts=[types.Part(text=request)]
            ),
        ):
            if event.is_final_response() and event.content:
                # 尝试直接从 event.content 获取文本
                # 以避免迭代部分
                if hasattr(event.content, 'text') and event.content.text:
                    final_result = event.content.text
                elif event.content.parts:
                    # 后备：迭代部分并提取文本（可能触发警告）
                    text_parts = [part.text for part in event.content.parts if part.text]
                    final_result = "".join(text_parts)
                # 假设循环应在最终响应后中断
                break
        
        print(f"协调器最终响应: {final_result}")
        return final_result
    except Exception as e:
        print(f"处理您的请求时出错: {e}")
        return f"处理您的请求时出错: {e}"

async def main():
    """运行 ADK 示例的主函数。"""
    print("--- Google ADK 路由示例（ADK 自动流风格）---")
    print("注意：这需要安装并认证 Google ADK。")
    
    runner = InMemoryRunner(coordinator)
    
    # 示例用法
    result_a = await run_coordinator(runner, "给我在巴黎预订一家酒店。")
    print(f"最终输出 A: {result_a}")
    
    result_b = await run_coordinator(runner, "世界上最高的山是什么？")
    print(f"最终输出 B: {result_b}")
    
    result_c = await run_coordinator(runner, "告诉我一个随机事实。") # 应该去 Info
    print(f"最终输出 C: {result_c}")
    
    result_d = await run_coordinator(runner, "查找下个月去东京的航班。") # 应该去 Booker
    print(f"最终输出 D: {result_d}")

if __name__ == "__main__":
    import nest_asyncio
    nest_asyncio.apply()
    await main()

此脚本由一个主协调器 Agent 和两个专门的子 Agent 组成：Booker 和 Info。每个专门的 Agent 都配备了一个 FunctionTool，它包装了一个模拟操作的 Python 函数。booking_handler 函数模拟处理航班和酒店预订，而 info_handler 函数模拟检索一般信息。unclear_handler 作为协调器无法委托的请求的后备包含在内，尽管当前协调器逻辑在主 run_coordinator 函数中没有明确使用它进行委托失败。

协调器 Agent 的主要角色，如其指令中定义的，是分析传入的用户消息并将它们委托给 Booker 或 Info Agent。这种委托由 ADK 的自动流机制自动处理，因为协调器定义了 sub_agents。run_coordinator 函数设置了一个 InMemoryRunner，创建一个用户和会话 ID，然后使用运行器通过协调器 Agent 处理用户的请求。runner.run 方法处理请求并产生事件，代码从 event.content 中提取最终响应文本。

main 函数通过使用不同请求运行协调器来演示系统的用法，展示了它如何将预订请求委托给 Booker，将信息请求委托给 Info Agent。

概览

是什么： Agent 系统必须经常响应各种各样的输入和情况，这些无法由单一的线性流程处理。简单的顺序工作流缺乏基于上下文做出决策的能力。没有为特定任务选择正确工具或子流程的机制，系统仍然是僵化和非自适应的。这种限制使得难以构建能够管理现实世界用户请求的复杂性和可变性的复杂应用程序。

为什么： 路由模式通过将条件逻辑引入 Agent 的操作框架提供了标准化解决方案。它使系统能够首先分析传入查询以确定其意图或性质。基于此分析，Agent 动态地将控制流定向到最合适的专门工具、功能或子 Agent。这个决策可以由各种方法驱动，包括提示 LLM、应用预定义规则或使用基于嵌入的语义相似性。最终，路由将静态的预定执行路径转变为能够选择最佳可能操作的灵活和上下文感知工作流。

经验法则： 当 Agent 必须根据用户输入或当前状态在多个不同的工作流、工具或子 Agent 之间做出决策时，使用路由模式。它对于需要对传入请求进行分类或分类以处理不同类型任务的应用程序至关重要，例如客户支持机器人区分销售查询、技术支持和帐户管理问题。

视觉摘要：


图 1：路由模式，使用 LLM 作为路由器

关键要点

  • 路由使 Agent 能够根据条件动态决定工作流中的下一步。
  • 它允许 Agent 处理各种输入并调整其行为，超越线性执行。
  • 路由逻辑可以使用 LLM、基于规则的系统或嵌入相似性实现。
  • 像 LangGraph 和 Google ADK 这样的框架提供了在 Agent 工作流中定义和管理路由的结构化方式，尽管采用不同的架构方法。

结论

路由模式是构建真正动态和响应式 Agent 系统的关键步骤。通过实现路由，我们超越了简单的线性执行流，使我们的 Agent 能够就如何处理信息、响应用户输入以及利用可用工具或子 Agent 做出智能决策。

我们已经看到路由如何应用于各个领域，从客户服务聊天机器人到复杂的数据处理管道。分析输入并有条件地指导工作流的能力是创建能够处理现实世界任务固有可变性的 Agent 的基础。

使用 LangChain 和 Google ADK 的代码示例展示了实现路由的两种不同但有效的方法。LangGraph 基于图的结构提供了定义状态和转换的可视化和显式方式，使其成为具有复杂路由逻辑的复杂多步工作流的理想选择。另一方面，Google ADK 通常专注于定义不同的能力（工具）并依赖框架将用户请求路由到适当工具处理程序的能力，这对于具有明确定义的离散操作集的 Agent 可能更简单。

掌握路由模式对于构建能够智能地导航不同场景并根据上下文提供定制响应或操作的 Agent 至关重要。它是创建多功能和健壮 Agent 应用程序的关键组件。

参考文献

  1. LangGraph Documentation: https://www.langchain.com/
  2. Google Agent Developer Kit Documentation: https://google.github.io/adk-docs/

展开全文 >>

Agentic Design Patterns - Chapter 1_ Prompt Chaining

第 1 章：提示词链

提示词链模式概述

提示词链（Prompt Chaining），有时也称为管道模式（Pipeline pattern），是利用大型语言模型（LLM）处理复杂任务时的一种强大范式。提示词链不是期望 LLM 在单一的、整体化的步骤中解决复杂问题，而是采用分而治之的策略。其核心思想是将原始复杂问题分解为一系列更小、更易管理的子问题。每个子问题通过专门设计的提示词单独处理，并且一个提示词的输出会作为输入传递给链中的下一个提示词。

这种顺序处理技术为与 LLM 的交互引入了模块化和清晰性。通过分解复杂任务，更容易理解和调试每个单独的步骤，使整个过程更加健壮和可解释。链中的每一步都可以精心设计和优化，专注于更大问题的特定方面，从而产生更准确和聚焦的输出。

一个步骤的输出作为下一个步骤的输入至关重要。这种信息传递建立了依赖链（因此得名），其中先前操作的上下文和结果指导后续处理。这使得 LLM 能够在其先前工作的基础上构建，完善理解，并逐步接近期望的解决方案。

此外，提示词链不仅仅是分解问题；它还支持外部知识和工具的集成。在每一步，LLM 都可以被指示与外部系统、API 或数据库交互，丰富其超越内部训练数据的知识和能力。这种能力极大地扩展了 LLM 的潜力，使它们不仅仅作为独立模型运行，而是作为更广泛智能系统的组成部分。

提示词链的重要性超越了简单的问题解决。它是构建复杂 AI Agent 的基础技术。这些 Agent 可以利用提示词链在动态环境中自主规划、推理和行动。通过战略性地构建提示词序列，Agent 可以参与需要多步推理、规划和决策的任务。这样的 Agent 工作流可以更紧密地模拟人类思维过程，从而实现与复杂领域和系统更自然有效的交互。

单一提示词的局限性： 对于多方面的任务，为 LLM 使用单一的复杂提示词可能效率低下，导致模型在约束和指令方面遇到困难，可能导致指令忽略（提示词的某些部分被忽视）、上下文漂移（模型失去对初始上下文的跟踪）、错误传播（早期错误被放大）、需要更长上下文窗口（模型获得的信息不足无法响应）以及幻觉（认知负荷增加导致错误信息的可能性）。例如，一个要求分析市场研究报告、总结发现、识别带数据点的趋势并起草电子邮件的查询可能会失败，因为模型可能总结得很好，但未能提取数据或正确起草电子邮件。

通过顺序分解增强可靠性： 提示词链通过将复杂任务分解为聚焦的顺序工作流来解决这些挑战，显著提高了可靠性和控制力。基于上面的例子，管道或链式方法可以描述如下：

  1. 初始提示词（总结）：”总结以下市场研究报告的主要发现：[文本]。” 模型的唯一焦点是总结，提高了这一初始步骤的准确性。
  2. 第二个提示词（趋势识别）：”使用摘要，识别前三个新兴趋势并提取支持每个趋势的具体数据点：[步骤 1 的输出]。” 此提示词现在更受约束，并直接建立在经过验证的输出之上。
  3. 第三个提示词（电子邮件撰写）：”向营销团队起草一封简明的电子邮件，概述以下趋势及其支持数据：[步骤 2 的输出]。”

这种分解允许对流程进行更精细的控制。每一步都更简单且更清晰，这减少了模型的认知负荷，并导致更准确和可靠的最终输出。这种模块化类似于计算管道，其中每个函数在执行特定操作后将结果传递给下一个。为了确保每个特定任务的准确响应，可以在每个阶段为模型分配不同的角色。例如，在给定的场景中，初始提示词可以被指定为”市场分析师”，后续提示词为”贸易分析师”，第三个提示词为”专家文档撰写者”，依此类推。

结构化输出的作用： 提示词链的可靠性高度依赖于步骤之间传递的数据完整性。如果一个提示词的输出不明确或格式不佳，后续提示词可能由于错误的输入而失败。为了缓解这一问题，指定结构化输出格式（如 JSON 或 XML）至关重要。

例如，趋势识别步骤的输出可以格式化为 JSON 对象：

1
2
3
4
5
6
7
8
9
10
11
12
{
  "trends": [
    {
      "trend_name": "AI-Powered Personalization",
      "supporting_data": "73% of consumers prefer to do business with brands that use personal information to make their shopping experiences more relevant."
    },
    {
      "trend_name": "Sustainable and Ethical Brands",
      "supporting_data": "Sales of products with ESG-related claims grew 28% over the last five years, compared to 20% for products without."
    }
  ]
}

这种结构化格式确保数据是机器可读的，可以精确解析并插入到下一个提示词中，而不会产生歧义。这种做法最大限度地减少了解释自然语言可能产生的错误，是构建健壮的多步骤 LLM 系统的关键组成部分。

实际应用与用例

提示词链是一种多用途模式，在构建 Agent 系统时适用于广泛的场景。其核心效用在于将复杂问题分解为顺序的、可管理的步骤。以下是几个实际应用和用例：

1. 信息处理工作流： 许多任务涉及通过多次转换处理原始信息。例如，总结文档、提取关键实体，然后使用这些实体查询数据库或生成报告。提示词链可能如下所示：

  • 提示词 1：从给定的 URL 或文档中提取文本内容。
  • 提示词 2：总结清理后的文本。
  • 提示词 3：从摘要或原始文本中提取特定实体（例如，姓名、日期、位置）。
  • 提示词 4：使用实体搜索内部知识库。
  • 提示词 5：生成包含摘要、实体和搜索结果的最终报告。

这种方法应用于自动化内容分析、AI 驱动的研究助手开发和复杂报告生成等领域。

2. 复杂查询回答： 回答需要多步推理或信息检索的复杂问题是一个主要用例。例如，”1929 年股市崩盘的主要原因是什么，政府政策如何应对？”

  • 提示词 1：识别用户查询中的核心子问题（崩盘原因、政府响应）。
  • 提示词 2：专门研究或检索有关 1929 年崩盘原因的信息。
  • 提示词 3：专门研究或检索有关政府对 1929 年股市崩盘的政策响应的信息。
  • 提示词 4：将步骤 2 和 3 的信息综合成对原始查询的连贯答案。

这种顺序处理方法是开发能够进行多步推理和信息综合的 AI 系统的核心。当查询无法从单个数据点回答，而是需要一系列逻辑步骤或来自不同来源的信息集成时，需要这样的系统。

例如，设计用于生成关于特定主题的综合报告的自动化研究 Agent 执行混合计算工作流。最初，系统检索大量相关文章。从每篇文章中提取关键信息的后续任务可以为每个来源并发执行。此阶段非常适合并行处理，其中独立的子任务同时运行以最大化效率。

然而，一旦完成单个提取，过程就变得本质上是顺序的。系统必须首先整理提取的数据，然后将其综合成连贯的草稿，最后审查和完善此草稿以生成最终报告。这些后期阶段中的每一个在逻辑上都依赖于前一个阶段的成功完成。这就是应用提示词链的地方：整理的数据作为综合提示词的输入，生成的综合文本成为最终审查提示词的输入。因此，复杂操作经常将独立数据收集的并行处理与综合和完善的依赖步骤的提示词链结合起来。

3. 数据提取和转换： 将非结构化文本转换为结构化格式通常通过迭代过程实现，需要顺序修改以提高输出的准确性和完整性。

  • 提示词 1：尝试从发票文档中提取特定字段（例如，姓名、地址、金额）。
  • 处理：检查是否提取了所有必需字段以及它们是否满足格式要求。
  • 提示词 2（条件性）：如果字段缺失或格式错误，制作新提示词要求模型专门查找缺失/格式错误的信息，可能提供失败尝试的上下文。
  • 处理：再次验证结果。如有必要重复。
  • 输出：提供提取的、验证的结构化数据。

这种顺序处理方法特别适用于从表单、发票或电子邮件等非结构化来源进行数据提取和分析。例如，解决复杂的光学字符识别（OCR）问题，如处理 PDF 表单，通过分解的多步方法更有效地处理。

最初，使用大型语言模型从文档图像执行主要文本提取。随后，模型处理原始输出以规范化数据，在这一步中，它可能将数字文本（如”一千零五十”）转换为其数字等价物 1050。LLM 的一个重大挑战是执行精确的数学计算。因此，在随后的步骤中，系统可以将任何所需的算术运算委托给外部计算器工具。LLM 识别必要的计算，将规范化的数字馈送到工具，然后合并精确的结果。这种文本提取、数据规范化和外部工具使用的链式序列实现了最终的准确结果，这通常很难从单个 LLM 查询中可靠地获得。

4. 内容生成工作流： 复杂内容的创作是一个程序化任务，通常分解为不同的阶段，包括初始构思、结构大纲、起草和后续修订。

  • 提示词 1：根据用户的一般兴趣生成 5 个主题想法。
  • 处理：允许用户选择一个想法或自动选择最佳想法。
  • 提示词 2：基于选定的主题，生成详细的大纲。
  • 提示词 3：根据大纲中的第一点编写草稿部分。
  • 提示词 4：根据大纲中的第二点编写草稿部分，提供前一部分作为上下文。对所有大纲点继续这样做。
  • 提示词 5：审查和完善完整草稿的连贯性、语气和语法。

这种方法用于一系列自然语言生成任务，包括创意叙事、技术文档和其他形式的结构化文本内容的自动创作。

5. 具有状态的对话 Agent： 尽管全面的状态管理架构采用比顺序链接更复杂的方法，提示词链提供了保持对话连续性的基础机制。这种技术通过将每个对话轮次构建为新提示词来维护上下文，该提示词系统地合并来自对话序列中先前交互的信息或提取的实体。

  • 提示词 1：处理用户话语 1，识别意图和关键实体。
  • 处理：使用意图和实体更新对话状态。
  • 提示词 2：基于当前状态，生成响应和/或识别下一个所需的信息片段。
  • 对于后续轮次重复，每个新的用户话语启动一个利用累积对话历史（状态）的链。

这一原则是开发对话 Agent 的基础，使它们能够在扩展的、多轮对话中保持上下文和连贯性。通过保留对话历史，系统可以理解并适当响应依赖于先前交换信息的用户输入。

6. 代码生成和完善： 功能代码的生成通常是一个多阶段过程，需要将问题分解为逐步执行的离散逻辑操作序列。

  • 提示词 1：理解用户对代码函数的请求。生成伪代码或大纲。
  • 提示词 2：基于大纲编写初始代码草稿。
  • 提示词 3：识别代码中的潜在错误或改进领域（可能使用静态分析工具或另一个 LLM 调用）。
  • 提示词 4：基于识别的问题重写或完善代码。
  • 提示词 5：添加文档或测试用例。

在 AI 辅助软件开发等应用中，提示词链的效用源于其将复杂编码任务分解为一系列可管理的子问题的能力。这种模块化结构降低了每一步大型语言模型的操作复杂性。至关重要的是，这种方法还允许在模型调用之间插入确定性逻辑，实现中间数据处理、输出验证和工作流中的条件分支。通过这种方法，一个可能导致不可靠或不完整结果的单一多面请求被转换为由底层执行框架管理的结构化操作序列。

7. 多模态和多步推理： 分析具有不同模态的数据集需要将问题分解为更小的、基于提示词的任务。例如，解释包含带嵌入文本的图片、突出显示特定文本段的标签以及解释每个标签的表格数据的图像，需要这样的方法。

  • 提示词 1：从用户的图像请求中提取和理解文本。
  • 提示词 2：将提取的图像文本与其相应的标签链接起来。
  • 提示词 3：使用表格解释收集的信息以确定所需的输出。

实操代码示例

实现提示词链的范围从脚本中的直接顺序函数调用到利用专门设计用于管理控制流、状态和组件集成的框架。诸如 LangChain、LangGraph、Crew AI 和 Google Agent Development Kit (ADK) 等框架提供了用于构建和执行这些多步过程的结构化环境，这对于复杂架构特别有利。

出于演示目的，LangChain 和 LangGraph 是合适的选择，因为它们的核心 API 明确设计用于组合操作链和图。LangChain 为线性序列提供基础抽象，而 LangGraph 扩展了这些能力以支持状态化和循环计算，这对于实现更复杂的 Agent 行为是必需的。此示例将专注于基础线性序列。

以下代码实现了一个两步提示词链，作为数据处理管道运行。初始阶段旨在解析非结构化文本并提取特定信息。后续阶段然后接收此提取的输出并将其转换为结构化数据格式。

要复制此过程，必须首先安装所需的库。这可以使用以下命令完成：

1
pip install langchain langchain-community langchain-openai langgraph

请注意，langchain-openai 可以替换为不同模型提供商的适当包。随后，必须为选定的语言模型提供商（如 OpenAI、Google Gemini 或 Anthropic）配置执行环境所需的 API 凭据。

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
import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

## 为了更好的安全性，从 .env 文件加载环境变量
## from dotenv import load_dotenv
## load_dotenv()
## 确保你的 OPENAI_API_KEY 在 .env 文件中设置

## 初始化语言模型（推荐使用 ChatOpenAI）
llm = ChatOpenAI(temperature=0)

## --- 提示词 1：提取信息 ---
prompt_extract = ChatPromptTemplate.from_template(
    "从以下文本中提取技术规格：\n\n{text_input}"
)

## --- 提示词 2：转换为 JSON ---
prompt_transform = ChatPromptTemplate.from_template(
    "将以下规格转换为 JSON 对象，使用 'cpu'、'memory' 和 'storage' 作为键：\n\n{specifications}"
)

## --- 使用 LCEL 构建链 ---
## StrOutputParser() 将 LLM 的消息输出转换为简单字符串。
extraction_chain = prompt_extract | llm | StrOutputParser()

## 完整的链将提取链的输出传递到转换提示词的 'specifications' 变量中。
full_chain = (
    {"specifications": extraction_chain}
    | prompt_transform
    | llm
    | StrOutputParser()
)

## --- 运行链 ---
input_text = "新款笔记本电脑型号配备 3.5 GHz 八核处理器、16GB 内存和 1TB NVMe 固态硬盘。"

## 使用输入文本字典执行链。
final_result = full_chain.invoke({"text_input": input_text})

print("\n--- 最终 JSON 输出 ---")
print(final_result)

这段 Python 代码演示了如何使用 LangChain 库处理文本。它利用两个独立的提示词：一个从输入字符串中提取技术规格，另一个将这些规格格式化为 JSON 对象。ChatOpenAI 模型用于语言模型交互，StrOutputParser 确保输出为可用的字符串格式。LangChain 表达式语言（LCEL）用于优雅地将这些提示词和语言模型链接在一起。第一个链 extraction_chain 提取规格。然后 full_chain 获取提取的输出并将其用作转换提示词的输入。提供了描述笔记本电脑的示例输入文本。full_chain 通过这两个步骤调用此文本进行处理。最后打印最终结果，这是一个包含提取和格式化规格的 JSON 字符串。

上下文工程和提示工程

上下文工程（见图 1）是在 token 生成之前系统地设计、构建和向 AI 模型提供完整信息环境的学科。这种方法论断言，模型输出的质量较少依赖于模型架构本身，而更多依赖于所提供上下文的丰富性。

图 1：上下文工程是为 AI 构建丰富、全面的信息环境的学科，因为此上下文的质量是实现高级 Agent 性能的主要因素。

它代表了从传统提示工程的重大演进，传统提示工程主要关注优化用户即时查询的措辞。上下文工程将此范围扩展到包括多个信息层，例如系统提示词，这是定义 AI 操作参数的基础指令集——例如，*”你是一名技术作家；你的语气必须正式而精确。”* 上下文通过外部数据进一步丰富。这包括检索的文档，其中 AI 主动从知识库中获取信息以告知其响应，例如提取项目的技术规格。它还包含工具输出，这是 AI 使用外部 API 获取实时数据的结果，例如查询日历以确定用户的可用性。这些显式数据与关键的隐式数据结合，如用户身份、交互历史和环境状态。核心原则是，即使是高级模型，在提供有限或构建不良的操作环境视图时，也会表现不佳。

因此，这种实践将任务从仅仅回答问题重新定义为为 Agent 构建全面的操作图景。例如，上下文工程化的 Agent 不仅会响应查询，而且首先会整合用户的日历可用性（工具输出）、与电子邮件收件人的专业关系（隐式数据）以及之前会议的笔记（检索的文档）。这使得模型能够生成高度相关、个性化和实用的输出。”工程”组件涉及创建健壮的管道以在运行时获取和转换此数据，并建立反馈循环以持续改进上下文质量。

要实现这一点，可以使用专门的调优系统来大规模自动化改进过程。例如，像 Google 的 Vertex AI 提示词优化器这样的工具可以通过系统地根据一组样本输入和预定义的评估指标评估响应来提高模型性能。这种方法对于在不同模型之间调整提示词和系统指令而无需大量手动重写非常有效。通过向这样的优化器提供样本提示词、系统指令和模板，它可以以编程方式完善上下文输入，为实现复杂上下文工程所需的反馈循环提供结构化方法。

这种结构化方法是区分基本 AI 工具与更复杂、上下文感知系统的关键。它将上下文本身视为主要组件，对 Agent 知道什么、何时知道以及如何使用该信息给予关键重要性。这种实践确保模型对用户的意图、历史和当前环境有全面的理解。最终，上下文工程是将无状态聊天机器人提升为高度能干、情境感知系统的关键方法论。

概览

是什么： 复杂任务在单个提示词内处理时通常会使 LLM 不堪重负，导致严重的性能问题。模型的认知负荷增加了错误的可能性，如忽略指令、失去上下文和生成错误信息。单体提示词难以有效管理多个约束和顺序推理步骤。这导致不可靠和不准确的输出，因为 LLM 未能解决多方面请求的所有方面。

为什么： 提示词链通过将复杂问题分解为一系列较小的、相互关联的子任务来提供标准化解决方案。链中的每一步使用聚焦的提示词执行特定操作，显著提高可靠性和控制力。一个提示词的输出作为下一个提示词的输入传递，创建逐步构建最终解决方案的逻辑工作流。这种模块化的分而治之策略使过程更易于管理、更易于调试，并允许在步骤之间集成外部工具或结构化数据格式。这种模式是开发能够规划、推理和执行复杂工作流的复杂多步 Agent 系统的基础。

经验法则： 当任务对于单个提示词过于复杂、涉及多个不同的处理阶段、需要在步骤之间与外部工具交互，或者在构建需要执行多步推理并维护状态的 Agent 系统时，使用此模式。

视觉摘要

图 2：提示词链模式：Agent 从用户接收一系列提示词，每个 Agent 的输出作为链中下一个 Agent 的输入。

关键要点

以下是一些关键要点：

  • 提示词链将复杂任务分解为一系列更小的、聚焦的步骤。这有时被称为管道模式。
  • 链中的每一步涉及 LLM 调用或处理逻辑，使用前一步的输出作为输入。
  • 这种模式提高了与语言模型的复杂交互的可靠性和可管理性。
  • 像 LangChain/LangGraph 和 Google ADK 这样的框架提供了强大的工具来定义、管理和执行这些多步序列。

结论

通过将复杂问题分解为一系列更简单、更易于管理的子任务，提示词链为指导大型语言模型提供了一个健壮的框架。这种”分而治之”策略通过一次专注于一个特定操作，显著提高了输出的可靠性和控制力。作为基础模式，它支持开发能够进行多步推理、工具集成和状态管理的复杂 AI Agent。最终，掌握提示词链对于构建能够执行远超单个提示词能力的复杂工作流的健壮、上下文感知系统至关重要。

参考文献

  1. LangChain Documentation on LCEL: https://python.langchain.com/v0.2/docs/core_modules/expression_language/
  2. LangGraph Documentation: https://langchain-ai.github.io/langgraph/
  3. Prompt Engineering Guide - Chaining Prompts: https://www.promptingguide.ai/techniques/chaining
  4. OpenAI API Documentation (General Prompting Concepts): https://platform.openai.com/docs/guides/gpt/prompting
  5. Crew AI Documentation (Tasks and Processes): https://docs.crewai.com/
  6. Google AI for Developers (Prompting Guides): https://cloud.google.com/discover/what-is-prompt-engineering?hl=en
  7. Vertex Prompt Optimizer https://cloud.google.com/vertex-ai/generative-ai/docs/learn/prompts/prompt-optimizer

展开全文 >>

tag:

    缺失模块。
    1、请确保node版本大于6.2
    2、在博客根目录（注意不是yilia根目录）执行以下命令：
    npm i hexo-generator-json-content --save

    3、在根目录_config.yml里添加配置： 

      jsonContent:
    meta: false
    pages: false
    posts:
      title: true
      date: true
      path: true
      text: false
      raw: false
      content: false
      slug: false
      updated: false
      comments: false
      link: false
      permalink: false
      excerpt: false
      categories: false
      tags: true

程序员，关于技术我所知甚少，唯有探索，方得真知。