企业官网建设流程全解析

Cross-Border Data Router：基于 A2A AgentCard 元数据与 OpenEAGO 模式的多智能体跨境数据合规路由实战

【免费下载链接】adk-samplesA collection of sample agents built with Agent Development Kit (ADK)项目地址: https://gitcode.com/GitHub_Trending/ad/adk-samples

本篇文章围绕 adk-samples 仓库中contrib/python/cross-border-data-router这一多智能体示例（recipe），系统讲解如何在 Agent Development Kit（ADK）中实现企业级跨境数据合规路由：由根编排智能体接收数据处理请求，依据每个候选智能体在 A2AAgentCard上声明的管辖权、数据驻留与合规元数据，通过"先硬过滤、后打分"的策略引擎决定"谁有资格处理这份数据"，并在无合格者时直接拒绝而非降级到不合规区域。读完本文，你将掌握用capabilities.extensions扩展 A2A 协议表达地域元数据的方法、三阶段路由算法及其源码级实现、CLI/FastAPI 两种运行方式与测试组织方式，可直接复用到自己的多区域数据处理场景。

一、解决什么问题：合规判断先于任务委派

在真实企业中，"把一份数据交给哪个区域的智能体处理"往往不是技术路由问题，而是合规问题：欧盟公民的 PII 必须留在 EU/EEA 驻留区域并受 GDPR 约束，美国客户的财务记录要遵守 CCPA，而合同中可能还明文禁止数据流向某个司法管辖区。如果让编排智能体凭"感觉"选择区域处理器，一旦选错就是合规事故。

本 recipe 演示的正是这一场景的工程化解法（详见 README）：

根编排智能体收到一个数据处理请求（如数据分类 "PII" + 数据来源区域），把它交给一个显式的策略引擎去评估，只有策略引擎批准的智能体才会被委派执行；若没有任何智能体符合条件，则直接拒绝，绝不静默降级。

核心设计是"先合规评估、后任务执行"两阶段分离。策略引擎读取的不是查询文本，而是每个候选智能体在 A2AAgentCard中自我声明的元数据（app/policy/engine.py中的evaluate_routing_policy），这与按查询文本抽取路由目标的做法有本质区别——后者读的是"被检查对象"的信息，前者读的是"处理器自身"的合规声明，更接近真实多智能体注册中心的决策方式。

本 recipe 还特意澄清了术语边界：这里的 "policy" 指用于路由决策的声明式数据驻留/管辖权规则，不同于仓库中core/python/long-horizon-harness的 tool-call guardrails（后者门控单个智能体自身的行为，而非在智能体之间做选择）。

二、整体架构：编排者 + 区域处理器 + 策略引擎

从源码结构看，recipe 由三部分构成：

根智能体root_agent的定义（app/agent.py）清晰展示了"决策工具 + 执行子智能体"的分工：

