Google 在 GCP 上用 LangGraph 与 Gemini 构建多智能体对话式商务原型…

Google 跟踪代码管理器 (noscript) 结束 Google 跟踪代码管理器 (noscript) HTML 在 GCP 上用多智能体集群构建对话式商务：一个可运行的原型 2026 年 10 月 27 日 Prateek Khushalani 软件工程师

引言：从蓝图到可运行的原型

想象一下，像给朋友发短信一样自然地订购每周的杂货：“嘿，家里牛奶和鸡蛋没了。你能再为周五电影之夜加点零食吗？”

在我们上一篇文章在 GCP 上用多智能体集群构建对话式商务中，我们分享了对话式商务的架构蓝图——提出了一个三层解耦设计（Ingress、Memory 和 Cognition），以解决消息超时、状态跟踪和响应可靠性问题。

蓝图在纸面上确立了架构，但要达到生产就绪，还需要可运行的软件。在本文中，我们将把这份蓝图付诸实践，用一个可运行的原型来呈现，它名为 HappyDelivery.

，使用 LangGraph, Gemini, FastAPI和 Google Cloud，我们将逐步讲解代码、状态机、确定性金融工具，以及一套自动化验证套件，展示这个多智能体集群在实践中如何运作。

1. 原型技术栈：3 层，3 张图

为了构建一个可靠的商务集群，我们实现了三个解耦层，它们负责处理每条客户短信的完整旅程：

让我们结合代码，逐一考察每一层如何凭借其专属架构发挥作用。

第 1 层：Ingress 层（部署在 GKE/Cloud Run 上的全渠道 Webhook 网关 + Cloud Pub/Sub）

在设计对话式商务时，一个常见的问题是： 客户在智能手机上发送的消息究竟是如何落到第 1 层的？这套架构能否支持 WhatsApp 之外的消息平台？

客户端设备如何连接到第 1 层

最终用户并不会直接与你的 Google Cloud 服务器通信。相反，客户消息会先经过运营商生态系统，然后才落到你的 webhook 上：

支持多种消息渠道：WhatsApp、RCS、Apple Messages 和 SMS

每个主流消费者消息平台都遵循完全相同的基本约定： 异步 HTTPS POST webhook 投递，并设有严格的 HTTP 确认超时:

WhatsApp（Meta Cloud API）: 要求及时返回 HTTP 200 OK 确认 并使用 HMAC-SHA256 对照你的 Meta App Secret 校验 X-Hub-Signature-256 请求头。如果你的服务器超时或返回非 200 状态，Meta 会将投递标记为失败，并启动指数退避重试（最长持续 7 天），这可能导致订单重复投递。
Google RCS 商业消息（RBM）: 通过 Google Cloud Pub/Sub 推送或 HTTPS webhook 投递，并通过 OAuth2 / 服务账号 JWT 持有者令牌或载荷 HMAC 签名验证进行认证（X-Goog-Signature）。需要及时返回 HTTP 200 ，以避免 webhook 投递重试。
Apple Messages for Business（AMB）: 通过 Apple 批准的 Messaging Service Provider（MSP）投递。通过提供商 webhook 签名、JSON Web 签名（JWS）或双向 TLS（mTLS）进行认证，并期望及时 HTTP 200 OK 确认。
短信 / 彩信（Twilio, Sinch, Infobip): 标准表单编码或 JSON POST webhook，使用供应商特定的 HMAC 令牌签名（例如， X-Twilio-Signature).

虽然各平台之间的载荷 schema 和加密标头略有差异， 第 1 层抽象掉了所有客户端差异:

安全与签名验证器： 通过平台特定的签名验证器路由传入请求（MetaHmacValidator, RcsJwtValidator, AppleJwsValidator, TwilioHmacValidator).
载荷规范化适配器： 接收供应商特定的载荷，并将其转换为统一的、与渠道无关的规范 schema（CommonMessageEvent):

{
       "channel": "whatsapp",  # or "rcs", "apple_messages", "sms"
       "sender_id": "+1xxxxxxxxxx",  # Verified E.164 phone number
       "text": "Out of milk & eggs",
       "timestamp": 1773428900,
       "raw_metadata": {...}
   }

