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 KitADK中实现企业级跨境数据合规路由由根编排智能体接收数据处理请求依据每个候选智能体在 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 由三部分构成部分路径职责根编排智能体app/agent.py解析请求字段调用路由工具再按selected_agent_id委派给对应子智能体策略引擎与数据模型app/policy/声明式合规路由的全部逻辑engine.py算法、models.py数据形状、cards.py智能体注册表区域处理子智能体app/sub_agents/三个模拟后端处理器的子智能体eu_processor、uk_processor、us_processor路由工具app/tools/routing_tool.py编排者唯一用来做路由决策的入口evaluate_and_route根智能体root_agent的定义app/agent.py清晰展示了决策工具 执行子智能体的分工def create_agent() - Agent: return Agent( nameroot_agent, modelGemini( modelos.getenv(MODEL_NAME), retry_optionstypes.HttpRetryOptions(attempts3), ), description( Routes>def process_record(record_summary: str) - str: return ( Processed in the EU-WEST region (Frankfurt data center) under fGDPR-compliant controls. Record: {record_summary} )三、用 A2A AgentCard 表达管辖权与数据驻留A2AAgent2Agent协议的AgentCard原生只描述能力capabilities没有内置的管辖权jurisdiction或数据驻留data residency概念。OpenEAGOFINOS 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.AgentCardagent_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_REGISTRYagent_id管辖权数据中心数据驻留区域跨境限制合规标签eu_processorEUGCP-EUW1-FRANKFURTEU, EEAUS, CHINA, RUSSIAReg:GDPR, Residency:EU, Control:EncryptionAtRestuk_processorUKGCP-EUW2-LONDONUK, EUUS, CHINA, RUSSIAReg:UK-GDPR, Residency:UK, Control:EncryptionAtRestus_processorUSGCP-US-CENTRAL1-IOWAUS无Reg:CCPA, Residency:US, Control:EncryptionAtRest注意uk_processor声明了[UK, EU]双驻留因此对 EU 驻留要求同样合规而us_processor仅覆盖US。这组声明正是后面路由算法演示三个示例查询的输入基础。四、策略引擎先硬过滤、后打分的三阶段路由算法策略引擎位于 app/policy/engine.py其文档字符串明确说明它实现了 OpenEAGO 文档所描述的 Phase 2Planning Negotiation发现流程中三阶段智能体选择算法的简化版Stage 1 — 数据驻留硬过滤淘汰任何声明的data_residency_regions未覆盖请求全部必需区域的候选Stage 2 — 管辖权排除硬过滤淘汰任何jurisdiction落在请求排除清单上的幸存候选Stage 3 — 打分按管辖权偏好与合规标签重叠度对幸存者排序并列时按注册表顺序打破。硬过滤先于打分且永不被打分推翻——不合规的智能体不可能靠高分挤进合格名单。若无人幸存请求被整体拒绝理由中明确要求升级给人审human-in-the-loop而不是尽力而为地路由这与 OpenEAGO 将不可解决的跨境冲突视为合规违规其ComplianceViolationError/ 人审门的处理框架一致。打分权重定义于 engine.pyOpenEAGO 自身模型还含成本/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是入口函数遍历候选卡片 → 依次执行驻留过滤、排除过滤 → 无幸存者则返回decisionrejected并附上每个淘汰原因eliminated字典→ 有幸存者则打分排序取最高分者返回decisionapproved同时输出score_breakdown全体幸存者分数与selection_reasons胜出原因保证决策透明可审计。candidates参数默认取模块级AGENT_REGISTRY但可注入覆盖这正是单元测试得以直接驱动它的设计。五、数据模型DataRequest 与 RoutingDecisionapp/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_INSTRUCTIONapp/prompt.py给根智能体规定了一条严格四步序列解析请求中的data_classification、origin_region、required_residency从来源与提及的法规推断欧盟来源 GDPR → EU/EEA 驻留美国数据 CCPA → US 驻留若调用方显式声明则以其为准、excluded_jurisdictions、preferred_jurisdictions必须调用evaluate_and_route绝不跳过、绝不自行猜测区域若返回rejected如实转达reason并停止不自行处理记录、不降级到不合规区域跨境冲突应交给人工审查若返回approved调用名称匹配selected_agent_id的子智能体工具eu_processor/uk_processor/us_processor转达确认结果并用selection_reasons简要解释选择理由。始终对策略推理保持透明——这个系统的存在是为了让跨境数据处理决策可审计而不只是正确。prompt 结尾原话路由工具evaluate_and_routeapp/tools/routing_tool.py是编排者与策略引擎之间的薄适配层把参数组装成DataRequest列表转元组调用evaluate_routing_policy再asdict序列化为 LLM 可读的字典返回。其 docstring 明确要求调用任何区域处理器之前必须先调用本工具收到rejected时不得自行尝试处理记录或挑选降级目标而应把拒绝转达给调用方。把评估与执行拆成两个独立步骤正是决策可审计的关键——编排者绝不能把记录交给未通过合规审查的子智能体。七、环境准备与安装前置条件与 README 一致uvPython 包管理器本 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_NAMEgemini-3.5-flash # Vertex AI 方式二选一 # GOOGLE_CLOUD_PROJECTTODO: update-this-value # GOOGLE_CLOUD_LOCATIONglobal # GOOGLE_GENAI_USE_VERTEXAITrue # Gemini API Key 方式二选一 # GEMINI_API_KEYTODO: update-this-value依赖声明见 pyproject.tomlgoogle-adk[gcp,a2a]2.0.0,3.0.0、a2a-sdk0.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 appREADME 给出了三个可直接试用的示例提示词恰好覆盖算法的三种分支A customer in Germany wants their PII record processed. Which regional agent should handle it?——德国客户的 PII推断必需驻留 EU/EEAeu_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 customers 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 --reloadapp/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 --ignoretests/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_processorjurisdiction US。配套的 tests/unit/test_tools.py 覆盖evaluate_and_route工具及各区域子智能体的process_recordtests/test_runnability.py则保证 recipe 可被 import 与启动。这组测试既是行为规范也是读者理解算法边界的可运行样例。十、命令速查命令说明uv sync安装依赖recipe 根目录下执行cp .env.example .env复制环境变量模板并填写凭据uv run adk run app以交互式 CLI 模式运行智能体uv run uvicorn app.fast_api_app:app --reload启动本地 FastAPI 开发服务器含 ADK Dev UIuv run pytest运行单元测试套件tests/integration默认排除uv run pytest tests/unit仅运行策略引擎与工具的单测uv run pytest tests/integration运行需要真实凭据的集成测试结语可复用的合规路由范式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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 
长河君 · 2026/9/15 15:50:58 · 阅读 1.2k Cross-Border Data Router：基于 A2A AgentCard 元数据与 OpenEAGO 模式的多智能体跨境数据合规路由实战
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 KitADK中实现企业级跨境数据合规路由由根编排智能体接收数据处理请求依据每个候选智能体在 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 由三部分构成部分路径职责根编排智能体app/agent.py解析请求字段调用路由工具再按selected_agent_id委派给对应子智能体策略引擎与数据模型app/policy/声明式合规路由的全部逻辑engine.py算法、models.py数据形状、cards.py智能体注册表区域处理子智能体app/sub_agents/三个模拟后端处理器的子智能体eu_processor、uk_processor、us_processor路由工具app/tools/routing_tool.py编排者唯一用来做路由决策的入口evaluate_and_route根智能体root_agent的定义app/agent.py清晰展示了决策工具 执行子智能体的分工def create_agent() - Agent: return Agent( nameroot_agent, modelGemini( modelos.getenv(MODEL_NAME), retry_optionstypes.HttpRetryOptions(attempts3), ), description( Routes>def process_record(record_summary: str) - str: return ( Processed in the EU-WEST region (Frankfurt data center) under fGDPR-compliant controls. Record: {record_summary} )三、用 A2A AgentCard 表达管辖权与数据驻留A2AAgent2Agent协议的AgentCard原生只描述能力capabilities没有内置的管辖权jurisdiction或数据驻留data residency概念。OpenEAGOFINOS 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.AgentCardagent_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_REGISTRYagent_id管辖权数据中心数据驻留区域跨境限制合规标签eu_processorEUGCP-EUW1-FRANKFURTEU, EEAUS, CHINA, RUSSIAReg:GDPR, Residency:EU, Control:EncryptionAtRestuk_processorUKGCP-EUW2-LONDONUK, EUUS, CHINA, RUSSIAReg:UK-GDPR, Residency:UK, Control:EncryptionAtRestus_processorUSGCP-US-CENTRAL1-IOWAUS无Reg:CCPA, Residency:US, Control:EncryptionAtRest注意uk_processor声明了[UK, EU]双驻留因此对 EU 驻留要求同样合规而us_processor仅覆盖US。这组声明正是后面路由算法演示三个示例查询的输入基础。四、策略引擎先硬过滤、后打分的三阶段路由算法策略引擎位于 app/policy/engine.py其文档字符串明确说明它实现了 OpenEAGO 文档所描述的 Phase 2Planning Negotiation发现流程中三阶段智能体选择算法的简化版Stage 1 — 数据驻留硬过滤淘汰任何声明的data_residency_regions未覆盖请求全部必需区域的候选Stage 2 — 管辖权排除硬过滤淘汰任何jurisdiction落在请求排除清单上的幸存候选Stage 3 — 打分按管辖权偏好与合规标签重叠度对幸存者排序并列时按注册表顺序打破。硬过滤先于打分且永不被打分推翻——不合规的智能体不可能靠高分挤进合格名单。若无人幸存请求被整体拒绝理由中明确要求升级给人审human-in-the-loop而不是尽力而为地路由这与 OpenEAGO 将不可解决的跨境冲突视为合规违规其ComplianceViolationError/ 人审门的处理框架一致。打分权重定义于 engine.pyOpenEAGO 自身模型还含成本/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是入口函数遍历候选卡片 → 依次执行驻留过滤、排除过滤 → 无幸存者则返回decisionrejected并附上每个淘汰原因eliminated字典→ 有幸存者则打分排序取最高分者返回decisionapproved同时输出score_breakdown全体幸存者分数与selection_reasons胜出原因保证决策透明可审计。candidates参数默认取模块级AGENT_REGISTRY但可注入覆盖这正是单元测试得以直接驱动它的设计。五、数据模型DataRequest 与 RoutingDecisionapp/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_INSTRUCTIONapp/prompt.py给根智能体规定了一条严格四步序列解析请求中的data_classification、origin_region、required_residency从来源与提及的法规推断欧盟来源 GDPR → EU/EEA 驻留美国数据 CCPA → US 驻留若调用方显式声明则以其为准、excluded_jurisdictions、preferred_jurisdictions必须调用evaluate_and_route绝不跳过、绝不自行猜测区域若返回rejected如实转达reason并停止不自行处理记录、不降级到不合规区域跨境冲突应交给人工审查若返回approved调用名称匹配selected_agent_id的子智能体工具eu_processor/uk_processor/us_processor转达确认结果并用selection_reasons简要解释选择理由。始终对策略推理保持透明——这个系统的存在是为了让跨境数据处理决策可审计而不只是正确。prompt 结尾原话路由工具evaluate_and_routeapp/tools/routing_tool.py是编排者与策略引擎之间的薄适配层把参数组装成DataRequest列表转元组调用evaluate_routing_policy再asdict序列化为 LLM 可读的字典返回。其 docstring 明确要求调用任何区域处理器之前必须先调用本工具收到rejected时不得自行尝试处理记录或挑选降级目标而应把拒绝转达给调用方。把评估与执行拆成两个独立步骤正是决策可审计的关键——编排者绝不能把记录交给未通过合规审查的子智能体。七、环境准备与安装前置条件与 README 一致uvPython 包管理器本 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_NAMEgemini-3.5-flash # Vertex AI 方式二选一 # GOOGLE_CLOUD_PROJECTTODO: update-this-value # GOOGLE_CLOUD_LOCATIONglobal # GOOGLE_GENAI_USE_VERTEXAITrue # Gemini API Key 方式二选一 # GEMINI_API_KEYTODO: update-this-value依赖声明见 pyproject.tomlgoogle-adk[gcp,a2a]2.0.0,3.0.0、a2a-sdk0.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 appREADME 给出了三个可直接试用的示例提示词恰好覆盖算法的三种分支A customer in Germany wants their PII record processed. Which regional agent should handle it?——德国客户的 PII推断必需驻留 EU/EEAeu_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 customers 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 --reloadapp/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 --ignoretests/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_processorjurisdiction US。配套的 tests/unit/test_tools.py 覆盖evaluate_and_route工具及各区域子智能体的process_recordtests/test_runnability.py则保证 recipe 可被 import 与启动。这组测试既是行为规范也是读者理解算法边界的可运行样例。十、命令速查命令说明uv sync安装依赖recipe 根目录下执行cp .env.example .env复制环境变量模板并填写凭据uv run adk run app以交互式 CLI 模式运行智能体uv run uvicorn app.fast_api_app:app --reload启动本地 FastAPI 开发服务器含 ADK Dev UIuv run pytest运行单元测试套件tests/integration默认排除uv run pytest tests/unit仅运行策略引擎与工具的单测uv run pytest tests/integration运行需要真实凭据的集成测试结语可复用的合规路由范式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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考