Adk Fleet Safety Multi Agent System

舰队安全情报平台

问题陈述

如果我们希望自动驾驶车队是安全的，我们的路线系统需要发展。这个项目是我解决这个问题的尝试。

一次事故可能会让车队付出代价 数以万计 一旦考虑到停机时间和间接成本，大多数路由工具仍然会针对预计到达时间进行优化，而忽略：

• 事故密度 --历史热点与碰撞模式
• 限速波动 --可变限制、学区、建筑
• 疲劳信号 --驾驶时间、休息合规性、警觉性模式
• 现场事件 --实时交通事件和危险
• 电动汽车行驶里程行为 --电池退化、寒冷天气影响、充电可用性

这个问题是多方面的。传统路线优化了单一目标（最快路线）。舰队安全需要同时进行多个维度的推理，并随着条件的变化而调整。

为什么这很重要： 它可以拯救生命，而不仅仅是金钱。

______________________________________________________________________

代理人案例

传统软件之所以举步维艰，是因为 车队安全需要同时在多个维度上进行连续推理我选择基于代理的架构是出于特定的技术原因：

1. 多种专业智能

车队安全涉及五个相互关联的问题：

• 路线生成（顺序推理）
• 安全评估（并行处理）
• 实时监控（连续循环）
• 高管见解（批量聚合）
• 总体协调（情境感知决策）

一个单一的系统将与这些相互竞争的问题作斗争。专业代理在其特定领域表现出色。

2. 动态决策

条件不断变化（交通、天气、疲劳）。代理实时对这些变化进行推理，并根据情况的发展调整建议。

3. 大规模工具编排

该系统需要协调7+谷歌地图API。代理自然会选择正确的工具，智能地链调用，并优雅地处理故障。

4. 可解释推理

代理人可以解释 *为什么* 选择了一条路线（例如，“避开事故频发的高速路段”），这对信任至关重要。

______________________________________________________________________

建造了什么

总体架构

A. 6-agent协调系统 在那里，专业的人工智能代理协同工作，防止车队事故。这里的关键架构决策是使用ADK AgentTool pattern——这意味着子代理显示为 嵌套调用 在痕迹中，这样你就可以真正看到委托的发生。

                 ┌─────────────────────────────┐
                 │  fleet_safety_orchestrator  │
                 │  (Central Coordinator)      │
                 │  Uses AgentTool to delegate │
                 └──────────┬──────────────────┘
                            │ AgentTool calls
      ┌──────────┬──────────┼──────────┬──────────┐
      │          │          │          │          │
      ▼          ▼          ▼          ▼          ▼
 ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐ ┌────────┐
 │ route_ │ │safety_ │ │rerouter│ │ risk_  │ │analy-  │
 │planner_│ │scorer_ │ │_agent  │ │monitor_│ │ tics_  │
 │ agent  │ │ agent  │ │        │ │ agent  │ │ agent  │
 └───┬────┘ └───┬────┘ └───┬────┘ └───┬────┘ └───┬────┘
     │          │          │          │          │
     └──────────┴──────────┴──────────┴──────────┘
                        │
                        ▼
           ┌────────────────────────┐
           │ Google Maps MCP Server │
           │  (11 Tools)            │
           │                        │
           │  • get_directions      │
           │  • calculate_route_    │
           │    safety_factors      │
           │  • search_places       │
           │  • get_traffic_        │
           │    conditions          │
           │  • snap_to_roads       │
           │  • geocode_address     │
           │  • get_speed_limits    │
           │  • distance_matrix     │
           │  • get_place_details   │
           │  • reverse_geocode     │
           │  • get_route_          │
           │    elevation_gain      │
           └────────────────────────┘

代理人

每个代理人都是一个合适的 LlmAgent 有自己的指导和工具。当编排者通过以下方式调用它们时 AgentTool，它们以自己的LLM推理独立运行——您将在跟踪中看到这是嵌套调用。

