使用 Google ADK 编写订单异常分流 Agent，并通过 Conductor 编排引擎调用

导读

本篇文章基于 docs/devguide/ai/cookbook/google-adk-order-triage.md 展开，讲解如何在 Google ADK（Agent Development Kit） 中编写一个只读的订单异常分流（order-exception triage）Agent，再通过 Conductor 的 Python SDK 将其编译、部署为可复用的 Conductor Agent，最后在父工作流中像调用任何普通任务一样以 AGENT 任务方式调用它。读完本文，你将掌握"框架 Agent 对象 → 编译成工作流图 → 部署注册 → 启动工具 Worker → 工作流按名调用"的完整链路，以及运行环境、JSON 定义、CLI 命令与生产注意事项。

这是 Conductor AI Cookbook 中"AI Agents"一族的经典配方，与 LangChain 权益调查 Agent、可复用 Conductor Agent 并列为"外部框架编写 + Conductor 执行"的三种路径。

配方目标与整体数据流

该配方的目标非常明确：

用 Google ADK 编写一个非变更（non-mutating）的订单异常分流 Agent，并通过 Conductor 调用它。

所谓"非变更"，指的是该 Agent 只负责推荐处置方案（recommend a disposition），绝不真正执行退款、履约或通知客户等变更操作——实际动作必须由后续的人工审批环节完成。

整体数据流可以用下面的流程描述（源自原文档的 mermaid 图）：

Written with Google ADK
        │
        ▼
Deployed with the Conductor SDK
        │
        ▼
Called like any other agent
        │
        ▼
Triage recommendation（分流推荐结果）

即：编写阶段使用 Google ADK 的 Agent 类；部署阶段由 Conductor Python SDK 将 ADK Agent 对象编译成工作流图并注册到服务器；调用阶段由父工作流中的 AGENT 任务按名称触发，最终产出分流推荐。

前置条件与编写路径（Prerequisites and authoring path）

安装与导入

当前 Python SDK 快速上手文档规定的方式是：

python -m pip install 'conductor-python[adk]'

即通过 adk 可选依赖（extra）安装 Conductor SDK 的 Google ADK 支持。核心导入如下（对应 deploy_local_cookbook_agents.py 中的真实用法）：

from conductor.ai.agents import AgentRuntime
from google.adk.agents import Agent
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

⚠️ 版本提示：原文档明确提醒——当前 Python SDK 快速上手使用的正是 python -m pip install 'conductor-python[adk]'、google.adk.agents.Agent 与 AgentRuntime 这套 API。在升级包或改动框架 Agent 调用方式之前，务必先核对 Python SDK 仓库中 框架 Agent 指南 的最新写法，因为框架 Agent 的 API 仍在演进。

编写 ADK Agent 的最小示例

原文档给出的最小示例，通过 McpToolset + StreamableHTTPConnectionParams 将本地 MCP 测试工具暴露给 ADK Agent 使用：

from conductor.ai.agents import AgentRuntime
from google.adk.agents import Agent
from google.adk.tools.mcp_tool import McpToolset, StreamableHTTPConnectionParams

agent = Agent(
    name="adk_order_exception_triage",
    model="openai/gpt-4o",
    instruction="Use MCP evidence to recommend a disposition; never execute it.",
    tools=[McpToolset(connection_params=StreamableHTTPConnectionParams(url="http://127.0.0.1:3001/mcp"))],
)
with AgentRuntime() as runtime:
    runtime.run(agent, "Order O-42 arrived damaged.")

要点解析：

• name：ADK Agent 名称，部署后将成为 Conductor 侧被引用的能力名（父工作流 AGENT 任务通过该名称解析）；
• model：推理模型标识，本文示例为 openai/gpt-4o；
• instruction：系统指令，此处刻意声明"用 MCP 证据推荐处置方案，绝不执行它"，这与"只读分流"目标一致；
• tools：通过 McpToolset 挂接本地 MCP 服务端点 http://127.0.0.1:3001/mcp（本地 MCP Testkit 演示服务器），使 Agent 具备查证订单异常证据的工具能力；
• AgentRuntime 上下文管理器：runtime.run(agent, input) 一次完成"编译 + 执行"，是开发迭代期的用法。

部署脚本与本地运行

