视觉驱动的基础设施检测AI代理
去年夏天，我女儿的滑板掉进了一个雨水排水沟。这需要我抬起一个60磅的井盖，并在她监督下爬下去。
从那天起，她就对街道下的东西着迷了。首先是为找回滑板而感到高兴。然后情况升级了。“如果一只小猫掉下去怎么办？” “有多少只小猫卡在街道下的管道里？” “我们怎么知道管道的位置？” 最后，她用完全真诚的态度提出了一个百万美元的建议：“爸爸，我们应该发明一种街道X光机，这样我们就能找到丢失的东西。”
我把它当作玩笑，然后我开始思考。然后我想到：AI真的能像公用事业工人那样通过视觉线索来读取街道上的线索吗？
1、这实际上是什么
《隐形城市》 是一个多智能体视觉系统的实验。让我明确说明，它不是一种生产用的定位工具，而是一种探索能否将推理编码到协作AI代理中，仅凭视觉线索检测地表基础设施标记（井盖、消防栓、喷漆代码）并推断地下情况。具有纳米香蕉和分割工具的AI代理，LFG！
可以这样理解：有经验的公用事业工人可以通过观察街道做出关于地下网络的有根据的猜测。他们知道 APWA 颜色代码。他们知道典型的管道深度。他们识别模式。
这个项目问的是：我们可以将这种推理编码到协作的 AI 代理中吗？
答案出人意料地复杂……但我认为这正是它引人入胜的原因。
这是什么：
- 使用 Google 的 Agent Dev Kit (ADK) 的多代理编排技术演示
- 使用 Gemini 3 Pro Preview 和 Gemini 3 ProImage Preview 探索结构化视觉输出
- 实现 Agent-to-Agent (A2A) 协议的真实案例
- 一种有趣的方式向好奇的孩子解释复杂的基础设施
这不是什么：
- 足够准确以指导实际挖掘（这需要穿透地面雷达和认证的定位器）
- 在你挖掘之前拨打 811 的替代品
- 任何你可以用于施工决策的信任对象
在给出这个警告之后，让我们谈谈它是如何工作的——因为这里的 技术模式适用于任何多代理视觉系统，无论领域如何。
2、架构
Surface Eye 是我 A2A 租户中的三个专业代理之一。实际上，我最初是为另一个目的构建这个代理的，但鉴于它拥有的工具，它在这个用例中表现得非常好：
Surface Eye (视觉) → Pattern Oracle (推理) → Depth Renderer (可视化)
检测标记 推理地下拓扑结构 生成 X 光视图
本文重点介绍第一个代理。未来的帖子将涵盖推理、协调和将它们联系在一起的 Agent-to-Agent (A2A) 协议。
该组件的技术栈：
- 最初使用 Gemini 2.5 Flash 进行视觉 + 结构化输出，但升级到 Gemini 3 Pro Image Preview 用于此演示
- Pydantic 用于类型安全的模式
- Google ADK 1.19.0 用于代理工具
- CloudRun/FastAPI 用于部署
3、核心挑战：结构化视觉输出
大多数视觉 AI 演示返回自然语言描述：“我在两辆停着的汽车之间看到一个红色消防栓。” 这对人类很好，但对于代理间通信无用。
Surface Eye 需要产生结构化的、类型化的数据，下游代理可以解析和推理。具体来说：
{
"markers": [
{
"id": "marker_000",
"marker_type": "fire_hydrant",
"utility_type": "water",
"bounding_box": {"y0": 508, "x0": 390, "y1": 680, "x1": 441},
"confidence": 1.0,
"reasoning": "红消防栓可见于停放车辆之间"
}
]
}
这就是 Gemini 的 JSON 模式 变得关键的地方。
4、定义你的类型系统
在编写任何检测代码之前，我使用 Pydantic 定义了模式。这有三个目的：
- 类型安全：Python 知道它期望的数据
- 验证：无效数据会立即被发现
- 文档：模式 就是 API 合同
以下是基础：
from enum import Enum
from pydantic import BaseModel, Field
from typing import Optional
class MarkerType(str, Enum):
"""我们要检测的表面基础设施标记类型。"""
MANHOLE = "manhole"
UTILITY_MARKER = "utility_marker"
FIRE_HYDRANT = "fire_hydrant"
VALVE_COVER = "valve_cover"
ELECTRICAL_BOX = "electrical_box"
JUNCTION_BOX = "junction_box"
SURVEY_MARKER = "survey_marker"
PAVEMENT_CUT = "pavement_cut"
class UtilityType(str, Enum):
"""APWA 颜色编码的公用事业类型。"""
ELECTRIC = "electric" # 红色
GAS = "gas" # 黄色
WATER = "water" # 蓝色
SEWER = "sewer" # 绿色
TELECOM = "telecom" # 橙色
STORM = "storm" # 绿色
RECLAIMED = "reclaimed" # 紫色
UNKNOWN = "unknown"
class BoundingBox(BaseModel):
"""归一化的边界框 [y0, x0, y1, x1] 在 0-1000 范围内。"""
y0: int = Field(..., ge=0, le=1000)
x0: int = Field(..., ge=0, le=1000)
y1: int = Field(..., ge=0, le=1000)
x1: int = Field(..., ge=0, le=1000)
class SurfaceMarker(BaseModel):
"""检测到的表面基础设施标记。"""
id: str = Field(..., description="此标记的唯一标识符")
marker_type: MarkerType
utility_type: UtilityType
bounding_box: BoundingBox
mask_base64: Optional[str] = Field(None, description="Base64 PNG 分割掩码")
label: str = Field(..., description="Gemini 的描述性标签")
confidence: float = Field(..., ge=0.0, le=1.0)
reasoning: str = Field(default="", description="检测和位置的推理")
properties: dict = Field(default_factory=dict, description="类型特定的属性")
@property
def centroid(self) -> tuple[float, float]:
"""返回归一化的中心点 (x, y) 在 0-1 范围内。"""
cx = ((self.bounding_box.x0 + self.bounding_box.x1) / 2) / 1000
cy = ((self.bounding_box.y0 + self.bounding_box.y1) / 2) / 1000
return (cx, cy)
为什么使用枚举？ *它们防止拼写错误。MarkerType.FIRE_HYDRANT 在解析时被验证。字符串如 "firehydrant" (没有下划线) 会失败验证，提前捕获错误。
为什么 0–1000 坐标？ Gemini 的边界框默认使用此比例。我们将其归一化为 0–1 用于渲染，但保留原始格式以与模型输出保持一致。
为什么 Field(...)？ 省略号使字段为必填项—Pydantic 如果缺少它们会引发验证错误。这会提前捕获 LLM 的不完整数据。
为什么 mask_base64？ Gemini 可以选择返回 Base64 PNG 图像作为分割掩码。我们存储这些用于下游可视化。
完整的响应包装所有内容：
class SurfaceAnalysis(BaseModel):
"""Surface Eye 的完整分析结果。"""
image_id: str
image_width: int
image_height: int
markers: list[SurfaceMarker]
analysis_timestamp: str
model_version: str
processing_time_ms: int
5、检测的提示工程
让 Gemini 返回结构化、准确的检测需要仔细的提示设计。以下方法有效：
prompt = """
分析这张街道视图图像并识别公用事业基础设施标记。
首先，对图像进行推理：
1. 确定视角和地面平面。
2. 寻找特定的公用事业指示器：
- 路面/草地上的喷漆标记（红色=电力，黄色=燃气，蓝色=水，绿色=污水，橙色=电信）
- 颜色旗帜或桩子
- 井盖和阀门盖
- 公用事业箱和底座
3. 对于每个潜在标记，验证它是否确实是公用事业标记，而不是通用物体（如石头或垃圾）。
返回一个对象列表的 JSON。对于每个对象，提供：
- label: 简短的描述
- marker_type: [manhole, utility_marker, fire_hydrant, valve_cover, electrical_box, junction_box, survey_marker, pavement_cut] 中的一个
- utility_type: [electric, gas, water, sewer, telecom, storm, reclaimed, unknown] 中的一个
- box_2d: [ymin, xmin, ymax, xmax] 格式的 2D 边界框（0-1000 比例）
- confidence: 0.0 到 1.0 之间的分数
- reasoning: 为什么将其识别为公用事业标记以及如何确定位置的简要解释。
专注于寻找喷漆标记、彩色旗帜和地面上的公用事业盖。
"""
关键提示设计决策：
- 思维链推理： 要求模型“首先对图像进行推理”提高了准确性。它迫使系统化分析后再进行检测。
- 领域知识注入： 将 APWA 颜色代码直接包含在提示中，为模型提供了它未明确训练过的上下文。
- 显式字段定义： 列出确切的枚举值可防止虚构类别。
- 推理字段： 要求模型解释“为什么”它检测到某物可减少误报。如果它无法解释，它很可能错了。
小技巧： “验证它确实是一个公用事业标记”这一短语是在 Gemini 不断将随机圆形物体检测为井盖后添加的。对假阳性进行明确说明很重要。
6、使用 JSON 模式调用 Gemini
这里是核心检测函数：
import os
import json
import io
from datetime import datetime
from google import genai
from google.genai import types
from PIL import Image
def get_genai_client() -> genai.Client:
"""获取一个配置正确的 Genai 客户端。"""
project = os.getenv("GOOGLE_CLOUD_PROJECT")
location = os.getenv("GOOGLE_CLOUD_LOCATION", "us-central1")
if project:
# 在 GCP 中使用 Vertex AI
return genai.Client(vertexai=True, project=project, location=location)
else:
# 使用 API 密钥进行本地开发
api_key = os.getenv("GOOGLE_API_KEY")
if not api_key:
raise ValueError("设置 GOOGLE_CLOUD_PROJECT 或 GOOGLE_API_KEY")
return genai.Client(api_key=api_key)
async def segment_infrastructure(
image_data: bytes,
prompt_type: str = "comprehensive",
client: genai.Client | None = None
) -> SurfaceAnalysis:
"""
使用 Gemini 从图像中分割基础设施标记。
Args:
image_data: 原始图像字节（JPEG 或 PNG）
prompt_type: 使用的提示模板
client: 可选的预配置 Genai 客户端
Returns:
包含所有检测到的标记的 SurfaceAnalysis
"""
if client is None:
client = get_genai_client()
# 加载图像以获取尺寸
image = Image.open(io.BytesIO(image_data))
width, height = image.size
start_time = datetime.utcnow()
# 配置为结构化 JSON 输出
config = types.GenerateContentConfig(
response_mime_type="application/json"
)
response = await client.aio.models.generate_content(
model="gemini-3-pro-image-preview",
contents=[
types.Part.from_bytes(
data=image_data,
mime_type=f"image/{image.format.lower()}"
),
prompt
],
config=config
)
end_time = datetime.utcnow()
processing_time_ms = int((end_time - start_time).total_seconds() * 1000)
# 解析响应为类型化的对象
markers = _parse_segmentation_response(response.text, width, height)
return SurfaceAnalysis(
image_id=f"img_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}",
image_width=width,
image_height=height,
markers=markers,
analysis_timestamp=datetime.utcnow().isoformat(),
model_version="gemini-3-pro-preview",
processing_time_ms=processing_time_ms
)
神奇的一行：
config = types.GenerateContentConfig(
response_mime_type="application/json"
)
这告诉 Gemini*: “只返回有效的 JSON。不要有 markdown 包装。不要有解释性文本。只有 JSON。”*
实际上，Gemini 有时仍然用 markdown 代码块包装输出。这就是为什么我们需要强大的解析。
7、健壮的响应解析
即使启用了 JSON 模式，LLM 有时也会返回意外的格式。这里是有防御性的解析：
def _parse_segmentation_response(
response_text: str,
img_width: int,
img_height: int
) -> list[SurfaceMarker]:
"""将 Gemini 的 JSON 响应解析为类型化的 SurfaceMarker 对象。"""
# 处理 markdown 包裹的 JSON
text = response_text.strip()
if text.startswith("```json"):
text = text[7:]
if text.startswith("```"):
text = text[3:]
if text.endswith("```"):
text = text[:-3]
text = text.strip()
try:
items = json.loads(text)
except json.JSONDecodeError as e:
logger.error("无法解析分割 JSON", error=str(e))
return [] # 优雅失败
markers = []
for i, item in enumerate(items):
try:
# 解析边界框（Gemini 的 0-1000 比例）
box = item.get("box_2d", [0, 0, 100, 100])
bbox = BoundingBox(
y0=int(box[0]),
x0=int(box[1]),
y1=int(box[2]),
x1=int(box[3])
)
# 映射到类型化的枚举并提供回退
marker_type_str = item.get("marker_type", "unknown").lower().replace(" ", "_")
try:
marker_type = MarkerType(marker_type_str)
except ValueError:
marker_type = MarkerType.UTILITY_MARKER # 安全回退
utility_type_str = item.get("utility_type", "unknown").lower()
try:
utility_type = UtilityType(utility_type_str)
except ValueError:
utility_type = UtilityType.UNKNOWN
marker = SurfaceMarker(
id=f"marker_{i:03d}",
marker_type=marker_type,
utility_type=utility_type,
bounding_box=bbox,
label=item.get("label", f"标记 {i}"),
confidence=float(item.get("confidence", 0.8)),
reasoning=item.get("reasoning", ""),
properties={}
)
markers.append(marker)
except Exception as e:
logger.warning("无法解析标记", index=i, error=str(e))
continue # 跳过此标记，处理其他标记
return markers
为什么这很重要： 如果一个检测的边界框格式错误，我们仍然希望返回其他 10 个有效的检测。在解析错误时快速失败会浪费整个 LLM 调用。
小技巧： 记录解析失败。当你调试为什么某些标记没有出现时，这些日志是黄金。
8、封装为 ADK 工具
现在我们将这个函数提供给 ADK 代理：
from google.adk.tools import FunctionTool, ToolContext
async def segment_infrastructure_tool(
image_base64: str,
prompt_type: str = "comprehensive",
tool_context: ToolContext = None
) -> dict:
"""
从街景图像中分割公用事业基础设施。
Args:
image_base64: Base64 编码的图像数据
prompt_type: 分割提示类型（comprehensive, manholes, utility_markers 等）
tool_context: ADK 工具上下文
Returns:
包含 SurfaceAnalysis 结果的字典
"""
import base64
from .segmentation import segment_infrastructure
image_data = base64.b64decode(image_base64)
analysis = await segment_infrastructure(image_data, prompt_type)
return analysis.model_dump()
# 创建 ADK 的 FunctionTool
segment_tool = FunctionTool(func=segment_infrastructure_tool)
为什么 prompt_type？ Surface Eye 支持针对不同检测场景的专用提示, "manholes" 专注于井盖, "utility_markers" 关注喷漆和旗帜等。默认的 "comprehensive" 捕捉所有内容。
为什么 base64？ ADK 代理通过 JSON 通信。二进制图像数据不能很好地序列化，但 base64 字符串可以。它稍微低效，但与代理协议兼容。
9、定义代理
最后，我们将所有内容封装成一个 ADK 代理：
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
SYSTEM_INSTRUCTION = """你是 Surface Eye，一位专家基础设施检测代理。
你的角色是分析街景或航拍图像并识别所有可见的地下公用事业指标。你使用 Gemini 的对话图像分割来为每个标记生成精确的掩码。
## 检测优先级（按顺序）：
1. 井盖 - 分类类型（污水、雨水、电力、电信）
2. APWA 公用事业标记 - 通过颜色代码识别
3. 消防栓 - 注意流量速率的颜色编码
4. 电力基础设施 - 变压器、接线盒
5. 阀门盖 - 水、气
6. 路面切割 - 表示最近的公用事业工作
7. 测量标记 - 粉色旗帜/桩子
## 输出要求：
- 为每个检测到的标记生成分割掩码
- 使用 APWA 颜色标准对每个进行分类
- 根据可见性和清晰度估计置信度
- 注明任何模糊或部分遮挡的标记
## APWA 颜色标准：
- 红色：电力
- 黄色：燃气、油、蒸汽
- 橙色：电信、电缆、信号
- 蓝色：饮用水
- 绿色：污水、排水
- 紫色：再生水
- 粉色：测量/临时
- 白色：拟议的挖掘
当给定一张图像时，使用 segment_infrastructure 工具检测所有标记，
然后使用 classify_marker 和 extract_marker_properties 进行详细分析。
"""
surface_eye_agent = LlmAgent(
name="surface_eye",
model="gemini-2.5-flash",
instruction=SYSTEM_INSTRUCTION,
tools=[segment_tool, classify_tool, extract_tool],
description="从街景图像中分割和分类表面基础设施标记"
)SYSTEM_INSTRUCTION = """你是 Surface Eye，一位专家基础设施检测代理。
系统指令做了三件事：
- 建立身份： “你是 Surface Eye” 给代理提供了回应的上下文
- 编码领域知识： APWA 标准成为代理推理过程的一部分
- 将工具串联在一起： 代理使用
segment_infrastructure进行检测，然后使用classify_marker和extract_marker_properties进行深入分析
10、实际输出示例
这是 Surface Eye 在分析带有消防栓的街景照片时产生的内容：
{
"image_id": "img_20251129_224633",
"image_width": 2048,
"image_height": 1153,
"markers": [
{
"id": "marker_000",
"marker_type": "fire_hydrant",
"utility_type": "water",
"bounding_box": {"y0": 508, "x0": 390, "y1": 680, "x1": 441},
"label": "消防栓",
"confidence": 1.0,
"reasoning": "这是一个明显的红色消防栓，是常见的水公用事业基础设施标记。它的位置在两辆停着的汽车之间。"
},
{
"id": "marker_001",
"marker_type": "utility_marker",
"utility_type": "unknown",
"bounding_box": {"y0": 356, "x0": 400, "y1": 671, "x1": 415},
"label": "临时公用事业标记桩",
"confidence": 0.9,
"reasoning": "在消防栓后面可以看到一根细长的红白条纹桩。这些临时标记由公用事业团队用来标记地下线路。"
},
{
"id": "marker_002",
"marker_type": "valve_cover",
"utility_type": "unknown",
"bounding_box": {"y0": 624, "x0": 504, "y1": 660, "x1": 539},
"label": "圆形公用事业盖",
"confidence": 0.95,
"reasoning": "一个嵌入沥青中的圆形金属盖，特征是阀门盖或小型井盖，表示地下公用事业接入。"
}
],
"analysis_timestamp": "2025-11-29T22:46:33.716168",
"model_version": "gemini-2.5-flash",
"processing_time_ms": 11545
}
处理时间：约 11 秒用于 2048×1153 图像。足以用于交互使用。
Surface Eye 正确识别了：
- 以 1.0 置信度正确识别了消防栓
- 发现了实用标记桩（容易被忽视）
- 在路面中找到了阀门盖
下一步： Pattern Oracle 会利用这个结构化输出来推断地下网络拓扑结构——水管深度、管道直径、连接点。
你会失去类型检查。请到处使用枚举。
11、结束语
Surface Eye 生成结构化数据，但它不会推理它所看到的内容。
这就是另一个专门代理，称为 Pattern Oracle 的地方。它接收 Surface Eye 的检测并应用领域知识：
- “这个消防栓是红色的→很可能是水主管，流量大于 1000 GPM”
- “井盖间距是 300 英尺→这是污水管，不是雨水排水管”
- “橙色喷漆+附近的电信箱→光纤导管”
Pattern Oracle 推断不可见的内容：管道深度、直径、材料、连接点。全部从地表线索。有趣的东西。
在下一篇文章中，我们将使用结构化推理链构建 Pattern Oracle，并展示代理如何使用 A2A 协议传递类型化数据。
原文链接：Building a Vision-Powered Infrastructure Detection Agent with Gemini 3
汇智网翻译整理，转载请标明出处