1. 快速_安全_辅导员

• 角色：中央协调员----决定让哪些专业人员参与
• 模式：主代理 AgentTool 授权
• 它的作用：维护车队状态，将请求发送给专家，做出最终决定，汇总结果

2. 路线规划代理

• 角色：生成最佳路线备选方案
• 模式：顺序工作流（验证→ 生成→ 计算成本→ rank)
• 关键能力：呼叫谷歌地图MCP获取真实路线，处理柴油和 电动车 车辆

3. 安全_核心_代理

• 角色：评估路线安全（0-100分）
• 模式：多重因素的平行评估
• 得分封面：限速、交通模式、路线复杂性、历史事故数据、天气、驾驶员体验
• 电动汽车专用：考虑寒冷天气下的航程风险

4. 重新路由代理

• 角色：监控活动行程，必要时触发重新路由
• 模式：事件驱动（响应事件）
• 触发器：交通延误>15分钟，道路封闭，天气危险， 电动汽车临界低电量

5. 风险监控代理

• 角色：实时遥测监测
• 模式：持续监测
• 检测：急刹车、超速、驾驶员疲劳模式

6. analytics_agent

• 角色：历史分析和预测
• 模式：用于报告的批处理
• 能力：趋势分析、驾驶员评分、预测风险建模， 能效跟踪 （电动汽车千瓦时/英里）

谷歌地图MCP服务器

