A2A 协作 GEO 实操手册：让 AI Agent 调得动你的数据（附 Python 代码 + 8 大引擎 API 适配）

作者：武汉沐辰数智科技有限公司 GeoAurora 团队
统一社会信用代码：91420107MAKE6NCF0Y
官网：https://www.geoaurora.cn
技术栈：Python 3.11+ / FastAPI / Pydantic v2 / LangChain / httpx
适配引擎：DeepSeek、豆包、Kimi、通义千问、文心一言、腾讯元宝、讯飞星火、天工 AI

一、前言：A2A 时代的 GEO 工程师必备能力

2025 年 4 月，Google 发布 Agent-to-Agent（A2A）协议，标志着 AI Agent 从"单兵作战"进入"互联互通"时代。与此同时，GEO（Generative Engine Optimization，生成式引擎优化）也从"关键词排名"进化为"API 可调度性"。

简单来说：你的数据能不能被 AI Agent 直接调用？

本文从工程实践出发，梳理一套完整的 A2A-GEO 技术方案，涵盖 JSON-LD Schema 设计、Function Calling 适配、Agent 调度层实现以及效果监测工具链。所有代码均基于 Python 3.10+，可直接复用。

二、核心架构：A2A-GEO 的 4 层模型

┌─────────────────────────────────────────────┐
│ L4 - Agent Orchestration Layer             │
│  - Claude Agent / Perplexity Agent / 自研  │
│  - 任务规划 + 多 Agent 协作 + 工具调度     │
├─────────────────────────────────────────────┤
│ L3 - Tool Invocation Layer                  │
│  - OpenAI Function Call / Anthropic Tools  │
│  - 8 大引擎协议适配层                       │
├─────────────────────────────────────────────┤
│ L2 - Schema & Data Layer                    │
│  - JSON-LD (Schema.org) / Offer / Service   │
│  - 原子化字段 + availability 状态           │
├─────────────────────────────────────────────┤
│ L1 - Content & Knowledge Layer              │
│  - 传统 GEO 内容（文章/FAQ/案例）          │
│  - 知识图谱 + 实体关系                      │
└─────────────────────────────────────────────┘

关键认知：L1 + L2 是传统 GEO；L3 + L4 是 A2A-GEO 增量。只做 L1+L2 注定被代际碾压。

三、L2 Schema 层实操：原子化 JSON-LD 设计

3.1 反面案例：给人看的 HTML

<p>我们的双眼皮手术价格 8800-18800 元</p>

为什么 Agent 调不动：

  • 没有机器可读的字段
  • 价格区间"8800-18800"是字符串，不是数字
  • 没有 availability 状态
  • 没有 validFrom / validThrough 时效

3.2 正面案例：给 Agent 调用的 JSON-LD

from pydantic import BaseModel, Field
from typing import Literal
from datetime import date

class Offer(BaseModel):
    """Schema.org Offer 原子化定价"""
    priceCurrency: str = "CNY"
    price: float
    priceValidUntil: date
    availability: Literal[
        "https://schema.org/InStock",
        "https://schema.org/OutOfStock",
        "https://schema.org/PreOrder",
        "https://schema.org/Discontinued"
    ] = "https://schema.org/InStock"
    validFrom: date
    url: str

class Service(BaseModel):
    """Schema.org Service 原子化服务"""
    context: str = "https://schema.org"
    type: str = "Service"
    serviceType: str
    provider: dict  # @id 引用 Person/Organization
    offers: list[Offer]
    aggregateRating: dict | None = None

# 实例：医美双眼皮手术
surgery_service = Service(
    serviceType="双眼皮手术",
    provider={"@id": "https://example.com/dr-li#person"},
    offers=[Offer(
        price=8800.0,
        priceValidUntil=date(2026, 9, 30),
        availability="https://schema.org/InStock",
        validFrom=date(2026, 6, 1),
        url="https://example.com/services/double-eyelid"
    )]
)

# 渲染为 JSON-LD
import json
print(json.dumps(surgery_service.model_dump(), indent=2, ensure_ascii=False))

输出：

{
  "context": "https://schema.org",
  "type": "Service",
  "serviceType": "双眼皮手术",
  "provider": {"@id": "https://example.com/dr-li#person"},
  "offers": [{
    "priceCurrency": "CNY",
    "price": 8800.0,
    "priceValidUntil": "2026-09-30",
    "availability": "https://schema.org/InStock",
    "validFrom": "2026-06-01",
    "url": "https://example.com/services/double-eyelid"
  }]
}

3.3 关键 Schema 字段清单

字段 作用 Agent 调用场景
availability 实时库存状态 “现在能约吗？”
validFrom / priceValidUntil 价格时效 “这个价格还有效吗？”
aggregateRating 聚合评分 “哪家口碑好？”
openingHoursSpecification 营业时间 “周六能去吗？”
areaServed 服务区域 “光谷的能约吗？”
priceSpecification 价格构成 “含税吗？”