Python 已复制

低于150毫秒的异步握手： 将规范消息排入 Google Cloud Pub/Sub 队列，并在 HTTP 200 OK 内返回 < 150ms.

这确保系统的其余部分——会话存储（第 2 层）和智能体集群（第 3 层）——在干净、规范化的数据上运行，而无需关心购物者是从 iPhone、Android RCS 聊天还是 WhatsApp 会话中发送的消息。

第 2 层：会话与购物车层（Cloud Memorystore Redis）

消息平台完全是无状态的：每条文本消息都以孤立的 webhook 事件形式到达。为提供连贯的购物体验，系统必须即时重建客户的完整上下文，而无需在每一轮对话中重新读取或重新解析数天的聊天记录。

我们使用 Cloud Memorystore（Redis） 以客户已验证的电话号码为键（session:{phone_number}）来存储活跃状态：

由该层驱动的真实客户旅程（CUJ）

异步购物之旅（暂停与恢复）： 现实生活中，人们不会一次性完成订单。客户可能在上午 9:00 发短信： “往我的购物车里加一加仑牛奶”，然后关闭 WhatsApp 去赶火车，下午 2:00 再回复： “再加面包和鸡蛋，我的总计是多少？” 由于会话状态缓存在 Redis 中，系统能即时恢复其活跃购物车（[Milk]）并重新计算总额，丝毫不卡顿。
上下文护栏（饮食偏好与地址）： 如果用户档案中指定了 dietary_preferences: ["Nut-Free"]，库存智能体在推荐电影之夜零食时会自动遵循该偏好——用户无需在每条短信中重复声明自己的过敏信息。
安全、令牌化结账： 客户档案保存令牌化支付引用（customer_id: "cust_9821", payment_ref: "Visa ending in 4242"）。LLM 只会读取友好的掩码标签，而交易工具则将令牌化客户 ID 传给 Stripe。原始信用卡号绝不会出现在聊天提示或日志中。

💡 规模与性能：基础基础设施能承受多少 QPS？

对话式商务中一个常见问题是：状态查询是否会引入延迟。以下是基线基础设施所支持的能力：

原始吞吐量： 单个基线 Google Cloud Memorystore for Redis 实例（基础层，1–5 GB 内存）可轻松提供 10,000 至 50,000+ QPS ，读写延迟低于 1 毫秒（< 1ms）。
对话 QPS： 每轮对话需要 2–3 次 Redis 操作（会话获取、购物车变更、状态提交）。这意味着单个标准 Redis 实例可轻松处理 每秒 3,000 至 15,000 个活跃对话轮次 ，之后才需要 Redis Cluster 水平分片或只读副本。
内存容量： 一个活跃会话负载（用户档案 + 活跃购物车 + 偏好设置）序列化为 JSON 后约占 2 KB 至 5 KB。一个适中的 5 GB Redis 实例可在内存中同时容纳 超过 1,000,000 个并发活跃购物车 。

关键要点： 在实践中，会话与购物车层 永远不会成为瓶颈——它能轻松超越上游的 webhook 推理和下游的 LLM 推理。

第 3 层：智能体集群层（使用 Gemini 的多智能体编排）

我们不再依赖单一的庞大提示来处理每项任务，而是将推理拆分为一群专业化的智能体：

监督者（Gemini Flash）： 充当低延迟路由器，对客户的请求进行分类，并将其委派给合适的专家。
用户档案专家： 检索个人偏好，并为新买家签发安全的一次性认证链接。
库存专家（Gemini Pro）： 绑定仓库工具；查询库存、推荐搭配商品，并修改购物车。
交易专家（Gemini Pro）： 绑定支付工具；确定性地计算总额并发起扣款。

2. 从架构到代码：构建三层结构

既然我们已经了解了这三层的高层设计，现在让我们看看让每一层真正运转起来的实际 Python 原型代码。

第 1 层原型：全渠道 FastAPI 入口与安全规范化

该服务的目标很简单： 跨任何入站渠道验证真实性，将负载规范化为统一的标准模式，并在 150 毫秒内脱离关键路径.