此项目使用我的自定义 谷歌地图MCP服务器 --用于谷歌地图平台API的生产就绪模型上下文协议服务器。

 ](https://pypi.org/project/google-maps-mcp-server/) 

主要特点：

• 11工具 跨越地点、方向、地理编码、距离矩阵、道路和高程API
• 生产准备就绪：强大的错误处理，指数回退的自动重试
• 通用集成：适用于Claude Desktop、Google ADK和任何兼容MCP的客户端
• 类型安全：带Pydantic验证的完整类型注释

可用工具：

安装：

uv pip install google-maps-mcp-server

请参阅 全部文件 用于设置和使用。

关键技术特点

• 适当的多代理委托：使用ADK AgentTool 模式——子代理在跟踪中显示为嵌套调用
• 6名专业代理：每个人都有不同的职责和自己的法学硕士推理
• 电动汽车和柴油车队支持：处理充电停止、里程焦虑、能源成本
• 实时天气集成：Open-Meteo API数据输入安全评分
• 自定义MCP服务器：7个谷歌地图API中的11个工具
• 异步贯穿始终：适当 async/await 生产用图案

______________________________________________________________________

展示和演示（ADK Web UI）

ADK Web UI是您在开发过程中花费大部分时间的地方。它确切地显示了正在发生的事情：哪些代理被调用，他们在想什么，以及每一步需要多长时间。

阅读展示指南 查看演示提示和要查找的内容。

截图

代理商回复安全建议

编排者委托 route_planner_agent 和 risk_monitor_agent，然后汇总他们的发现：

ADK Response

带有嵌套代理调用的跟踪视图

这是关键点——你可以看到子代理出现在 嵌套调用，而不仅仅是函数调用。每个代理都运行自己的LLM：

ADK Trace

使用工具调用的事件详细信息

展开任何步骤，查看传递了哪些参数以及返回了哪些参数：

ADK Event Details

检验申请

有助于调试——请参阅LLM看到的系统说明和可用工具：

ADK Request

代理架构图

编排器及其工具的视觉概述（包括 AgentTool 子代理包装）：

ADK Graph

云部署（顶点AI代理引擎）

生产部署到代理引擎，MCP服务器在GKE上单独运行：

Cloud Deployment

快速开始

5分钟后起床跑步。

1.克隆和安装

git clone https://github.com/ettysekhon/adk-fleet-safety-multi-agent-system.git
cd adk-fleet-safety-multi-agent-system
make install

2.设置环境

cp .env_example .env
# Edit .env with your API keys (see below)

3.本地运行

make playground
# Open http://localhost:8000

环境变量

最小化 .env 快速启动：

GOOGLE_API_KEY=your-gemini-api-key
MCP_SERVER_URL=http://XX.XX.X.XXX/sse

小贴士：使用 MCP_SERVER_URL 意味着您不需要自己的谷歌地图API密钥。

______________________________________________________________________

本地运行

选项1:ADK Web用户界面

Web UI提供视觉跟踪、工具调用日志和对话历史记录。

make playground

打开 在您的浏览器中。

选项2:CLI演示

运行交互式命令行演示：

make demo

选项3：快速验证

直接运行代理模块：

make run

______________________________________________________________________

部署到Google Cloud

看 部署.md.

分步部署指南

按照以下步骤将舰队安全代理部署到您自己的GCP项目中。

第一步：先决条件

确保您拥有：

• 谷歌云帐户 已启用计费
• gcloud命令行界面 已安装(安装指南)
• Python 3.12 和 紫外线 包管理器

步骤2：使用GCP进行身份验证

# Login to Google Cloud
gcloud auth login

# Set your project
gcloud config set project YOUR_PROJECT_ID

# Enable required APIs
gcloud services enable aiplatform.googleapis.com storage.googleapis.com

步骤3：设置环境变量

# Copy the example file
cp .env_example .env

# Edit .env with your values

你的 .env 文件应包含：

# Required for local development (Gemini LLM)
GOOGLE_API_KEY=your-gemini-api-key

# For MCP Server - choose ONE:
# Option A: Use remote MCP server (recommended)
MCP_SERVER_URL=http://XX.XX.X.XXX/sse

# Option B: Use local MCP server (requires your own Maps API key)
# GOOGLE_MAPS_API_KEY=your-maps-api-key

备注: GOOGLE_API_KEY 在部署过程中会自动过滤掉。代理引擎改用服务帐户身份验证。

步骤4：安装依赖项

make install

步骤5：部署

make deploy

或者直接使用自定义选项运行部署脚本：

uv run python app/app_utils/deploy.py \
  --display-name="fleet-safety-agent" \
  --description="Fleet Safety Multi-Agent System" \
  --env-file=.env

这将：

1. 打包您的代理代码
2. 上传到GCS暂存桶（自动创建）
3. 部署到顶点AI代理引擎
4. 生成 deployment_metadata.json 使用您的代理引擎ID

预期产量：

Deployment successful!
Service Account: service-XXX@gcp-sa-aiplatform-re.iam.gserviceaccount.com

Open Console Playground: https://console.cloud.google.com/vertex-ai/agents/...

步骤6：测试部署

# Interactive mode
uv run python scripts/query_deployed_agent.py

# Single query
uv run python scripts/query_deployed_agent.py --query "What is the fleet status?"

带有GitHub操作的CI/CD

该仓库包括两个工作流：

需要GitHub机密

将这些添加到您的仓库设置（设置→ 秘密→ 行动）：

设置工作负载身份联合

如果您没有配置WIF，请运行以下命令一次：

# Variables
PROJECT_ID=your-project-id
PROJECT_NUMBER=$(gcloud projects describe $PROJECT_ID --format='value(projectNumber)')
POOL_NAME=github-pool
PROVIDER_NAME=github-provider
SA_NAME=github-actions-sa

# Create workload identity pool
gcloud iam workload-identity-pools create $POOL_NAME \
  --location="global" \
  --project=$PROJECT_ID

# Create OIDC provider for GitHub
gcloud iam workload-identity-pools providers create-oidc $PROVIDER_NAME \
  --location="global" \
  --workload-identity-pool=$POOL_NAME \
  --issuer-uri="https://token.actions.githubusercontent.com" \
  --attribute-mapping="google.subject=assertion.sub,attribute.repository_owner=assertion.repository_owner" \
  --attribute-condition="assertion.repository_owner == 'YOUR_GITHUB_USERNAME'" \
  --project=$PROJECT_ID

# Create service account
gcloud iam service-accounts create $SA_NAME \
  --display-name="GitHub Actions" \
  --project=$PROJECT_ID

# Grant required roles
gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="serviceAccount:$SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/aiplatform.admin"

gcloud projects add-iam-policy-binding $PROJECT_ID \
  --member="serviceAccount:$SA_NAME@$PROJECT_ID.iam.gserviceaccount.com" \
  --role="roles/storage.admin"

# Allow GitHub to impersonate the service account
gcloud iam service-accounts add-iam-policy-binding \
  $SA_NAME@$PROJECT_ID.iam.gserviceaccount.com \
  --role="roles/iam.workloadIdentityUser" \
  --member="principalSet://iam.googleapis.com/projects/$PROJECT_NUMBER/locations/global/workloadIdentityPools/$POOL_NAME/attribute.repository_owner/YOUR_GITHUB_USERNAME" \
  --project=$PROJECT_ID

# Print the provider path (use this for GCP_WORKLOAD_IDENTITY_PROVIDER secret)
echo "projects/$PROJECT_NUMBER/locations/global/workloadIdentityPools/$POOL_NAME/providers/$PROVIDER_NAME"

替换 YOUR_GITHUB_USERNAME 使用您的GitHub用户名或组织（例如。， ettysekhon).