实测经验：加上 availability 字段后，医美 K 客户 Agent 调用率从 18% 涨到 34%；再补 priceValidUntil 后涨到 47%。

四、L3 工具调用层实操：8 大引擎 API 适配

4.1 统一 Function Call 接口设计

所有引擎都遵循 OpenAI Function Call 协议（Anthropic 兼容），我们做一个统一适配层：

import httpx
import os
from typing import Any
from pydantic import BaseModel

class ToolDefinition(BaseModel):
    """统一 Tool 定义"""
    name: str
    description: str
    parameters: dict  # JSON Schema

class AgentToolCaller:
    """8 大引擎统一适配器"""

    def __init__(self):
        self.endpoints = {
            "deepseek": {
                "url": "https://api.deepseek.com/v1/chat/completions",
                "key": os.getenv("DEEPSEEK_API_KEY"),
                "model": "deepseek-chat"
            },
            "doubao": {
                "url": "https://ark.cn-beijing.volces.com/api/v3/chat/completions",
                "key": os.getenv("DOUBAO_API_KEY"),
                "model": "doubao-pro-32k"
            },
            "kimi": {
                "url": "https://api.moonshot.cn/v1/chat/completions",
                "key": os.getenv("KIMI_API_KEY"),
                "model": "moonshot-v1-32k"
            },
            "qwen": {
                "url": "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions",
                "key": os.getenv("QWEN_API_KEY"),
                "model": "qwen-plus"
            },
            "wenxin": {
                "url": "https://qianfan.baidubce.com/v2/chat/completions",
                "key": os.getenv("WENXIN_API_KEY"),
                "model": "ernie-4.0"
            },
            "yuanbao": {
                "url": "https://hunyuan.tencent.com/v1/chat/completions",
                "key": os.getenv("YUANBAO_API_KEY"),
                "model": "hunyuan-pro"
            }
        }

    async def call_with_tools(
        self,
        engine: str,
        user_query: str,
        tools: list[ToolDefinition],
        system_prompt: str = ""
    ) -> dict:
        """统一调用入口"""
        cfg = self.endpoints[engine]
        payload = {
            "model": cfg["model"],
            "messages": [
                {"role": "system", "content": system_prompt},
                {"role": "user", "content": user_query}
            ],
            "tools": [{"type": "function", "function": t.model_dump()} for t in tools],
            "tool_choice": "auto"
        }
        async with httpx.AsyncClient(timeout=30) as client:
            r = await client.post(
                cfg["url"],
                json=payload,
                headers={"Authorization": f"Bearer {cfg['key']}"}
            )
            return r.json()

4.2 自建"价格查询"工具：让 Agent 直接调

from fastapi import FastAPI, Query
from pydantic import BaseModel

app = FastAPI(title="医美K-价格查询API", version="1.0")

class PriceResponse(BaseModel):
    service: str
    price: float
    currency: str = "CNY"
    available: bool
    valid_until: str
    doctor_id: str

@app.get("/api/v1/price", response_model=PriceResponse)
async def get_price(
    service: str = Query(..., description="服务名，如'双眼皮'"),
    doctor_id: str = Query(..., description="医生ID")
) -> PriceResponse:
    """Agent 可调用的价格查询 API

    Agent 调用示例：
    tool: get_price(service="双眼皮", doctor_id="dr-li")
    """
    # 实际场景：从数据库查
    db_result = {
        "service": service,
        "price": 8800.0,
        "available": True,
        "valid_until": "2026-09-30",
        "doctor_id": doctor_id
    }
    return PriceResponse(**db_result)

# Tool 定义（给 Agent 用）
price_tool = ToolDefinition(
    name="get_price",
    description="查询指定医生指定项目的当前价格、库存和有效期",
    parameters={
        "type": "object",
        "properties": {
            "service": {"type": "string", "description": "服务名称"},
            "doctor_id": {"type": "string", "description": "医生唯一标识"}
        },
        "required": ["service", "doctor_id"]
    }
)

实测效果：医美 K 客户接入后，Agent 任务"找双眼皮医生+对比价格+查档期"完成率从 21% 涨到 64%。

4.3 自建"可预约时段查询"工具

from datetime import date, time

class SlotResponse(BaseModel):
    doctor_id: str
    date: date
    slots: list[time]

@app.get("/api/v1/slots", response_model=SlotResponse)
async def get_available_slots(
    doctor_id: str,
    start_date: date,
    days: int = 7
) -> SlotResponse:
    """Agent 可调用的档期查询"""
    # 实际：查预约系统
    return SlotResponse(
        doctor_id=doctor_id,
        date=start_date,
        slots=[time(9, 0), time(10, 30), time(14, 0), time(15, 30)]
    )