以下是我们展示多个消息渠道（WhatsApp、RCS、Apple Messages、SMS）如何由可插拔适配器处理的实现：

import hashlib, hmac, json, time
from typing import Any, Dict, Optional
from pydantic import BaseModel




class CommonMessageEvent(BaseModel):
  """Channel-agnostic canonical representation of an inbound customer message."""
  channel: str  # "whatsapp", "rcs", "apple_messages", "sms"
  sender_id: str  # E.164 phone number: "+1xxxxxxxxxx"
  text: str
  timestamp: float
  raw_metadata: Dict[str, Any]




class HappyDeliveryIngressService:
  """Omni-Channel FastAPI Ingress with cryptographic verification and async queueing."""


  def __init__(self, meta_secret: str, rcs_secret: str = ""):
    self.meta_secret = meta_secret
    self.rcs_secret = rcs_secret
    self.publisher = CloudPubSubPublisher()


  def verify_whatsapp_signature(self, payload_bytes: bytes, sig_header: str) -> bool:
    """Verifies X-Hub-Signature-256 header using constant-time comparison."""
    if not sig_header or not sig_header.startswith("sha256="):
      return False
    expected_hash = hmac.new(
        self.meta_secret.encode("utf-8"), payload_bytes, hashlib.sha256
    ).hexdigest()
    provided_hash = sig_header.split("sha256=")[1]
    return hmac.compare_digest(expected_hash, provided_hash)


  def normalize_payload(self, channel: str, raw_payload: Dict[str, Any]) -> Optional[CommonMessageEvent]:
    """Extracts sender and message text into a canonical schema."""
    if channel == "whatsapp":
      # Meta Graph Webhook structure
      entry = raw_payload.get("entry", [{}])[0]
      change = entry.get("changes", [{}])[0].get("value", {})
      message = change.get("messages", [{}])[0]
      return CommonMessageEvent(
          channel="whatsapp",
          sender_id=message.get("from", ""),
          text=message.get("text", {}).get("body", ""),
          timestamp=float(message.get("timestamp", time.time())),
          raw_metadata={"message_id": message.get("id")},
      )
    elif channel == "rcs":
      # Google RCS Business Messaging structure
      return CommonMessageEvent(
          channel="rcs",
          sender_id=raw_payload.get("senderPhoneNumber", ""),
          text=raw_payload.get("text", ""),
          timestamp=time.time(),
          raw_metadata={"rcs_message_id": raw_payload.get("messageId")},
      )
    return None


  def handle_incoming_webhook(
      self, channel: str, payload: Dict[str, Any], signature: str
  ) -> Dict[str, Any]:
    """Validates signature, normalizes schema, and offloads to Pub/Sub in < 150ms."""
    payload_bytes = json.dumps(payload, sort_keys=True).encode("utf-8")
    
    # 1. Platform-specific security validation
    if channel == "whatsapp" and not self.verify_whatsapp_signature(payload_bytes, signature):
      return {"status": "error", "message": "Invalid HMAC signature", "http_code": 403}


    # 2. Normalize to canonical schema
    event = self.normalize_payload(channel, payload)
    if not event or not event.sender_id:
      return {"status": "error", "message": "Unrecognized payload format", "http_code": 400}


    # 3. Offload canonical event onto Cloud Pub/Sub queue (< 150ms)
    msg_id = self.publisher.publish(topic="inbound-messages", data=event.model_dump())
    return {"status": "accepted", "message_id": msg_id, "http_code": 200}

JSON 已复制

通过立即返回 HTTP 200 OK 并将负载交给 Pub/Sub，我们在所有消息生态系统中将消息接收与下游 AI 推理时间完全解耦。

部署并运行第 1 层

部署入口服务（Cloud Run 或 GKE）：

# Provision Cloud Pub/Sub queue
   gcloud pubsub topics create inbound-messages


   # Deploy the Ingress container to Cloud Run
   gcloud run deploy happydelivery-ingress \
     --image=gcr.io/my-gcp-project/ingress:latest \
     --set-env-vars=APP_SECRET="happy_secret_meta_key_2026" \
     --allow-unauthenticated \
     --port=8080

纯文本已复制