原文档要求把配套脚本 deploy_local_cookbook_agents.py 下载到工作目录。该脚本是一个"本地演示配置"：它把 adk_order_exception_triage 连同另外四个 Agent（guarded-incident-planner、langchain_entitlement_investigator、security_reviewer、reliability_reviewer）一起注册，并通过 MCP_TESTKIT_URL = "http://127.0.0.1:3001/mcp" 让每个 Agent 都能发现完整的 Testkit 工具目录。

脚本中 ADK Agent 的完整定义（源码证据）：

adk_order_exception_triage = AdkAgent(
    name="adk_order_exception_triage",
    model="openai/gpt-4o",
    instruction=(
        "Use the available MCP Testkit tools to investigate an order exception, "
        "then recommend a disposition. Do not perform a refund or fulfillment action."
    ),
    tools=[
        McpToolset(
            connection_params=StreamableHTTPConnectionParams(url=MCP_TESTKIT_URL)
        )
    ],
)

安全声明（源码注释原文）：脚本明确说明这是本地演示配置——每个 Agent 都能发现完整的 Testkit 工具目录。生产环境必须使用受限的 MCP 允许列表（scoped MCP allowlist）和按工具的访问策略（per-tool policy），绝不能仅仅因为演示方便就把整个工具目录暴露给 Agent。

脚本的使用方式为两个命令，先部署一次，然后让工具 Worker 保持运行，再调用父工作流：

python3 deploy_local_cookbook_agents.py deploy
python3 deploy_local_cookbook_agents.py serve

• deploy：AgentRuntime.deploy(*AGENTS) 将所有 Agent 编译并注册为命名、带版本的 Conductor Agent，打印各自的 registered_name；
• serve：runtime.serve(*AGENTS) 阻塞运行，作为长期存活的工具 Worker 进程执行 Agent 的工具调用。生产环境中 serve 应归属独立的长期进程，而 deploy 放在 CI/CD 里。

脚本的 main() 对动作做了白名单校验，只接受 deploy 或 serve，其余输入会报出 Usage: deploy_local_cookbook_agents.py {deploy|serve}。

输入输出契约

• 输入：orderId（订单 ID）与 exception（异常描述，如 "Package damaged in transit."）；
• 输出：一个推荐（recommendation），即处置建议，而非处置动作；
• 安全边界：该 Agent 不得持有退款（refund）、履约（fulfillment）或客户通知（customer-notification）相关的凭据——这是"只读分流"设计在权限层面的落地。

可运行的工作流定义（Runnable definition）

将下面的定义保存为 google-adk-order-triage.json（完整内容来自 assets/google-adk-order-triage.json）：

{
  "name": "google_adk_order_exception_triage",
  "description": "Invokes a deployed Google ADK-authored Conductor Agent for a non-mutating order exception recommendation.",
  "version": 1,
  "schemaVersion": 2,
  "timeoutSeconds": 300,
  "timeoutPolicy": "TIME_OUT_WF",
  "inputParameters": [
    "orderId",
    "exception"
  ],
  "tasks": [
    {
      "name": "triage_order_exception",
      "taskReferenceName": "triage_order_exception",
      "type": "AGENT",
      "inputParameters": {
        "agentType": "conductor",
        "name": "adk_order_exception_triage",
        "prompt": "Order ${workflow.input.orderId}; exception: ${workflow.input.exception}. Recommend a disposition only."
      }
    }
  ],
  "outputParameters": {
    "triage": "${triage_order_exception.output.output}"
  }
}

对关键字段的解读：

• schemaVersion: 2：Conductor 工作流定义的当前 schema 版本；
• timeoutSeconds: 300 / timeoutPolicy: "TIME_OUT_WF"：整个工作流 300 秒超时，超时即终止工作流。同时可留意 Conductor 对已部署 Agent 的运行级护栏（见 conductor-agents.md）：maxDurationSeconds 默认 86400 秒限制完整运行时长，maxPollFailures 默认 30 限制连续瞬时轮询失败次数，二者与任务定义自身的超时相互独立；
• inputParameters：声明工作流入口参数 orderId 与 exception；
• tasks[].type = "AGENT"：Agent 调用任务类型；
• agentType: "conductor"：执行模式选择为"已部署的 Conductor Agent"，而非默认的 a2a（远程 A2A 端点）。重点：agentType 选择的是执行模式，不是编写框架——Google ADK、LangChain 等都是 SDK 编写路径，都不是 agentType 的取值（详见 conductor-agents.md）；
• name: "adk_order_exception_triage"：与上一步 deploy 注册的 Agent 名称一一对应，父工作流靠它解析已部署能力；
• prompt：使用 ${workflow.input.xxx} 表达式把父工作流输入拼装成给 Agent 的提示词，并再次强调"只做推荐"；
• outputParameters.triage：把 AGENT 任务的 output.output（结构化输出）透出为工作流输出 triage。