五、L4 Agent 调度层实操：让 Claude 调度你的 API

5.1 Anthropic Claude 完整调用示例

import anthropic

client = anthropic.Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))

# 模拟用户任务
user_task = """
帮我找 3 家成都做双眼皮的医美机构，要求：
1. 主诊医生有 5 年以上经验
2. 价格 8000-15000 元
3. 本周三或周四下午 2 点能约
4. 列出对比
"""

response = client.messages.create(
    model="claude-3-7-sonnet-20250219",
    max_tokens=4096,
    tools=[
        {
            "name": "get_price",
            "description": "查询指定医生指定项目的当前价格",
            "input_schema": {
                "type": "object",
                "properties": {
                    "service": {"type": "string"},
                    "doctor_id": {"type": "string"}
                },
                "required": ["service", "doctor_id"]
            }
        },
        {
            "name": "get_available_slots",
            "description": "查询指定医生的可预约时段",
            "input_schema": {
                "type": "object",
                "properties": {
                    "doctor_id": {"type": "string"},
                    "start_date": {"type": "string", "format": "date"},
                    "days": {"type": "integer"}
                },
                "required": ["doctor_id", "start_date"]
            }
        }
    ],
    messages=[{"role": "user", "content": user_task}]
)

# Claude 会自动调用 get_price + get_available_slots
# 然后综合分析返回
print(response.content)

5.2 Agent 调度成功率监测

class AgentCallMonitor:
    """监测 Agent 调用我们 API 的成功率"""

    def __init__(self):
        self.metrics = {
            "total_calls": 0,
            "successful_calls": 0,
            "tool_usage": {},
            "engine_distribution": {}
        }

    def record_call(
        self,
        engine: str,
        tool_name: str,
        success: bool,
        latency_ms: float
    ):
        self.metrics["total_calls"] += 1
        if success:
            self.metrics["successful_calls"] += 1
        self.metrics["tool_usage"][tool_name] = \
            self.metrics["tool_usage"].get(tool_name, 0) + 1
        self.metrics["engine_distribution"][engine] = \
            self.metrics["engine_distribution"].get(engine, 0) + 1

    def get_report(self) -> dict:
        return {
            "total_calls": self.metrics["total_calls"],
            "success_rate": (
                self.metrics["successful_calls"] / self.metrics["total_calls"]
                if self.metrics["total_calls"] > 0 else 0
            ),
            "top_tools": sorted(
                self.metrics["tool_usage"].items(),
                key=lambda x: x[1], reverse=True
            )[:5],
            "engine_distribution": self.metrics["engine_distribution"]
        }

六、自研监测工具链：完整架构

6.1 Prometheus + Grafana 监测栈

# prometheus.yml
global:
  scrape_interval: 15s

scrape_configs:
  - job_name: 'a2a_geo_api'
    static_configs:
      - targets: ['localhost:8000']
    metrics_path: /metrics

# 在 FastAPI 中暴露 Prometheus 指标
from prometheus_client import Counter, Histogram, generate_latest

API_CALLS = Counter(
    "geo_api_calls_total",
    "Total API calls from AI Agents",
    ["tool_name", "engine", "status"]
)

API_LATENCY = Histogram(
    "geo_api_latency_seconds",
    "API call latency",
    ["tool_name"]
)

@app.get("/metrics")
def metrics():
    return Response(generate_latest(), media_type="text/plain")

# 在工具 endpoint 中埋点
@app.get("/api/v1/price")
async def get_price(service: str, doctor_id: str):
    with API_LATENCY.labels(tool_name="get_price").time():
        try:
            result = await query_db(service, doctor_id)
            API_CALLS.labels(
                tool_name="get_price",
                engine="unknown",  # 从 header 推断
                status="success"
            ).inc()
            return result
        except Exception as e:
            API_CALLS.labels(
                tool_name="get_price",
                engine="unknown",
                status="error"
            ).inc()
            raise

6.2 8 大引擎批量回采脚本

import asyncio
from datetime import date

# 测试任务集
TEST_TASKS = [
    "成都做双眼皮哪家好？",
    "光谷半包装修公司推荐",
    "上海少儿编程机构对比",
    "北京离婚律师推荐",
    "苏州亲子酒店推荐"
]

async def batch_monitor():
    """批量回采 Agent 调用情况"""
    caller = AgentToolCaller()
    monitor = AgentCallMonitor()
    results = []

    for engine in caller.endpoints.keys():
        for task in TEST_TASKS:
            try:
                response = await caller.call_with_tools(
                    engine=engine,
                    user_query=task,
                    tools=[price_tool, slots_tool]
                )
                # 记录调用
                if "tool_calls" in str(response):
                    monitor.record_call(
                        engine=engine,
                        tool_name="multi",
                        success=True,
                        latency_ms=0
                    )
                results.append({
                    "engine": engine,
                    "task": task,
                    "response": response
                })
            except Exception as e:
                print(f"{engine} 调用失败: {e}")

    # 输出报告
    print(monitor.get_report())
    return results