实时运行时验证（部署在 GCP 上 xxxxxx): 我们已将 Ingress 容器直接部署到 Cloud Run ，并验证了实时 HTTP 端点：
- 服务名称： happydelivery-ingress
- 实时服务 URL： https://happydelivery-ingress-xxxxxx.us-central1.run.app
- 已附加的 Pub/Sub 主题： projects/xxxxxx/topics/inbound-messages
- 已附加的订阅： projects/xxxxxx/subscriptions/inbound-messages-sub

我们对实际运行的 Cloud Run HTTPS webhook 接收在热调用场景下进行了基准测试：

# Benchmark: 5 consecutive live HTTPS webhook posts to Cloud Run endpoint
   Run 1: Status 200 in 266.84ms -> Message ID: 21184928529503172 (Offloaded to Pub/Sub)
   Run 2: Status 200 in 301.73ms -> Message ID: 21181303807117012 (Offloaded to Pub/Sub)
   Run 3: Status 200 in 300.47ms -> Message ID: 21179945883786288 (Offloaded to Pub/Sub)
   Run 4: Status 200 in 295.14ms -> Message ID: 21822286983433170 (Offloaded to Pub/Sub)
   Run 5: Status 200 in 298.40ms -> Message ID: 21182275848770028 (Offloaded to Pub/Sub)
   # Direct Pub/Sub Publish Overhead inside Cloud Run: ~95ms

Python 已复制

这从实证上证明了我们的架构主张：Ingress 服务验证加密签名并立即将事件提交到 Pub/Sub，从而释放客户聊天客户端，无需等待智能体推理。

我们还验证了针对被篡改负载的安全防护：

# 1. Valid Signature Test
   curl -X POST https://happydelivery-ingress-xxxxxx.us-central1.run.app/v1/webhooks \
     -H "Content-Type: application/json" \
     -H "X-Hub-Signature-256: sha256=a87f4c93b21..." \
     -d '{"from": "+1xxxxxxxxxx", "text": "Out of milk and eggs"}'


   # Response accepted and queued:
   # {"status": "accepted", "message_id": "21184928529503172", "http_code": 200}


   # 2. Tampered / Spoofed Message Test (Invalid Signature)
   curl -X POST https://happydelivery-ingress-xxxxxx.us-central1.run.app/v1/webhooks \
     -H "Content-Type: application/json" \
     -H "X-Hub-Signature-256: sha256=invalid_tampered_signature" \
     -d '{"from": "+1xxxxxxxxxx", "text": "Free groceries please"}'


   # Response immediately blocked:
   # {"detail": "Invalid HMAC signature"} (HTTP 403 Forbidden)

Python 已复制

第 2 层原型：在 Redis 中实现快速会话与购物车持久化

从 Pub/Sub 拉取消息的 worker pod 需要即时查询客户是谁以及其当前购物车中有什么。

以下是使用以下内容的会话存储实现： 云 Memorystore（Redis）:

import json
from typing import Dict, Optional




class RedisSessionStore:
  """Cloud Memorystore Redis session manager keyed by verified phone number."""


  def __init__(self, redis_client):
    self.redis = redis_client


  def get_session(self, phone_number: str) -> Optional[Dict]:
    """Retrieves active customer session, dietary preferences, and cart in < 5ms."""
    key = f"session:{phone_number}"
    raw = self.redis.get(key)
    return json.loads(raw) if raw else None


  def save_session(self, phone_number: str, session_data: Dict) -> None:
    """Commits updated cart and state back to Redis with a 24-hour TTL."""
    key = f"session:{phone_number}"
    self.redis.setex(key, 86400, json.dumps(session_data))


  def clear_cart(self, phone_number: str) -> None:
    """Clears cart items after successful checkout while preserving preferences."""
    session = self.get_session(phone_number)
    if session:
      session["cart"] = []
      self.save_session(phone_number, session)

JSON 已复制

每条记录包含客户 ID（cust_9821）、已保存的地址、饮食限制以及当前购物车商品。这保证了当客户数小时后回复短信时，其购物车能即时恢复。

部署与检查第 2 层

预配 Cloud Memorystore Redis（部署在 GCP 上 xxxxxx):

