基于Gemini 3.1和ADK的五智能体支持台：任务分类、信息检索、执行操作、质量检测以及协调管理。

想象一下出现问题的订单：顾客称收到的商品与订购的不同，要求退款，并希望当天就寄送正确的商品。这一句话背后其实涉及三项任务：核实订单、处理退货以及加急寄送替换品。每项任务都需要对应的数据、政策及操作流程。一个仅有简单指令的单一客服机器人只能处理简单的查询，而一旦多项任务同时出现，它就容易出错——处理速度变慢、稳定性下降且错误频发。

解决方案是采用多智能体设计：多名专家各自负责工作流中的不同环节，由协调器来调度任务。下文将介绍如何利用 Google Cloud、Gemini 3.1 模型以及 Google 的 Agent Development Kit (ADK) 来实现这一设计。内容包括如何为智能体划分职责并明确边界、协调器如何路由任务、分配工作并整合结果、智能体之间如何传递状态，以及同样重要的——在何种情况下单个智能体仍是更佳选择。

单一长提示词会带来什么问题

想象一下这样一名客服智能体：它起初功能简单，但逐渐获得了更多能力——能够从知识库中检索信息、用更温和的语气与不满的客户沟通、还能自行完成质量检查。其系统提示词的长度超过了 3,000 个标记，每一次新增内容都会干扰现有的功能。此时会出现四种问题：

• 约束泛化问题。退款规则开始适用于与退款无关的查询，导致简单的产品咨询也会被用退款逻辑来回答。
• 注意力分散。模型的注意力是有限的。当需要同时处理数十条指令，且其中大部分与当前查询无关时，每条指令所获得的重视程度都会降低。
• 难以调试。当回复出错时，很难确定是哪条指令导致了问题。
• 统一处理流程。无论查询的难度如何，所有查询都必须经过相同的处理流程。

通过分解任务，每个模型都能将全部注意力集中在手头的任务上。在处理退款时只考虑退款政策，在审阅草稿时只关注语气表达。

ADK正是基于这一理念设计的。其AutoFlow机制负责管理各专家之间的任务交接：当协调器拥有子代理时，ADK会添加一个transfer_to_agent()工具，并将所有可用专家的描述写入其上下文中。这样，协调器就能清楚知道团队中有哪些成员以及每个人的专长是什么。

系统架构

该系统并非由单一提示词驱动，而是由一个中央协调器统领着一组功能有限的子代理。这些模型无需完全相同——推理能力最强的模型负责需要判断力的任务（如协调与质量审核），而速度最快、成本最低的模型则负责对效率要求较高的任务（如分类筛选）：

• 协调器：负责规划与任务分配。使用Gemini 3.1 Pro，其工具即为各专家代理。
• 分类筛选代理：负责分析情感与意图并进行任务路由。使用Gemini 3.1 Flash-Lite，辅以自然语言处理分类器。

研究代理：负责收集事实与背景信息。采用 Gemini 3.1 Flash，并配备 Vertex 搜索功能及 CRM API。

操作代理：负责执行退款、更新等交易操作。采用 Gemini 3.1 Flash，并配备订单与支付 API。

质量检查代理：负责核实事实及政策合规性。采用 Gemini 3.1 Pro，并配备政策知识库。

Gemini 2.5 Flash 依然是单代理流程的理想选择。不过，当推理质量成为路由决策的关键因素时，自 2026 年初开始推出的 Gemini 3.1 系列更为合适，尤其是在流程协调与质量检查环节。

涉及的 Google Cloud 服务包括：

• Gemini 3.1 模型，作为经过工具使用优化的推理核心。
• Agent Development Kit，作为通过 AutoFlow 实现任务交接的框架。
• Vertex AI Agent Engine 作为用于部署和扩展智能体的托管运行时。
• Firestore 用作共享会话状态，智能体可在此读取和写入对话记录。
• Vertex AI Vector Search 用于在知识库中进行检索。

设置 ADK

ADK 提供了相应的框架，包括让一个智能体能够将通话有效转接给另一个智能体的交接模式。安装只需一行命令：