部署什么

生成的文件

部署故障排除

“权限被拒绝”错误：

# Ensure you're authenticated
gcloud auth login
gcloud auth application-default login

“API未启用”错误：

gcloud services enable aiplatform.googleapis.com storage.googleapis.com

部署时间过长（>10分钟）：

• 首次部署可能需要5-10分钟
• 在GCP控制台中检查云构建日志

MCP服务器架构

在本地，代理将MCP服务器作为子进程（stdio）生成。在生产环境中，代理引擎阻止子进程生成，因此MCP服务器通过SSE传输单独部署到GKE：

┌─────────────────────────────────────────────────────────────────────┐
│  Vertex AI Agent Engine                                             │
│  ┌─────────────────────┐                                            │
│  │  ADK Agent          │                                            │
│  └─────────────────────┘                                            │
└───────────│─────────────────────────────────────────────────────────┘
            │ HTTP/SSE
            ▼
┌─────────────────────────────────────────────────────────────────────┐
│  GKE (Google Kubernetes Engine)                                     │
│  ┌─────────────────────┐                                            │
│  │  MCP Server         │  ← http:///sse                │
│  │  (SSE Transport)    │                                            │
│  └─────────────────────┘                                            │
└─────────────────────────────────────────────────────────────────────┘

选项1:GKE部署（推荐）

MCP服务器部署到GKE，可通过SSE访问：

# Set the remote MCP server URL
export MCP_SERVER_URL=http://XX.XX.X.XXX/sse

# Test the connection
curl http://XX.XX.X.XXX/health
# Expected: {"status":"healthy","version":"0.2.1","service":"google-maps-mcp-server"}

# Run locally with remote MCP
make playground

选项2：云运行部署

gcloud run deploy google-maps-mcp-server \
    --source . \
    --set-env-vars="GOOGLE_MAPS_API_KEY=your-key" \
    --allow-unauthenticated

环境变量：

当 MCP_SERVER_URL 设置后，代理通过SSE连接。否则，它将退回到本地子流程（stdio）进行开发

______________________________________________________________________

项目结构

以下是代码库的组织方式。首先要了解的关键文件如下：