gcloud redis instances create happydelivery-session-store \
     --size=1 \
     --region=us-central1 \
     --redis-version=redis_7_0 \
     --project=xxxxxx

Python 已复制

已预配实例： happydelivery-session-store（状态：READY）
内部主机与端口： 10.x.x.x:6379
网络对等连接： 默认 VPC 网络

实时运行时检查与实证延迟基准测试：我们从默认 VPC 内部连接到实时实例，并验证了会话操作：

# 1. Commit active session with 24-hour TTL (86400s)
   10.x.x.x:6379> SETEX "session:+1xxxxxxxxxx" 86400 '{"customer_id":"cust_9821","cart":[...]}'
   +OK


   # 2. Benchmark 5 consecutive session retrievals over VPC peering
   Run 1: 0.299ms
   Run 2: 0.276ms
   Run 3: 0.305ms
   Run 4: 0.210ms
   Run 5: 0.503ms
   # Average Session Retrieval Latency: 0.318ms


   # 3. Query active session payload and verify TTL
   10.x.x.x:6379> GET "session:+1xxxxxxxxxx"
   {
     "customer_id": "cust_9821",
     "dietary_preferences": ["nut-free", "gluten-free"],
     "cart": [
       {"item_id": "prod_milk", "name": "Whole Milk (1 gal)", "qty": 1, "price": 4.29},
       {"item_id": "prod_eggs", "name": "Free-Range Eggs (1 doz)", "qty": 1, "price": 3.99}
     ],
     "delivery_address": "xxx xxxxx St",
     "payment_method": "Visa ending in 4242"
   }


   10.x.x.x:6379> TTL "session:+1xxxxxxxxxx"
   (integer) 86400

JSON 已复制

每次会话查找的执行耗时均为 0.318ms, 这证明了我们的架构主张：恢复客户上下文和持久购物车状态为对话轮次增加的延迟几乎为零。

第 3 层原型：LangGraph 多智能体集群

在由 Ingress 和 Redis 处理传入流量与记忆的同时，所有决策逻辑都位于 LangGraph 多智能体图中。

第 1 步：共享智能体状态

每一轮对话都会经过一个类型化数据结构，其中保存客户档案、活动购物车以及主管的路由决策：

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph.message import add_messages




class CartItem(TypedDict):
  item_id: str
  name: str
  qty: int
  price: float




class AgentState(TypedDict):
  phone_number: str
  customer_id: str
  dietary_preferences: list[str]
  cart: list[CartItem]
  cart_total: float
  messages: Annotated[list[dict], add_messages]
  next_step: Literal["profile", "inventory", "transaction", "END"]

Python 已复制

第 2 步：编译集群图

接下来，我们定义节点和条件边，使主管能够动态地将任务路由到合适的专家：

from langgraph.graph import END, START, StateGraph




def build_happydelivery_swarm():
  builder = StateGraph(AgentState)


  # 1. Register specialized agent nodes
  builder.add_node("supervisor", supervisor_node)
  builder.add_node("profile", profile_node)
  builder.add_node("inventory", inventory_node)
  builder.add_node("transaction", transaction_node)


  # 2. Ingress always enters via the Supervisor
  builder.add_edge(START, "supervisor")


  # 3. Dynamic hand-offs based on intent
  builder.add_conditional_edges(
      "supervisor",
      lambda state: state.get("next_step", END),
      {
          "profile": "profile",
          "inventory": "inventory",
          "transaction": "transaction",
          "END": END,
      },
  )


  # 4. Return edges
  builder.add_edge("profile", "supervisor")
  builder.add_edge("inventory", END)
  builder.add_edge("transaction", END)


  return builder.compile()

Python 已复制

第 3 步：为智能体配备确定性工具

我们蓝图的核心原则是： 将算术运算委托给确定性工具。 生成式模型擅长语言和推理，而财务计算需要精确无误。如果模型以自由文本形式计算税费或折扣，偶尔可能会引入计算错误。

相反，我们将确定性工具绑定到我们的专家节点。例如， Inventory 和 Transaction 智能体使用 Python 计算工具，而不是在提示词中计算价格：