AGENT 任务在运行完成后会写回 executionId、agentName、state、text，以及完成态的结构化 output；其 state 为归一化的 A2A 生命周期取值（working、input-required、completed、failed、canceled）。

注册并运行（Register and run）

使用 Conductor CLI 依次执行：

conductor workflow create google-adk-order-triage.json
conductor workflow start -w google_adk_order_exception_triage --sync -i '{"orderId":"O-42","exception":"Package damaged in transit."}'

• workflow create：注册（或按版本更新）工作流定义；
• workflow start -w <工作流名> --sync：同步启动并等待执行结果；-i 传入 JSON 格式的工作流输入，这里即 orderId 与 exception。

运行前提：先执行过 deploy 且 serve 进程仍在运行——AGENT 任务按名称调用已部署 Agent，而工具执行依赖存活的 Worker 进程。

生产注意事项（Production notes）

原文档给出的五条生产要点，逐条展开如下：

1. agentType 是 conductor，不是 adk。 无论 Agent 是用哪个框架编写的，部署后都由 Conductor SDK 执行，协议不随编写框架改变；AGENT 任务只关心 agentType 与 name。
2. 使用服务器实际配置好的模型。 示例中的 openai/gpt-4o 依赖你的服务器配好了 OpenAI 凭据；若服务端配置的是 Gemini，则改用 gemini-2.0-flash 等实际可用模型。在本地服务器上，启动前需先导出模型提供方的 API Key。
3. 在部署端收敛工具访问与迭代预算。 工具访问应使用 MCP 允许列表与按工具策略，迭代次数（iteration budget）也要设限——这些限制要落在"Agent 真正运行循环"的部署层，而不是只靠提示词约束。
4. 只推荐、不执行。 该 Agent 永远只产出处置建议，真正的退款/履约等动作必须路由到审批环节（如 人工审批 Agent 或 HITL 工作流）之后才能执行。
5. 按"订单 ID + 异常事件 ID"对账。 由于重试与取消可能产生重复执行，建议以 orderId 加异常事件 ID（exception event ID）作为幂等对账键，再结合 AGENT 任务的 executionId 排查重复。

更进一步：从框架对象到工作流步骤

本文配方本质上是"框架 Agent 编译为可复用工作流步骤"通用路径的 ADK 实例。该路径共四步（见 框架 Agent 参考）：

1. run（迭代）：把框架 Agent 对象交给 AgentRuntime.run()，SDK 编译并执行，首次运行即在 UI 中可见持久化执行；
2. plan（CI 检查）：runtime.plan(agent) 在部署前检查编译出的工作流图；
3. deploy（发布）：runtime.deploy(agent) 在服务器注册为命名、带版本的 Conductor Agent，调用方无需再引入你的框架依赖；
4. serve（运维）：runtime.serve(agent) 启动工具 Worker 并阻塞，保持已部署 Agent 可用。

部署完成后，任何父工作流都能用 AGENT 任务按名称调用它——这正是 可复用 Conductor Agent 配方中 invoke_reusable_conductor_agent 的工作方式。本配方则展示了 google_adk_order_exception_triage 这一实例：ADK 负责 Agent 的思考与工具编排，Conductor 提供其缺乏的持久化执行、按任务重试、人工审批边界与可观测性。

最后提醒两点运行前提：服务端需启用 AI 集成（配置项 conductor.integrations.ai.enabled=true），否则部署与 agentType: "conductor" 执行模式不可用；同时在推进到生产前，应结合 Agent 护栏 与 Agent 评估 对该分流 Agent 做策略验证与行为评测。