├── app/
│   ├── agents/fleet_safety/      # START HERE - the actual agents
│   │   ├── agent.py              # Entry point - creates & wires up all agents
│   │   ├── orchestrator.py       # Central coordinator with AgentTool wrappers
│   │   ├── route_planner_agent.py
│   │   ├── safety_scorer_agent.py
│   │   ├── dynamic_rerouter_agent.py
│   │   ├── risk_monitor_agent.py
│   │   ├── analytics_agent.py
│   │   └── evaluation/           # ADK evaluation configs
│   ├── agent.py                  # Root entry point (imports from agents/fleet_safety/)
│   ├── agent_engine_app.py       # Production wrapper for Vertex AI deployment
│   ├── app_utils/
│   │   ├── deploy.py             # Agent Engine deployment script
│   │   ├── requirements-deploy.txt  # Production dependencies
│   │   ├── telemetry.py          # OpenTelemetry setup
│   │   └── types.py              # Pydantic models
│   └── helpers/
│       ├── env.py                # Environment variable loading
│       └── weather.py            # Open-Meteo weather integration
├── scripts/
│   └── query_deployed_agent.py   # CLI for testing deployed agents
├── tests/
│   ├── unit/                     # No API keys needed - uses mocks
│   ├── integration/              # Needs MCP server running
│   └── load_test/                # Locust load testing
├── docs/
│   ├── screenshots/              # UI screenshots
│   └── ai_course/                # Kaggle course materials
├── .github/workflows/
│   ├── ci.yml                    # Lint + unit tests on PR
│   └── deploy.yml                # Deploy to Agent Engine on push to main
├── Makefile                      # Common commands - run `make help`
├── SHOWCASE_GUIDE.md             # Demo prompts and what to look for
├── DEPLOYMENT.md                 # Agent Engine vs GKE - architecture deep dive
└── deployment_metadata.json      # Auto-generated after deployment (gitignored)

从哪里开始：

1. 阅读 app/agents/fleet_safety/agent.py --查看代理是如何创建和连接在一起的
2. 阅读 orchestrator.py --看看如何 AgentTool 包装专业代理商
3. 跑 make playground 并尝试以下提示 SHOWCASE_GUIDE.md

场景1：智能路线规划（柴油机）

设置：从伦敦到曼彻斯特的HGV路线。

你会在痕迹中看到什么:

1. fleet_safety_orchestrator 接收请求
2. 嵌套 route_planner_agent 调用--生成路由备选方案
3. 嵌套 safety_scorer_agent 调用--评估每条路由
4. Orchestrator聚合并推荐最安全的选项

场景2：电动汽车路线

设置：从剑桥到爱丁堡的电动面包车（180英里航程vs 300英里行程）。

电动汽车有什么不同:

1. route_planner_agent 确定范围约束
2. 自动添加充电停止 处于最佳中点
3. 计算 充电时间 （45分钟快速充电）和 能源成本 （0.45英镑/千瓦时）
4. safety_scorer_agent 标记影响电池续航里程的寒冷天气风险

场景3：实时重新路由

设置M25发生交通事故。

你会看到什么:

1. rerouter_agent 检测到超过15分钟的延迟
2. 可以链接到 route_planner_agent 对于替代方案
3. 在推荐之前评估安全/时间权衡

场景4：高管仪表板

设置：日终审查。

你会看到什么:

1. 编排员 generate_executive_dashboard 工具已调用
2. analytics_agent 调用历史数据
3. 仪表板包括 能效 电动汽车指标

______________________________________________________________________

构建

技术栈

• 核心：谷歌ADK、Gemini 2.5 Flash、Python 3.12、异步
• 应用程序编程接口：谷歌地图平台（通过MCP提供7个API）
• 工具：MCP协议、折线、uv包管理器

关键架构决策

如果你正在构建类似的东西，这些是值得理解的：

1.自定义MCP服务器（与直接API调用相比）

我们构建了一个自定义的MCP服务器，而不是让代理直接调用谷歌地图API。

为什么这很重要：MCP服务器提供了一个干净的抽象。代理人只是调用以下工具 get_directions-他们不需要了解API身份验证、速率限制或响应解析。它还允许我们创建复合工具，如 calculate_route_safety_factors 协调多个API调用。