@tool
def calculate_cart_totals(cart_items: list[dict]) -> dict:
  """Calculates deterministic subtotal, tax, delivery fees, and grand total."""
  subtotal = sum(
      item.get("price", 0.0) * item.get("qty", 1) for item in cart_items
  )
  tax = round(subtotal * 0.0825, 2)  # 8.25% Sales tax
  delivery_fee = 0.00 if subtotal > 15.00 else 3.99
  total = round(subtotal + tax + delivery_fee, 2)
  return {
      "subtotal": round(subtotal, 2),
      "tax": tax,
      "delivery_fee": delivery_fee,
      "total": total,
  }

Python 已复制

第 4 步：运行并观察第 3 层执行

工作容器从 Pub/Sub 异步拉取消息，并实时流式传输 LangGraph 节点转换。

部署 Swarm Worker（Cloud Run Job 或 GKE）：

# Deploy as a scalable Cloud Run Job
   gcloud run jobs deploy happydelivery-swarm-worker \
     --source=./worker \
     --region=us-central1 \
     --project=xxxxxx \
     --set-env-vars=PROJECT_ID="xxxxxx",SUBSCRIPTION_ID="inbound-messages-sub"

Python 已复制

实时运行时执行追踪（GCP 执行输出）： 当工作进程拉取由 Ingress 服务发布的消息时，我们观察到 Supervisor 能够干净利落地路由到正确的专家智能体：

[Worker] Pulled message ID: 21182261471226730
   [Worker] Sender: +1xxxxxxxxxx, Text: 'Hey, we are out of milk and eggs. Also, add some snacks for Friday movie night.'
   [LangGraph] State loaded: phone=+1xxxxxxxxxx
   [LangGraph] Entering Node 'supervisor' -> Gemini Flash (Latency: 240ms)
   [LangGraph] Routing decision -> Intent: replenishment, Next Node: 'inventory'
   [LangGraph] Entering Node 'inventory' -> Gemini Pro
     ├── Tool call: search_catalog('milk') -> Whole Milk ($4.29)
     ├── Tool call: search_catalog('eggs') -> Free-Range Eggs ($3.99)
     ├── Tool call: calculate_cart_totals(items) -> Subtotal: $8.28, Tax: $0.68, Total: $12.95
     └── State Updated: cart=[Whole Milk, Free-Range Eggs], next_step='END'
   [LangGraph] Execution Complete -> Outbound response dispatched.
   [Worker] Successfully processed and acked in 0.03ms

Python 已复制

3. 将所有部分整合在一起：端到端流水线

现在我们已经具备了全部三个组件——Ingress（第 1 层）, Redis 记忆（第 2 层）, 以及 LangGraph Swarm（第 3 层）——它们实际上是如何作为一个单一的统一系统协同运行的？

以下是客户消息的完整端到端生命周期：

代码中的 Worker 流水线

协调全部三个层的 worker 循环简洁、清晰，并且将面向用户的 webhook 与后台 AI 推理完全解耦：

def process_pubsub_worker_turn(phone_number: str, user_text: str) -> str:
  """Coordinates Layer 1, Layer 2, and Layer 3 for each conversational turn."""
  # 1. Layer 2: Hydrate session and cart from Redis
  session = session_store.get_session(phone_number) or {}
  initial_cart = session.get("cart", [])


  # 2. Layer 3: Build the graph state
  state = {
      "messages": [HumanMessage(content=user_text)],
      "phone_number": phone_number,
      "customer_id": session.get("customer_id", ""),
      "cart": initial_cart,
      "dietary_preferences": session.get("dietary_preferences", []),
      "delivery_address": session.get(
          "delivery_address", "xxx xxxxx St"
      ),
      "payment_method": session.get(
          "payment_method", "Visa ending in 4242"
      ),
      "next_step": "",
  }


  # 3. Layer 3: Run the multi-agent swarm
  final_state = swarm.invoke(state)


  # 4. Layer 2: Commit updated cart back to Redis
  session["cart"] = final_state.get("cart", [])
  session_store.save_session(phone_number, session)


  # 5. Return outbound message back to WhatsApp messaging API
  return final_state["messages"][-1].content

Python 已复制