pip install google-adk google-generativeai google-cloud-firestore pydantic

明确列出 Pydantic 是因为 ADK 依赖它来定义智能体的输入和输出架构，标注该库名称有助于保证版本兼容性。

项目结构应尽早规划。多智能体系统最常在接口衔接处出问题：某个智能体以整数形式发送 refund_id，而另一个智能体却期望接收字符串。采用统一架构的扁平化结构可以避免这类错误：

support_agents/
├── main.py              # Entry point for Cloud Run
├── common/
│   ├── __init__.py
│   └── schemas.py       # Shared Pydantic models (the "handshake")
├── agents/
│   ├── __init__.py
│   ├── orchestrator.py
│   ├── triage.py
│   ├── research.py
│   ├── action.py
│   └── qa.py
└── config.py            # Environment & project IDs

医院是一个恰当的类比。统筹者是分诊护士，数据采集员是实验室技术员，而执行操作的人则是外科医生。他们不会同时与患者交流。他们会读取并更新患者的病历，这里的病历存储在 Firestore 中，而 schemas.py 则是大家共同认可的病历格式。

分诊模块

分诊的作用类似于空中交通管制。它本身并不解决问题，只是确保每个请求都能以正确的优先级送达合适的专家手中。

在各个模块之间传递原始字符串容易导致数据错误。严格的 Pydantic 模式能确保统筹者收到经过验证的 Python 对象，从而无需解析自由文本。格式规范必须优先考虑：

# common/schemas.py
from pydantic import BaseModel, Field