2. AgentTool 模式（与直接方法调用相比）

子代理被包裹在 AgentTool 而不是称为Python方法。

为什么这很重要：与 AgentTool，每个子代理都运行自己的LLM，并在跟踪中显示为嵌套调用。没有它，你只会看到一个函数调用——无法看到子代理在“想什么”。这对于调试和理解代理行为至关重要。

3.单一编排器（与点对点）

一个中央协调器协调所有专家，而不是代理直接相互交谈。

为什么这很重要：舰队行动往往有相互冲突的优先事项（安全、速度和成本）。中央协调器可以明确地做出这些权衡。这也使系统更容易推理——有一个地方会发生“最终决定”。

______________________________________________________________________

未来的增强功能

立即（欢迎PR）

1. 生产数据库：用真实的机群数据库（PostgreSQL/BigQuery）替换合成数据
2. 更多eval场景：当前的eval套件是基本的——添加边缘案例
3. 更好的错误处理：一些MCP故障可以更优雅地处理

中期

1. 驾驶员个性化：基于个人安全记录的路线偏好
2. 预测性维护集成：在路线决策中考虑车辆健康状况

长期

1. 计算机视觉：Dashcam代理用于实时危险检测
2. ML模型：基于实际车队数据的列车事故预测
3. V2V通信：车队行驶的车辆间协调

______________________________________________________________________

故障排除

“未列出工具‘calculate_route_safety_factors’”

MCP服务器包已过期。更新它：

uv lock --upgrade-package google-maps-mcp-server
uv sync

“只调用了一个代理”

LLM决定呼叫哪些代理。在提示中明确：

• ❌ “规划从a到B的路线”
• ✅ “规划从a到B的路线 评估安全性 每个选项”

“子代理未显示为嵌套调用”

检查 orchestrator.py 用途 AgentTool:

AgentTool(agent=self.route_planner, skip_summarization=True)

如果它直接调用方法（例如。， await self.route_planner.generate_route_options(...))，您将看不到嵌套调用。

“MCP连接失败”

任一组 MCP_SERVER_URL 用于远程或设置 GOOGLE_MAPS_API_KEY 对于本地。代理首先尝试远程，然后回退到本地。

______________________________________________________________________

测试与评估

这里有两层测试： 代码层 （代码是否有效？）以及 认知层 （代理人是否做出了明智的决定？）。

快速参考

1.单元测试（从这里开始）

这些使用 模拟MCP客户端-无API密钥，无网络。非常适合CI/CD和开发过程中的快速反馈。

make test-unit

# Or run specific tests
uv run pytest tests/unit/test_route_planner.py -v
uv run pytest tests/unit/test_safety_scorer.py -v

2.集成测试

这些击中了 真正的谷歌地图MCP服务器。您需要：

• MCP_SERVER_URL set（连接到远程服务器），或
• GOOGLE_MAPS_API_KEY set（生成本地MCP子流程）

# Remote MCP (recommended - no local setup needed)
export MCP_SERVER_URL=http://XX.XX.X.XXX/sse
make test-integration

# Or local MCP
export GOOGLE_MAPS_API_KEY=your-maps-api-key
make test-integration

3.ADK评估（代理推理）

这是一个有趣的部分——它测试代理是否 明智的决定，而不仅仅是代码是否运行。ADK发送测试提示并对响应进行评分。

make eval

eval配置位于 app/agents/fleet_safety/evaluation/请查看 .evalset.json 查看测试了哪些场景的文件。

4.全套套房

一次性完成所有操作：

./run_evals.sh

______________________________________________________________________

学习资源

这个项目是作为 Kaggle 5天人工智能代理强化课程 与谷歌。

谷歌白皮书

本课程包括五篇关于代理人工智能的优秀白皮书，非常值得从头到尾阅读：

课程材料

看 docs/ai_course/README.md 完整的课程细分、Kaggle笔记本链接和涵盖所有五天的示例脚本。