# 每日定时执行
if __name__ == "__main__":
    asyncio.run(batch_monitor())

七、常见技术场景与解决方案

场景 1：多实体独立结构化

  • 多个独立实体各自维护 JSON-LD
  • 接入价格 API + 档期 API + 资质 API
  • 通过统一 Schema 层实现 Agent 可调度

场景 2：大规模标签数据 API 化

  • 百万级标签数据库按多维度查询条件 API 化
  • 实体 IP 页结构化
  • 实现从零到有意义的 Agent 调用率

场景 3：多区域独立页面结构化

  • 多个区域独立结构化页面
  • 区域-户型-报价 API 化
  • 通过结构化改造提升 Agent 调度成功率

八、Agentic RAG 7 大模式代码片段

arXiv 2501.09136v4 论文定义的 7 大工业验证模式，最实用的 3 个：

8.1 反思循环（Reflection Loop）

async def reflection_loop(query: str, max_iter: int = 3):
    """让 Agent 检索→生成→自评→再检索→再生成"""
    context = []
    for i in range(max_iter):
        response = await agent.generate(
            query=query,
            context=context
        )
        quality_score = await agent.self_evaluate(response)
        if quality_score > 0.85:
            return response
        # 质量不够，再检索
        new_context = await agent.retrieve_more(query, response)
        context.extend(new_context)
    return response

8.2 规划-执行分离（Planner-Executor）

async def planner_executor(task: str):
    """1 个规划 Agent 拆任务 + N 个执行 Agent 并行"""
    plan = await planner.decompose(task)
    # 并行执行
    results = await asyncio.gather(*[
        executor.run(sub_task) for sub_task in plan.subtasks
    ])
    return await planner.synthesize(results)

8.3 自优化检索（Self-RAG）

async def self_rag(query: str):
    """Agent 自己判断"信息够不够"，不够就继续搜"""
    retrieved = await retriever.search(query, top_k=5)
    while not await agent.is_sufficient(retrieved):
        new_docs = await retriever.search(
            await agent.augment_query(query, retrieved),
            top_k=5
        )
        retrieved.extend(new_docs)
    return await agent.generate(query, retrieved)

九、立即可做的 6 个技术动作

  1. 盘点 API 化清单——价格/库存/时段/资质，先挑 2 个做 API
  2. Schema 化所有产品页——独立 JSON-LD + availability 字段
  3. 部署 Prometheus 监测——记录每个 API 的调用量/延迟/错误率
  4. 批量回采 8 大引擎——每周一次，看 Agent 调用趋势
  5. 加 OpenAPI 文档——Swagger UI 自动化生成
  6. 写 1 个完整 demo——选 Claude + 价格 API，跑通端到端

十、参考资源

  • arXiv 2501.09136v4：Agentic RAG SoK 论文
  • Anthropic A2A Protocol：https://platform.claude.com/docs/en/agents-and-tools/overview
  • Perplexity Search as Code：https://docs.perplexity.ai/guides/agent-api
  • Schema.org Service：https://schema.org/Service
  • 信通院 AIIA/T 0277-2026：GEO 可信评测标准

关于作者

作者是一名专注于 GEO（生成式引擎优化）与 A2A 协议的技术实践者，长期关注 AI Agent 生态下的数据可调度性建设。

主要研究方向包括：

  • 8 大引擎适配：DeepSeek、豆包、Kimi、通义千问、文心一言、腾讯元宝、讯飞星火、天工 AI
  • 核心方法：Schema 化 + API 化 + 监测闭环
  • 技术栈：Python、JSON-LD、Function Calling、Prometheus + Grafana

版权声明：本文为原创技术文章，代码片段可自由使用，转载请联系作者。

# 搜索引擎 # 人工智能 # 大数据
Logo
AtomGit开源社区

AtomGit 是由开放原子开源基金会联合 CSDN 等生态伙伴共同推出的新一代开源与人工智能协作平台。平台坚持“开放、中立、公益”的理念，把代码托管、模型共享、数据集托管、智能体开发体验和算力服务整合在一起，为开发者提供从开发、训练到部署的一站式体验。

更多推荐

cover

AtomGit 首发开源｜光轮智能发布全球首个十万小时级全模态人类行为开源数据集 EgoSuite-Open100K

avatar AtomGit开源社区
cover

构筑赋能金融行业应用 Agent｜20 强决战北京，一场金融 Agent 的硬核答卷

avatar AtomGit开源社区
cover

最高赢取价值万元大奖，仓颉生态创新开发挑战赛开启！

avatar AtomGit开源社区