整个流水线已全部接通，接下来我们看看当实际客户订单流经它时会发生什么。

4. 运行原型：一次真实的 3 轮购物旅程

为了验证该原型，我们对一次端到端客户订单进行了自动化模拟。以下是智能体群在三个自然对话轮次中的表现：

第1轮：对话式补货与发现

^{👤 客户：
“嘿，我们的牛奶和鸡蛋用完了。另外，为周五电影之夜加点零食。”
🤖 HAPPYDELIVERY 智能体：
“明白了！我已添加1加仑全脂牛奶（$4.29）和1打散养鸡蛋（$3.99）。
根据您过去的电影之夜，需要我添加多力多滋（$4.50）和6罐装雪碧（$5.99）吗？”}

幕后： 主管识别出补货请求 → 移交给 ProfileAgent（加载偏好）→ InventoryAgent 查询目录库存，并将请求与相关零食配对。

第2轮：实时购物车变更与准确定价

^{👤 客户：
“把雪碧换成零度可乐。就这样。”
🤖 HAPPYDELIVERY 智能体：
“完成！加上零度可乐（$5.99），您的小计为 $18.77
（另加 $1.55 税费），总计 $20.32。
我将从您尾号为 4242 的 Visa 卡扣款，并送达
周四下午5点到7点之间，送到 xxx xxxxx 街。可以吗？"}

幕后： 这个 InventoryAgent 会修改内存中的购物车（移除雪碧并添加零度可乐），执行 calculate_cart_totals，并在不丢失上下文的情况下确认配送详情。

第3轮：结账与令牌化扣款执行

^{👤 客户：}"是的，完美！"
🤖 HAPPYDELIVERY 智能体：
“订单已确认！🧾 订单 #HD-9821 已下单。司机出发时我会发短信通知您。”

幕后： 主管检测到肯定的结账同意 → 路由至 TransactionAgent → 调用 execute_stripe_charge(customer_id='cust_9821', amount_cents=2032)。成功后清空 Redis 购物车。

验证：自动化测试套件

我们通过自动化单元测试和集成测试验证了所有核心行为——webhook HMAC 校验、目录索引、确定性数学计算以及多轮状态转换：

[ RUN      ] HappyDeliverySwarmTest.test_catalog_search                [ OK ]
[ RUN      ] HappyDeliverySwarmTest.test_deterministic_pricing        [ OK ]
[ RUN      ] HappyDeliverySwarmTest.test_full_three_turn_shopping_flow [ OK ]
[ RUN      ] HappyDeliverySwarmTest.test_hmac_webhook_signature_verification [ OK ]
[ RUN      ] HappyDeliverySwarmTest.test_stripe_charge_execution      [ OK ]
----------------------------------------------------------------------
Ran 5 tests in 0.158s - OK (0 arithmetic errors, 100% test pass rate)

Python 已复制

Token经济学：一次购物会话的实际成本是多少？

评估多智能体架构时的一个关键问题是 token消耗与运营推理成本. 在一种朴素实现中，庞大的提示词和完整的对话历史在每一步都被反复重新发送，LLM API账单会迅速侵蚀电商零售利润率。

我们的原型专门通过 严格的智能体专业化、确定性工具和外部化Redis状态:

来防止token浪费。状态卸载到Redis： 我们不会在LLM上下文中维护无穷无尽的原始聊天记录。只传入当前用户意图和活跃购物车摘要，即使购物车不断增长，输入token数也保持平稳。
确定性代码承担繁重工作： 数学运算（税费、运费、折扣）、目录关键词匹配和Stripe扣款均由Python函数执行（0个LLM token）。
模型分层（Flash与Pro）： 高频路由运行在轻量、极速的 Gemini Flash上，将更深层的推理能力保留给 Gemini Pro 严格用于复杂库存推荐和支付最终确认。

实证令牌与成本明细（3轮购物旅程）

以下是实时3轮 HappyDelivery 会话中测得的精确令牌遥测数据：

定价示例基于标准 Gemini API 费率计算：Gemini Flash 为每100万输入令牌 $0.15、每100万输出令牌 $0.60；Gemini Pro 为每100万输入令牌 $1.25、每100万输出令牌 $10.00。

对商业利润率的关键启示

每单不足一美分： 一次完整的3轮多智能体购物会话消耗 约4,240个总令牌，成本仅为 $0.0061（远低于一美分）.
占购物篮价值的比例可忽略不计： 在一笔$20.32的杂货订单中，$0.0061仅占商品交易总额（GMV）的 0.030% ——与标准支付处理费（2.9% + $0.30）相比实际上可忽略不计。
上下文缓存机会： 对于拥有大型静态目录架构或固定智能体指令的生产级企业部署，启用Google Cloud的 Gemini上下文缓存 可将重复输入token成本再降低 75%，使每单成本更接近 $0.0033.

5. 构建原型得到的实践经验

将蓝图转化为可运行代码后，暴露出若干现实情况：

克服“群体税”（延迟）： 跨多次LLM调用的多智能体推理可能耗时2–4秒。为避免客户感觉聊天已无响应，worker会先通过消息API发送即时进度通知（“正在为您查看新鲜货架……🛒”），然后再流式传输最终结果。
将数学计算保留在确定性工具中： 即便是能力出众的模型，也可能偶尔在边缘情况下算错。将定价逻辑放在确定性函数中，可确保100%的财务准确性。
专精优于单体： 为各个智能体分别提供小而聚焦的系统指令（Profile, Inventory, Transaction, ），其行为可靠性远高于把所有职责都塞进一个庞大的提示词中。

6. 结论与后续展望

构建这个原型证明，我们蓝图中概述的解耦架构并非只是理论——它是切实可行、可靠，并且已准备好投入生产的，可部署在 Google Cloud.

通过将用于多智能体工作流的 LangGraph 、用于自然语言推理的 Gemini ，以及用于快速入口和持久记忆的 GCP 相结合，我们可以打造出如同与朋友聊天般简单的购物体验，同时保持商业所需的企业级严谨性。

发布于：

上一篇下一篇相关文章 Bringing Visual Graph Debugging to Custom Edge Compilers with Model Explorer

移动端 AI 案例研究学习

借助 Model Explorer 为自定义 Edge 编译器带来可视化图调试

2026年10月20日 Fast, reliable A2UI, with swappable inference formats

移动端 Web 教程操作指南

快速、可靠的 A2UI，支持可替换的推理格式

2026年10月27日 Step by Step guide: Build a Voice Agent That Never Goes Silent using ADK

AI 操作指南最佳实践

分步指南：使用 ADK 构建一个永不沉默的语音智能体

2026年10月19日

主题

标签

正文 · AI 翻译

引言：从蓝图到可运行的原型

1. 原型技术栈：3 层，3 张图

第 1 层：Ingress 层（部署在 GKE/Cloud Run 上的全渠道 Webhook 网关 + Cloud Pub/Sub）

客户端设备如何连接到第 1 层

支持多种消息渠道：WhatsApp、RCS、Apple Messages 和 SMS

第 2 层：会话与购物车层（Cloud Memorystore Redis）

由该层驱动的真实客户旅程（CUJ）

💡 规模与性能：基础基础设施能承受多少 QPS？

第 3 层：智能体集群层（使用 Gemini 的多智能体编排）

2. 从架构到代码：构建三层结构

第 1 层原型：全渠道 FastAPI 入口与安全规范化

部署并运行第 1 层

第 2 层原型：在 Redis 中实现快速会话与购物车持久化

部署与检查第 2 层

第 3 层原型：LangGraph 多智能体集群

第 1 步：共享智能体状态

第 2 步：编译集群图

第 3 步：为智能体配备确定性工具

第 4 步：运行并观察第 3 层执行

3. 将所有部分整合在一起：端到端流水线

代码中的 Worker 流水线

4. 运行原型：一次真实的 3 轮购物旅程

第1轮：对话式补货与发现

第2轮：实时购物车变更与准确定价

第3轮：结账与令牌化扣款执行

验证：自动化测试套件

Token经济学：一次购物会话的实际成本是多少？

实证令牌与成本明细（3轮购物旅程）

对商业利润率的关键启示

5. 构建原型得到的实践经验

6. 结论与后续展望