class TriageResult(BaseModel):
    intents: list[str] = Field(description="List of detected customer goals")
    urgency: str = Field(pattern="^(low|medium|high|critical)quot;)
    sentiment: str = Field(description="Current emotional state of the user")
    required_agents: list[str] = Field(
        description="List of sub-agents needed (research, action, etc.)"
    )

然后是生成它的代理：

# agents/triage.py
from google.adk.agents import LlmAgent
from common.schemas import TriageResult

triage_agent = LlmAgent(
    name="triage_agent",
    model="gemini-3.1-flash-lite",  # Fast and cheap for classification
    instruction="""You are a triage specialist. Analyze the customer's
    message and categorize it accurately.

    Urgency rules:
    - critical: Active outages or safety threats
    - high: Frustrated users or financial issues > $100
    - medium: Standard requests needing action
    - low: General info/feedback

    You must NOT attempt to solve the problem. Only classify it.""",
    output_schema=TriageResult,  # Force structured JSON output
    output_key="triage_data"
)

在该定义中，有四个设计要点：

• 单一任务。 该代理从不回答问题或执行操作，仅负责分类。这样既能保持提示语简短，又能提高准确性。
• 结构化输出。 output_schema 要求输出符合 TriageResult 格式的有效 JSON，ADK 会通过 Pydantic 自动对其进行验证。
• 同时处理多个意图。 由于 intents 是字符串列表，因此一条消息可以同时包含更新地址和查询最新账单的请求，而单标签分类器则无法处理这种情况。
• 状态传递。 output_key="triage_data" 会将结果保存在会话状态中，这样协调器就可以通过 session.state['triage_data'] 读取该数据。

研究型代理

研究代理即调查员，它从 Firestore 和检索流程中获取原始数据，然后将其整理成结构化的简报供行动代理使用。该简报的架构便是这两个角色之间的协作规范：

# common/schemas.py (add to existing file)

class ResearchBrief(BaseModel):
    order_status: str = Field(description="Current order state")
    customer_tier: str = Field(description="e.g. Gold, Standard")
    applicable_policy: str = Field(description="Relevant policy text")
    can_refund: bool = Field(description="Whether a refund is allowed")
    reasoning: str = Field(description="Why this conclusion was reached")

接下来是它的工具以及代理定义：

# agents/research.py
from google.adk.agents import LlmAgent
from google.cloud import firestore
from common.schemas import ResearchBrief

db = firestore.Client()

def lookup_order(order_id: str) -> dict:
    """Fetch real-time shipping status and line items for an order ID."""
    doc = db.collection('orders').document(order_id).get()
    if doc.exists:
        return doc.to_dict()
    return {"error": f"Order {order_id} not found"}

def lookup_customer(customer_id: str) -> dict:
    """Retrieve customer profile, tier, and interaction history."""
    doc = db.collection('customers').document(customer_id).get()
    if doc.exists:
        return doc.to_dict()
    return {"error": f"Customer {customer_id} not found"}

def search_knowledge_base(query: str) -> list:
    """Search product docs and company policies via RAG pipeline."""
    # Uses the RAG pipeline from our previous article
    from rag_pipeline import pipeline
    result = pipeline.query(query, top_k=5)
    return result['sources']

research_agent = LlmAgent(
    name="research_agent",
    model="gemini-3.1-flash",
    instruction="""You are a research specialist.
    1. Extract identifiers (Order ID, Customer ID) from the triage data.
    2. Use tools to gather order history and relevant company policies.
    3. If policy info is missing, use search_knowledge_base.

    CRITICAL: Do not answer the customer. Output a formal ResearchBrief.""",
    tools=[lookup_order, lookup_customer, search_knowledge_base],
    output_schema=ResearchBrief,
    output_key="research_brief"
)

状态传递无需额外代码。当研究代理启动时，ADK会将上一步的triage_data注入其上下文，因此它早已知道需要查询订单号12345，因为分诊机制已识别出该意图。每个代理都在前一个代理的工作基础上继续处理，不会重复做同样的事情。一个小型的生产技巧：ADK会自动将普通函数封装为FunctionTool对象，但使用from google.adk.tools import FunctionTool显式进行封装，能在Vertex AI Agent Engine中实现更准确的延迟跟踪和令牌使用情况记录。

操作代理

只有这种专用代理拥有对业务API的写入权限。由于它执行的是高风险的操作，因此会受到严格的架构规范和明确的约束限制。其输出架构首先为：

# common/schemas.py (add to existing file)

class ActionSummary(BaseModel):
    actions_taken: list[str] = Field(description="List of actions executed")
    refund_id: str | None = Field(default=None, description="Refund ID if issued")
    escalated: bool = Field(default=False)
    escalation_reason: str | None = Field(default=None)

然后是代理程序本身，它会在高价值操作时要求人工确认：

# agents/action.py
from google.adk.agents import LlmAgent
from google.adk.tools import FunctionTool
from common.schemas import ResearchBrief, ActionSummary

def process_refund(order_id: str, amount: float, reason: str) -> dict:
    """Process a financial refund. Required: valid order_id and amount."""
    # In production: call your payment API
    return {
        "status": "processed",
        "refund_id": f"REF-{order_id}",
        "amount": amount
    }

def create_replacement_order(
    order_id: str,
    item_id: str,
    shipping_speed: str
) -> dict:
    """Create a replacement order with specified shipping speed."""
    # In production: call your order management API
    return {
        "status": "created",
        "new_order_id": f"REPL-{order_id}",
        "estimated_delivery": "2 business days"
    }

def escalate_to_human(reason: str, priority: str) -> dict:
    """Escalate to a human agent with full context."""
    return {
        "status": "escalated",
        "queue": "priority" if priority == "high" else "standard"
    }

# Wrap refund tool with human confirmation for high-value transactions
refund_tool = FunctionTool(
    func=process_refund,
    require_confirmation=True  # Pauses for human approval before executing
)

action_agent = LlmAgent(
    name="action_agent",
    model="gemini-3.1-flash",
    instruction="""You are an action specialist.
    Review the 'research_brief' in the session state.

    CRITICAL RULES:
    1. Only refund if 'can_refund' is True in the research brief
    2. If amount > $500, you MUST use escalate_to_human
    3. You must call tools for every action — do not just tell the user
       it happened
    4. If information is missing from the brief, escalate""",
    tools=[refund_tool, create_replacement_order, escalate_to_human],
    input_schema=ResearchBrief,   # Only sees research data, not raw chat
    output_schema=ActionSummary,
    output_key="action_result"
)

两项设置决定了大部分安全机制。require_confirmation的存在是因为实际生产系统中，代理很少仅凭指令提示就自行处理退款操作。ADK的工具确认流程会暂停执行，直到有人批准，而调度器则会看到“待处理”状态，从而将其转至员工控制面板。input_schema也很重要，因为如果没有它，操作代理就能看到整个对话记录，包括过时的内容。将输入限制为ResearchBrief可以防止过时信息混入，确保决策基于最新数据。

还需注意，escalate_to_human是一项刻意设定的业务规则，并非故障状态。调度器能够准确了解代理为何停止工作，随后可将工单连同完整的调研摘要一起转交给相关人员。

QA代理

QA是最后的审核环节。它不仅检查文字表述，还会将执行任务的智能体实际操作与研究任务要求进行比对，确保行动与要求一致。

该功能依托Gemini 3.1 Pro运行，因为它需要记住整个对话的上下文并识别细微的矛盾点，而在这类任务上Pro的表现始终优于Flash。ADK还允许其在批准回复之前启用思考模式，进行更深入的自我反思。审核流程如下：

# common/schemas.py (add to existing file)

class QAReview(BaseModel):
    is_approved: bool = Field(description="True if the response meets all criteria")
    feedback: str = Field(description="Specific corrections if not approved")
    final_text: str = Field(description="The customer-facing message")

代理配置如下：

# agents/qa.py
from google.adk.agents import LlmAgent
from google.adk.planners import BuiltInPlanner
from google.genai.types import ThinkingConfig
from common.schemas import QAReview

qa_agent = LlmAgent(
    name="qa_agent",
    model="gemini-3.1-pro",
    instruction="""You are a senior QA auditor. Compare the 'action_result'
    against the 'research_brief' in the session state.

    CRITICAL CHECKS:
    1. Did the Action Agent refund the EXACT amount listed in the brief?
    2. Does the response cite the policy found during research?
    3. Is the tone empathetic but professional?
    4. Are ALL intents from the triage classification addressed?
    5. No unauthorized promises, discounts, or policy exceptions

    If any check fails, set is_approved to False with specific feedback.""",
    output_schema=QAReview,
    output_key="qa_review",
    # MEDIUM for routine QA; switch to HIGH for edge cases or audits
    planner=BuiltInPlanner(
        thinking_config=ThinkingConfig(thinking_level="MEDIUM")
    )
)

被拒绝并不意味着流程结束。当 qa_review.is_approved 的值为假时，协调器会读取 feedback 字段，并将具体的修改意见反馈给相应的操作或研究代理。正是这种修正循环使得生产级代理与演示脚本有所区别。该流程最多进行两次修订，之后案件会交由人工处理。

协调器

协调器既不进行研究也不执行操作，它负责管理状态，并确保各专家按合理顺序工作。在给定 sub_agents 的情况下，ADK的AutoFlow会添加 transfer_to_agent() 工具，而协调器的LLM则会根据其指令及当前会话状态选择下一位专家：

# agents/orchestrator.py
from google.adk.agents import LlmAgent
from agents.triage import triage_agent
from agents.research import research_agent
from agents.action import action_agent
from agents.qa import qa_agent

orchestrator = LlmAgent(
    name="support_orchestrator",
    model="gemini-3.1-pro",  # Best reasoning for delegation decisions
    instruction="""You are the lead coordinator of a customer support team.
    Use the following specialists in order:
    1. triage_agent: to classify the intent and urgency
    2. research_agent: to gather order data and relevant policies
    3. action_agent: to execute the required operations
    4. qa_agent: to verify the final response

    Special cases:
    - If triage classifies as 'critical': bypass the chain and escalate
    - If QA returns is_approved=False: re-route to the responsible agent
      with the feedback (maximum 2 revision cycles, then escalate)
    - If any agent fails: escalate to human with full context

    You must never respond to the customer directly.
    Always route through the specialist pipeline.""",
    sub_agents=[triage_agent, research_agent, action_agent, qa_agent]
)

运行它

ADK通过Runner来执行智能体，每当智能体调用某个工具时，Runner会暂停执行，待结果返回后再重新启动该智能体：

# main.py
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from agents.orchestrator import orchestrator

APP_NAME = "customer_support"
USER_ID = "system"

# For production, replace with a Firestore-backed session service
session_service = InMemorySessionService()

runner = Runner(
    agent=orchestrator,
    app_name=APP_NAME,
    session_service=session_service
)

async def handle_customer_message(customer_id: str, message: str):
    """
    Entry point for customer messages.
    """
    session_id = f"support-{customer_id}"

    # Ensure session exists
    session = await session_service.get_session(
        app_name=APP_NAME, user_id=USER_ID, session_id=session_id
    )
    if not session:
        session = await session_service.create_session(
            app_name=APP_NAME, user_id=USER_ID, session_id=session_id
        )

    # run_async yields events as agents execute
    final_response = None
    async for event in runner.run_async(
        user_id=USER_ID,
        session_id=session_id,
        new_message=message
    ):
        if event.is_final_response():
            final_response = event

    return {
        "response": final_response.content.parts[0].text if final_response else None,
        "session_id": session_id
    }

为何不直接调用智能体呢？Runner负责整个流程：它保存会话状态、跟踪当前活跃的智能体、处理工具调用并记录每一步操作。由于每个智能体会在output_key下输出结果，Runner能确保当执行智能体需要时研究简报已准备就绪，且无需手动编写任何连接代码。可追溯性也是自然而然具备的：每一次智能体调用、工具使用及任务交接都会被记录为事件，从而解决了之前提到的调试难题。一旦出现问题，这些记录就能显示是哪个智能体做出了何种决定以及原因。

部署方式

有两种部署方法。Vertex AI Agent Engine提供托管式运行环境，几乎无需操作；而Cloud Run则让开发者能够完全掌控容器。

Agent Engine

这是维护成本较低的方案，适合大多数基于ADK的设置。只需设置两个变量即可：

PROJECT_ID=your-project-id
LOCATION_ID=us-central1

然后通过一条命令即可部署：

adk deploy agent_engine \
  --project=$PROJECT_ID \
  --region=$LOCATION_ID \
  --display_name="Customer Support Team" \
  support_agents

Agent Engine负责处理扩展性、会话持久化以及针对代理的追踪日志记录。无需编写Dockerfile，也不需要处理会话代码或相关基础设施。

Cloud Run

当需要自定义系统库、特定的网络功能（如VPC Service Controls）或非ADK组件时，可选择此方案。示例中使用了Quart而非Flask，两者语法几乎相同，但Quart能够原生支持async/await，从而避免了为每个请求都创建新的事件循环的asyncio.run()不良做法。

# server.py
from quart import Quart, request, jsonify
from main import handle_customer_message

app = Quart(__name__)

@app.route("/support", methods=["POST"])
async def support():
    data = await request.get_json()
    # No asyncio.run needed — Quart handles the event loop
    result = await handle_customer_message(
        data["customer_id"],
        data["message"]
    )
    return jsonify(result)

@app.route("/health", methods=["GET"])
async def health():
    return jsonify({"status": "healthy"})

if __name__ == "__main__":
    app.run(host="0.0.0.0", port=8080)

容器定义：

# Dockerfile
FROM python:3.11-slim

WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .
EXPOSE 8080

CMD ["python", "server.py"]

部署命令如下：

gcloud run deploy support-agents \
  --source . \
  --region us-central1 \
  --memory 4Gi \
  --cpu 2 \
  --allow-unauthenticated

简而言之，Cloud Run的设置难度中等（只需一个Dockerfile和一个API），需要通过Firestore或Redis进行手动状态管理，日志输出到标准的Cloud Logging中，并且支持自定义运行时和框架。而Agent Engine只需一条命令即可，内置会话功能，能生成特定于代理的追踪信息，适合需要快速扩展的ADK原生代理。

保持对话持久化

之前的协调器示例使用了InMemorySessionService，虽然适合测试，但在进程重启时会丢失所有数据。生产环境需要持久化的服务，ADK提供了两种此类服务。

使用Agent Engine时，会话管理由Vertex AI的会话服务自动处理：

# config.py
from google.adk.sessions import VertexAiSessionService

session_service = VertexAiSessionService(
    project="your-project-id",
    location="us-central1",
    agent_engine_id="your-agent-engine-id"  # From adk deploy output
)

Agent Engine还能汇总冗长的历史数据并自动清理，因此无需编写任何Firestore相关代码。

对于Cloud Run而言，DatabaseSessionService可与PostgreSQL或SQLite后端配合使用：

# config.py
from google.adk.sessions import DatabaseSessionService

# PostgreSQL for production
session_service = DatabaseSessionService(
    db_url="postgresql+asyncpg://user:pass@host:5432/agents"
)

# Or SQLite for development
session_service = DatabaseSessionService(
    db_url="sqlite+aiosqlite:///sessions.db"
)

它在首次启动时会创建相应的表格。将其放入Runner中后，会话状态就能在重启后依然保留。

状态如何在各个代理之间传递

无需手动调用get()和set()函数。每个代理的output_key会自动将数据写入会话状态，后续的代理则从该状态中读取数据：

triage_agent (output_key="triage_data")
    ↓ writes to session.state["triage_data"]
research_agent (reads triage_data, output_key="research_brief")
    ↓ writes to session.state["research_brief"]
action_agent (input_schema=ResearchBrief, output_key="action_result")
    ↓ writes to session.state["action_result"]
qa_agent (reads action_result + research_brief, output_key="qa_review")

Runner会将该完整状态持久化到所配置的任何后端，无论是Firestore、PostgreSQL还是Agent Engine。

三个习惯有助于保持内存健康：

• 总结信息。存储简化后的结果而非原始API响应，否则上下文窗口会因冗余数据而占满。
• 设置会话有效期。支持会话的生命周期应为24至48小时，这样既能保持数据库规模较小，又能提升安全性。
• 保护个人数据。身份证号或密码等原始敏感信息不应存在于共享状态中，因此在写入会话之前必须对其进行遮蔽处理。

何时单个智能体仍是更佳选择

多智能体系统并不一定更优越。一个实用的决策方法为：

• 简单的常见问题解答或状态查询：单智能体。延迟和成本最低，无需复杂推理。
• 通过三到五次查询进行数据丰富：单智能体。一次大语言模型调用即可满足基本工具使用需求。
• 受监管的工作流程：多智能体。通过隔离将操作逻辑与用户输入分开，从而降低提示注入风险。
• 模糊的路由处理：多智能体。专门的分类智能体负责意图匹配，减少约束条件之间的干扰。
• 金额超过500美元的高风险操作：多智能体。在调用任何API之前，可由质量检查智能体进行二次确认。
• 审计要求：多智能体。每个智能体的输入和输出都会被单独记录。

一个经验法则：当单个智能体的提示词长度接近3,000个标记，出现约束冲突和注意力分散现象，或者工作流程的不同环节需要不同的工具时，就该考虑将其拆分了。

常见陷阱

• 全能型智能体。当一个智能体需要掌握十种或更多工具时，其性能会下降，因为模型会把精力花在选择工具上，而非理解请求内容。应按功能拆分：需要处理退款、物流和法律事务的智能体应拆分为三个专门的智能体。
• 上下文过载。如果每个智能体都要查看全部十轮的对话内容，标记成本会急剧上升。应压缩状态信息，仅传递检测到的意图，而非整个对话内容。

无止境的任务传递。代理A将任务发送给B，B再传回给A，如此循环不断，白白消耗资源。应为协调器设置最大传递次数限制；通常经过两轮传递后就需要人工介入处理。

模糊的职责边界。两个代理可能同时负责同一任务，或者某个任务处于两者之间。应明确定义input_schema和output_schema规范，确保职责不重叠。

忽视延迟问题。顺序处理流程会增加4到6秒的延迟，而单个代理仅需1到2秒。ADK的ParallelAgent能够并行处理多项任务，同时进行研究和合规性检查可将总延迟降低30%到40%。

其预期效果

多代理系统在处理复杂任务时以牺牲部分延迟为代价来提升可靠性。现有的测试结果和行业经验都体现了这些优势：

• 复杂任务的处理精度。将多步骤问题拆解为具体的子任务通常比使用整体式提示语效果更好，ADK的文档也显示，当智能体承担的任务范围较窄时，完成率会更高。
• 故障隔离性。一旦某个组件出现故障，其影响仅限于该智能体本身，不会破坏整个回复的质量。
• 可审计性。每个智能体都有独立的日志，这使得调试和合规检查更加便捷。

不过其代价也同样明显：

• 延迟问题。使用五个智能体的完整流程大约需要3到8秒，而单个智能体仅需1到2秒。顺序调用是主要的瓶颈。

API使用成本。五名智能体（一名协调员和四名专家）的处理成本约为单名智能体每次查询成本的4到5倍，不过模型混合技术有助于控制这一数值。Flash-Lite版本的筛选服务每百万输入token收费0.25美元，而专为协调与质量检测设计的Pro版本则每百万输入token收费2美元，这使得每次交互的总成本保持在1美分以内。

思考token。Pro版本的高级思考模式（thinking_level="HIGH"）可增加约20K个token，其费用按每百万输出token12美元的标准计算。因此，仅思考环节的一次深度质量检测就要花费约0.24美元。建议谨慎使用高级模式，因为大多数质量检测任务使用中等模式即可。

复杂性。这种方案需要监控的环节更多。对于复杂的工作流程来说效益显著，但对于简单任务而言则显得过于繁琐。

以每月10万次复杂查询为基准，大致费用如下：

• Gemini 3.1 Pro上的协调器路由功能：约80美元
• Flash-Lite上的分诊功能：约20美元
• Flash上的研究功能：约40美元
• Flash上的执行功能：约40美元
• Pro版本中带有MEDIUM思维能力的质量检测功能：约100美元
• 配备2个vCPU和2 GiB内存的Cloud Run服务：约120美元
• PostgreSQL会话功能：约30美元
• 与单智能体设置共享的向量搜索功能：约100美元
• 每月总费用约为530美元

模型混合使用是控制成本的关键。只有大约30%的请求（协调器功能和质量检测功能）需要按Pro版本定价，其余请求则在Flash或Flash-Lite上运行，其每令牌成本仅为Pro版本的八分之一到五十分之一。每次复杂交互的单元成本大约为半美分。这些数据来自Google Cloud公布的定价信息，实际费用会因查询复杂度、思维能力等级以及智能体调用的深度而有所变化。

采用顺序

合理的流程是先从一个智能体开始，调整其逻辑，通过检索功能为其提供支持，只有当系统提示过长或任务种类过多以至于单个模型无法可靠处理时，才将其拆分为多个专用智能体。每一层都是建立在下一层的基础之上的。

智能体的微服务时代

多智能体设计并不会让人工智能变得更复杂，它能让复杂的workflow变得可预测，就如同微服务取代了单体架构：一个提示会被分解为多个各自承担特定任务的分布式组件。要求单个智能体同时完成分类、研究、执行和审核等工作，就如同让一个人同时担任接待员、首席分析师、技术专家和审计员。这种做法在单人公司或许可行，但在企业环境中则行不通。

为每个智能体分配明确的任务、专用工具以及共享记忆层，这样就能保证准确性——因为明确的提示能减少在不同任务间的偏差；同时便于调试——因为出问题的智能体一目了然；此外成本也较低——在 Gemini 3.1 上运行五个智能体的每次复杂交互仅需约半美分，远低于人类智能体一分钟的工作时间。

该设计还有进一步发展的空间。借助由 Linux 基金会管理的开放标准“智能体间通信（A2A）协议”，这样的系统未来甚至可以与供应商或合作伙伴运行的智能体进行交互，例如客服智能体可直接与物流公司的智能体沟通，无需人工介入即可解决包裹丢失的问题。

目前建议很简单：先构建一个智能体，当提示过长或任务过于多样时再逐步拆分。