def create_agent() -> Agent: return Agent( name="root_agent", model=Gemini( model=os.getenv("MODEL_NAME"), retry_options=types.HttpRetryOptions(attempts=3), ), description=( "Routes>def process_record(record_summary: str) -> str: return ( "Processed in the EU-WEST region (Frankfurt data center) under " f"GDPR-compliant controls. Record: {record_summary}" )

三、用 A2A AgentCard 表达"管辖权"与"数据驻留"

A2A（Agent2Agent）协议的AgentCard原生只描述能力（capabilities），没有内置的管辖权（jurisdiction）或数据驻留（data residency）概念。OpenEAGO（FINOS Labs 的受监管行业多智能体治理规范）提出的方案是在 A2A 之上叠加"合规层"，把地理/监管元数据作为capabilities.extensions中的一个**能力扩展（AgentExtension）**携带，而不是修改协议本身。本 recipe 完整复刻了这一模式。

cards.py 定义了一个稳定唯一的扩展 URI 作为查找键：

GEOGRAPHIC_EXTENSION_URI = ( "https://openeago.finos.org/extensions/geographic-metadata/v1" )

_geographic_extension()构造的扩展携带六个参数字段，与 OpenEAGO 文档使用的字段一致：jurisdiction（管辖司法管辖区）、data_center、geographic_location、data_residency_regions（数据驻留区域列表）、cross_border_restrictions（禁止流向的区域）、compliance（合规标签列表）。

build_regional_agent_card()（cards.py）把这些元数据封装进一个真实的a2a.types.AgentCard：agent_id同时充当卡片name与本地子智能体名，url采用local://cross-border-data-router/agents/{agent_id}形式——在真实部署中这里会替换为从在线 A2A 注册中心获取的url。本 recipe 刻意保持注册表静态、进程内（manifest.yaml中architecture.datasources: hardcoded），让策略本身成为被检验的主角，而非网络/注册表管道。

geographic_metadata(card)（cards.py）是策略引擎读取卡片元数据的唯一入口：遍历card.capabilities.extensions，按 URI 匹配后返回params字典；若卡片未声明该扩展则抛出ValueError。路由资格完全由卡片声明决定，绝无旁路通道。

区域处理器注册表（AGENT_REGISTRY）

注意uk_processor声明了["UK", "EU"]双驻留，因此对 EU 驻留要求同样合规；而us_processor仅覆盖US。这组声明正是后面路由算法演示三个示例查询的输入基础。

四、策略引擎：先硬过滤、后打分的三阶段路由算法

策略引擎位于 app/policy/engine.py，其文档字符串明确说明它实现了 OpenEAGO 文档所描述的 Phase 2（Planning & Negotiation）发现流程中三阶段智能体选择算法的简化版：

1. Stage 1 — 数据驻留硬过滤：淘汰任何声明的data_residency_regions未覆盖请求全部必需区域的候选；
2. Stage 2 — 管辖权排除硬过滤：淘汰任何jurisdiction落在请求排除清单上的幸存候选；
3. Stage 3 — 打分：按管辖权偏好与合规标签重叠度对幸存者排序，并列时按注册表顺序打破。

硬过滤先于打分且永不被打分推翻——不合规的智能体不可能靠"高分"挤进合格名单。若无人幸存，请求被整体拒绝，理由中明确要求升级给人审（human-in-the-loop），而不是"尽力而为"地路由，这与 OpenEAGO 将不可解决的跨境冲突视为合规违规（其ComplianceViolationError/ 人审门）的处理框架一致。

打分权重定义于 engine.py（OpenEAGO 自身模型还含成本/SLA/延迟项，本 recipe 为聚焦驻留主题而刻意省略）：

_JURISDICTION_WEIGHT = 0.7 _COMPLIANCE_WEIGHT = 0.3

_score()（engine.py）的计算细节：

• 管辖权分：命中preferred_jurisdictions得 1.0；合规但不被偏好得 0.5；无偏好要求时合规即得 0.7；
• 合规分：从卡片compliance标签中筛出与数据分类相关或含reg:/residency:前缀的标签，compliance_score = min(len(relevant_tags) / 2, 1.0)；
• 总分：score = 0.7 × 管辖权分 + 0.3 × 合规分，保留三位小数。

evaluate_routing_policy()（engine.py）是入口函数：遍历候选卡片 → 依次执行驻留过滤、排除过滤 → 无幸存者则返回decision="rejected"并附上每个淘汰原因（eliminated字典）→ 有幸存者则打分排序，取最高分者返回decision="approved"，同时输出score_breakdown（全体幸存者分数）与selection_reasons（胜出原因），保证决策透明可审计。candidates参数默认取模块级AGENT_REGISTRY，但可注入覆盖，这正是单元测试得以直接驱动它的设计。

五、数据模型：DataRequest 与 RoutingDecision

app/policy/models.py 定义了三个 frozen dataclass，形状与 OpenEAGO 规范 Phase 2 发现流程的请求/响应结构对应（简化为"数据驻留 + 管辖权排除"两个教学维度）：

• DataRequest：请求方声明的一次数据处理请求，字段包括：
  • data_classification（如 "PII"、"financial"、"public"，信息性字段，仅随决策透传用于审计）；
  • origin_region（数据主体/记录来源，如 "Germany"、"EU"，同样仅供审计）；
  • required_residency（数据必须驻留的区域代码元组，卡片data_residency_regions必须全部覆盖）；
  • excluded_jurisdictions（无论驻留声明如何都禁止处理的司法管辖区，如合同性跨境限制，默认空）；
  • preferred_jurisdictions（软性偏好，只用于打分、绝不用于淘汰，默认空）。
• ScoredCandidate：通过硬过滤的候选及其score、reasons。
• RoutingDecision：评估结果，decision为"approved"/"rejected"；批准时含selected_agent_id（即AgentCard.name）、jurisdiction、selection_reasons、score_breakdown；拒绝时含reason与eliminated（被淘汰者及原因）。

六、编排者指令与路由工具：让 LLM 只做"传话人"

策略决策必须来自工具而非模型直觉。ORCHESTRATOR_INSTRUCTION（app/prompt.py）给根智能体规定了一条严格四步序列：

1. 解析请求中的data_classification、origin_region、required_residency（从来源与提及的法规推断：欧盟来源 + GDPR → EU/EEA 驻留；美国数据 + CCPA → US 驻留；若调用方显式声明则以其为准）、excluded_jurisdictions、preferred_jurisdictions；
2. 必须调用evaluate_and_route，绝不跳过、绝不自行猜测区域；
3. 若返回"rejected"：如实转达reason并停止，不自行处理记录、不降级到不合规区域，跨境冲突应交给人工审查；
4. 若返回"approved"：调用名称匹配selected_agent_id的子智能体工具（eu_processor/uk_processor/us_processor），转达确认结果，并用selection_reasons简要解释选择理由。

"始终对策略推理保持透明——这个系统的存在是为了让跨境数据处理决策可审计，而不只是正确。"（prompt 结尾原话）

路由工具evaluate_and_route（app/tools/routing_tool.py）是编排者与策略引擎之间的薄适配层：把参数组装成DataRequest（列表转元组），调用evaluate_routing_policy，再asdict序列化为 LLM 可读的字典返回。其 docstring 明确要求：调用任何区域处理器之前必须先调用本工具；收到"rejected"时不得自行尝试处理记录或挑选降级目标，而应把拒绝转达给调用方。把"评估"与"执行"拆成两个独立步骤，正是决策可审计的关键——编排者绝不能把记录交给未通过合规审查的子智能体。

七、环境准备与安装

前置条件（与 README 一致）：

• uv：Python 包管理器（本 recipe 使用 uv 管理依赖与运行环境）；
• Gemini API Key或启用了 Vertex AI 的 Google Cloud 项目（二选一即可）。

在 recipe 根目录contrib/python/cross-border-data-router/下执行：

# 1. 安装依赖（根据 pyproject.toml 锁定 python >=3.11,<3.14） uv sync # 2. 配置凭据：复制 .env.example 为 .env 并填写 cp .env.example .env

.env.example 中的关键配置：

# 模型名称 MODEL_NAME=gemini-3.5-flash # Vertex AI 方式（二选一） # GOOGLE_CLOUD_PROJECT=<TODO: update-this-value> # GOOGLE_CLOUD_LOCATION=global # GOOGLE_GENAI_USE_VERTEXAI=True # Gemini API Key 方式（二选一） # GEMINI_API_KEY=<TODO: update-this-value>

依赖声明见 pyproject.toml：google-adk[gcp,a2a]>=2.0.0,<3.0.0、a2a-sdk>=0.3,<1、python-dotenv，开发组依赖pytest。注意.env.example注释中的警告：LOGS_BUCKET_NAME若填成非真实桶名的占位值，会因被当作"已启用 GCS 工件存储"而破坏uv run uvicorn app.fast_api_app:app，本地开发保持留空即可；OTEL_TO_CLOUD默认在部署环境（Cloud Run 设置K_SERVICE）开启、本地关闭。

八、运行：CLI 交互模式与 FastAPI/ADK Dev UI

方式一：命令行交互

uv run adk run app

README 给出了三个可直接试用的示例提示词，恰好覆盖算法的三种分支：

• "A customer in Germany wants their PII record processed. Which regional agent should handle it?"——德国客户的 PII，推断必需驻留 EU/EEA，eu_processor与uk_processor双双通过硬过滤（uk_processor声明了 EU 驻留），进入打分阶段；
• "Route this to whichever EU-compliant agent is preferred in the UK."——同样的驻留要求，但请求表达了 UK 偏好，uk_processor凭借preferred_jurisdictions加分赢得打分阶段；
• "A US customer's financial record must stay in the US, but US-based processors are contractually excluded. Route it."——唯一驻留合规的us_processor恰好是被排除的司法管辖区，全体候选被淘汰，路由器直接拒绝，而非静默选择不合规区域。

方式二：FastAPI 开发服务器

uv run uvicorn app.fast_api_app:app --reload

app/fast_api_app.py 基于google.adk.cli.fast_api.get_fast_api_app构建，自动装配 recipe 目录（agents_dir）下的智能体，并暴露 ADK Web UI；默认启用内存会话（session_service_uri = None），可通过LOGS_BUCKET_NAME切换 GCS 工件存储、通过ALLOW_ORIGINS配置 CORS。文件末尾还注册了POST /feedback接口用于采集反馈（可用 Cloud Logging，失败时回退本地日志）。启动后可打开 ADK Dev UI 以图形界面与智能体对话。

九、测试与验证：策略行为由单测直接锁定

uv run pytest # 运行单元测试套件（tests/integration 默认被排除） uv run pytest tests/unit # 只跑策略引擎与工具的单测 uv run pytest tests/integration # 需要真实凭据的集成测试，CI 中排除

pytest 配置 通过addopts = "--ignore=tests/integration"默认排除集成测试（其需要真实 LLM 凭据），并静默了 google-adk 自身的弃用警告。

tests/unit/test_policy_engine.py 直接以DataRequest驱动evaluate_routing_policy，逐条锁定上文描述的算法行为：

• 注册表完备性：三个卡片都必须携带含非空jurisdiction与data_residency_regions的地理扩展；
• 欧盟 PII 路由：required_residency=("EU",)时决策批准，胜者 ∈ {eu_processor,uk_processor}，且score_breakdown非空；
• 偏好管辖获胜：同驻留要求下追加preferred_jurisdictions=("UK",)，胜者必须是uk_processor；
• 排除管辖绝不被选中：required_residency=("US",)且excluded_jurisdictions=("US",)时决策为rejected、selected_agent_id is None，且us_processor出现在eliminated中——即使它是唯一驻留合规者；
• 无合规者直接拒绝：请求 APAC 驻留（无任何卡片覆盖）时拒绝且三个候选全部进入eliminated，无打分；
• 美国财务记录无约束：required_residency=("US",)时胜者为us_processor，jurisdiction == "US"。

配套的 tests/unit/test_tools.py 覆盖evaluate_and_route工具及各区域子智能体的process_record；tests/test_runnability.py则保证 recipe 可被 import 与启动。这组测试既是行为规范，也是读者理解算法边界的可运行样例。

十、命令速查

结语：可复用的合规路由范式

cross-border-data-router的价值在于把一个常被"想当然"处理的问题——多智能体环境下谁有权碰数据——变成了一个声明式、可审计、可测试的工程问题：地域元数据由智能体在 A2AAgentCard扩展中自我声明，路由资格由显式策略引擎基于硬过滤 + 打分判定，LLM 只负责解析请求与转达结果，不合格时宁缺毋滥地拒绝并升级人审。从 manifest.yaml 的元数据（standalone 类型、multi-agent、无状态、数据源 hardcoded、依赖 ADK / a2a-sdk / pydantic / Vertex AI）到源码与单测，整个 recipe 都围绕"合规路由"这一主线展开。如果要在真实环境落地，只需把静态注册表替换为在线 A2A 注册中心、把process_record换成真实处理管道，策略引擎与决策流程可以原样复用。

【免费下载链接】adk-samplesA collection of sample agents built with Agent Development Kit (ADK)项目地址: https://gitcode.com/GitHub_Trending/ad/adk-samples