OCT. 27, 2026 (英语).
想象一下，像给朋友发短信一样自然地订购你的周用品："嘿，我们没有牛奶和鸡蛋了。 你能为周五电影之夜添加点心吗？"
我们分享了对话商业的建筑蓝图——提出一个三层脱钩的设计(入侵、记忆和Cognition)
虽然蓝图建立了纸面结构，但生产准备需要工作软件。 在本文中，我们拿这个蓝图，用一个叫做"快乐交付"的功能原型把它复活。
使用LangGraph,Gemini,FastAPI，以及Google Cloud，我们将走过代码，状态机器，确定性财务工具，以及一个自动验证套件，展示这种多agent 集群在实践中是如何运作的。
为了建立可靠的商业集群，我们实施了三层脱钩，处理每个客户短信的行程：
让我们来检查每个层是如何用自己的专用架构在代码中运行的。
在设计对话商务时一个常见的问题是：客户在智能手机上发送的信息实际上如何降落在Layle 1中，而这个架构能否支持WhatsApp以外的消息平台？
终端用户不直接与您的 Google 云服务器通信。 相反，客户信息在登陆您的 Webhook 之前会穿过操作器生态系统：
每个主要的消费者信息传递平台都运行在完全相同的基本合同上：一个同步的HTTPS POST webhook发送，并带有严格的HTTP确认超时：
-WhatsApp(Meta Cloud API)：需要快速的HTTP 200 OK确认，并验证X-Hub-Signature-256头，对你的Meta App Secret使用HMAC-SHA256. 如果您的服务器超时或返回非200状态，Meta 将发送标记为失败，并启动指数回落重试( 最多7天)，可以触发重复订单发送。
- Google RCS业务通讯(RBM)：通过Google Cloud Pub/Sub push或HTTPS webhooks通过OAuth2/服务账户JWT无记名标志或有效载荷HMAC签名验证(X-Goog-Signature)交付。 要求迅速返回 HTTP 200 以避免 Webhook 发送重试。
- Apple 业务信息(AMB)：通过Apple 批准的Message Services(MSP)提供商(MSP)交付。 通过提供商webhook签名，JSON网络签名(JWS)，或具有快速HTTP 200 OK确认期望的互通TLS(mTLS)认证。
- SMS / MMS(Twilio,Sinch,Infobip)：标准格式编码或JSON POST webhooks 与供应商特定的HMAC令牌(如X-Twilio-Signature)签名。
虽然各平台之间的有效载荷图和密码头略有不同，但第1层摘除所有客户端差异：
- 安全签名验证器：通过特定平台签名验证器(MetaHmacValidator,RcsJwtValidator,AppleJwsValidator,TwilioHmacValidator)发送请求。
- 有效载荷正常化适配器：将供应商特有的有效载荷转换成统一、通道不可知的罐装装置(CommonMessageEvent):
{
"频道":"Whatsapp",#或"rcs","apple messages","sms"
"sender id":"+1xxxxxxxxxxxxxxxxxxxxxx",#经验证的E.164电话号码
"文字":"出乳与鸡蛋"
"时间戳":1773428900
"raw metadata":{...} (中文(简体) ).
}
Python
复制
- 子- 150ms 同步握手：向 Google Cloud Pub/ Sub 发送犬科信息，并返回 HTTP 200 OK in < 150ms。
这保证了您的系统的其他部分—— Conference Store(Layer 2) 和 Agent 集群(Layer 3)—— 运行在干净的、正常化的数据上，而不考虑shopper是从 iPhone、 Android RCS 聊天或WhatsApp 线程发送信息。
消息平台是完全无国籍的：每个短信都作为孤立的webhook事件到达。 为了提供连贯的购物经验，系统必须立即重建客户的全部上下文，而不必在每个回合重新阅读或重新解析聊天日志的日子。
我们使用Cloud Memorystore(Redis)按客户已核实的电话号码(会话：{phone number})键来存储活动状态：
- 同步购物旅行(暂停和恢复)：在现实生活中，人们不会在一个座位上完成命令。 一个顾客可能会在早上9点发短信：“给我的车加一加仑牛奶”，关闭WhatsApp赶上火车，下午2点回信：“还要加面包和鸡蛋，我的总数是多少？” 由于会话状态在Redis中缓存，系统立即恢复他们的活动手推车([Millk])，并在不丢失节拍的情况下重新计算总数。
- 背景守护符(Dietary Perference & addresses)：如果用户的配置文件指定了膳食首选：["Nut-free"]，目录Agent在推荐电影夜宵时自动尊重这种首选——用户无需在每条短信中重新声明他们的过敏性。
- Safe,Tokenized checkout：客户简介持有象征性的支付参考文献(客户名："cust 9821"，付款名："Visa ending in 4242"). LLM只读过友好的面具标签，而交易工具将象征性的客户ID传递到Stripe. 原始信用卡号码从来不碰聊天提示或日志。
对话商业中一个常见的问题是，国家调查是否会引入暂时性。 以下是基线基础设施支持的内容：
- Raw Triput：一个单一的，基线 Google Cloud Memories for Redis exactory (Basic Tier, 1–5 GB memory) 很容易交付10,000至50,000+ QPS，分毫秒( < 1ms)读/写后缀。
- 对话式QPS：每个聊天回合需要2–3 Redis操作(会话获取，推车突变，状态承诺). 这意味着一个单一的标准Redis实例舒适地处理每秒3000到15000个活式对话转弯，然后需要Redis集群水平硬化或读取复制品。
- 内存能力：一个活动会话有效载荷(profile + 活动推车 + 偏好)需要~2 KB到5 KB的序列化JSON. 一个谦虚的5GB Redis实例可以同时持有超过1,000,000个同时运行的购物车作为内存。
Key Takeaway：在实践中，Session & Cart Layer永远不会成为瓶颈——它很容易超越上游webhook推论和下游LLM推论。
我们不依靠单一的提示来处理每一项任务，而是将推理分为集群，即专业agents:
- 主管(Gemini Flash)：作为低纬度路由器行事，对客户的要求进行分类，并将其交给合适的专家。
- 概况专家：获取个人喜好，并为新买家提供安全的单一用途认证链接。
- 库存专家(Gemini Pro)：以仓库工具为界；查询库存、推荐配对物品和改装车。
- 交易专家(Gemini Pro)：与支付工具挂钩；按确定性计算总额并开始收费。
现在，我们已经看到三层的高层次设计，让我们看看实际的Python原型代码，每层带来到生命。
这项服务的目标很简单：通过任何进入的通道核查真实性，使有效载荷正常化为舟体计划，并在 < 150ms中脱离关键路径。
以下是我们的执行显示多条消息通道(WhatsApp, RCS, Apple Messages，短消息)如何通过可插件适配器处理：
导入 hashlib, hmac, json，时间
从输入中导入 Any, Dict, option
从 pydantic 导入基模
类 Class CommunityMessageEvent( BaseModel)：常规
"""Channel-agnostic canonical 代表一个入境客户信息".
频道：str # "Whatsapp", "rcs", "apple messages", "sms"
发件人 id: str # E.164 电话号码："+1xxxxxxxxxxxxxxx"
文本：str
时间戳：浮动
原始数据：dict[str, Any]
类：
""Omni-Channel FastAPI入侵带有密码验证和同步排队".
def init (self, meta secret: str, rcs secret: str="): (中文(简体) ).
自我。 meta 秘密 = meta 秘密
自我。 rcs secret = rcs secret
自我.publisher = cloudPubSubpublisher () 维基语录链接：名人名言 - 分类
def 校验 whatsapp signition(自身，有效载荷字节：字节，sig header: str) - > bul:
""使用恒时比较法验证X-Hub-Signature-256头".
如果不是 sig header 或不是 sig header. starts with ("sha256="):
返回虚假
预期( hash) = hmac. new ()
self.meta secret.encode("utf-8")，有效载荷字节，hashlib.sha256
. hexdigest () (中文(简体) ).
提供 hash = sig header.split("sha256=") [1]
返回 hmac.compare digest(预期为 hash，提供为 hash)
def commonMessage Event:) - > 可选 [CommonMessage Event]:
"""将发件人和消息文字排入犬形图案".
如果频道=="whatsapp":
# Meta 图形 Webhook 结构
条目 = 原始负载.get ("输入", [{}]) [0]
更改 = 条目.get("更改",[{}]) [0].get("值",[{}])
message = change.get ("消息", [{}]) [0]
返回常见消息事件( E)
频道="什么app"
发件人 id=message.get(从，")
text=message.get("text", {}).get("body", ""),
timetamp=float (message.get ("timestamp", time.time ()) (英语).
原始的 metata {}“ message id”: message.get( “id”)}
)
Elif频道 == "rcs":
# Google RCS 业务通讯结构
返回常见消息事件( E)
频道=“弧”
发送器 id=raw payload.get("senderPhone Number","")
文本=raw pailload.get(“文本”、“”)
时间戳=时间。 time()
原始的 metata }“ rcs message id”：原始的 badload.get( “ messageId”) ) )
)
返回无
def 手柄 incoming webhook ()
本身，频道：str，有效载荷：dict[str, Any]，签名：str
- > 命令[str, Any]:
""Validates signment, common schema，和卸载到 Pub/Sub 以 < 150ms."".
有效载荷字节=json.dumps(载荷，排序键=True). encode("utf-8").
# 1. 平台专用安全验证
如果频道 ==="whatsapp" 而不是 self.very whatsapp signation(载荷字节，签名):
返回{“状态”:“错误”、“信息”:“无效的 HMAC 签名”、“http code”:403}
# 2. 规范化为犬科计划
事件 = 自我规范负载(通道、有效载荷)
如果不是事件或没有事件。 发件人 d:
返回{“状态”:“错误”、“信息”:“未识别的有效载荷格式”、“http code”:400}
# 3. 卸载划线事件到Cloud Pub/Sub队列( < 150ms)
msg id = self.publishher.publish(专题="输入消息")，数据=event.model dump ().
返回 {"状态": "接受", "message id": msg id, "http code": 200}
贾森
通过立即返回HTTP 200 OK，并将有效载荷交给Pub/Sub，我们完全解开下游AI推理时间的信息摄入，跨越所有信息生态系统。
- 部署入侵服务(云层运行或GKE):
# 提供云/子队列
gcloud pubsub 主题创建连接信息
{\fn方正粗倩简体\fs12\an8\1cHFFFF00\b0}把入侵者容器放进云中
Gcloud运行部署快乐的送货员
--image=gcr.io/my-gcp- project/ress:latest\) 互联网档案馆的存檔，存档日期2013-12-02.
- set-env-vars=APP SECRET="快乐机密 meta key 2026" \=
...没有经过认证？
--端口=8080
纯文本
- Live runtime 校验(部署在 GCP xxxxxxxx)：我们把入侵者容器直接部署到Cloud Run，并验证了HTTP的活端点：
- 服务名称：快乐的送货女郎
- Live Service URL: https://happydelivery-ingress-xxxxxx.us-central1.run.app
项目/xxxxx/专题/内容
- 订阅附加：项目/xxxxx/订阅/链接-sub
我们把现场的云跑HTTPS网游摄入量作为基准
# 基准测试：5个连续直播的HTTPS webhook帖子到云跑端点
运行 1：状态 200 in 266.84ms - > 消息ID: 21184928529503172 (装入 Pub/ Sub)
运行 2：状态 200 in 301.73ms - > 消息代号：21181303807117012(装入Pub/Sub)
运行 3：状态 200 in 300.47ms - > 消息ID: 2117994838786288 (装入 Pub/Sub)
运行 4：状态 200 in 295.14ms - > 消息编号：2182286983433170(装入Pub/Sub)
运行 5：状态 200 in 298.40ms - > 消息 ID: 21182275848770028(装入Pub/Sub)
# 直接的Pub/Sub 发布云运行内部的封面：~95ms
这个经验性地证明了我们的建筑主张：入侵服务验证了密码的签名，并立即投入前夕
本文由 AI图鉴自动抓取并翻译，版权归原作者所有；如需查看原文请点下方链接。
OCT. 27, 2026
Imagine ordering your weekly groceries as naturally as texting a friend:"Hey, we're out of milk and eggs. Can you also add some snacks for Friday movie night?"
In our previous post, Building Conversational Commerce with Multi-Agent Swarms on GCP, we shared the architectural blueprint for conversational commerce—proposing a 3-layer decoupled design (Ingress, Memory, and Cognition) to solve messaging timeouts, state tracking, and response reliability.
While blueprints establish the architecture on paper, production readiness demands working software. In this post, we take that blueprint and bring it to life with a functional prototype called HappyDelivery.
Using LangGraph, Gemini, FastAPI, and Google Cloud, we'll walk through the code, the state machine, deterministic financial tools, and an automated verification suite that demonstrates how this multi-agent swarm operates in practice.
To build a reliable commerce swarm, we implemented three decoupled layers that handle the journey of every customer text message:
Let's examine how each layer functions in code with its own dedicated architecture.
A frequent question when designing conversational commerce is: how does a message sent by a customer on their smartphone actually land in Layer 1, and can this architecture support messaging platforms beyond WhatsApp?
End users do not communicate directly with your Google Cloud servers. Instead, customer messages traverse an operator ecosystem before landing on your webhook:
Every major consumer messaging platform operates on the exact same fundamental contract: an asynchronous HTTPS POST webhook delivery with a strict HTTP acknowledgement timeout:
- WhatsApp (Meta Cloud API): Requires a prompt HTTP 200 OK acknowledgement and validating the X-Hub-Signature-256 header against your Meta App Secret using HMAC-SHA256. If your server times out or returns a non-200 status, Meta marks delivery as failed and initiates exponential backoff retries (for up to 7 days), which can trigger duplicate order deliveries.
- Google RCS Business Messaging (RBM): Delivered via Google Cloud Pub/Sub push or HTTPS webhooks authenticated via OAuth2 / Service Account JWT bearer tokens or payload HMAC signature verification (X-Goog-Signature). Requires returning HTTP 200 promptly to avoid webhook delivery retries.
- Apple Messages for Business (AMB): Delivered through an Apple-approved Messaging Service Provider (MSP). Authenticated via provider webhook signatures, JSON Web Signatures (JWS), or Mutual TLS (mTLS) with prompt HTTP 200 OK acknowledgement expectations.
- SMS / MMS (Twilio, Sinch, Infobip): Standard form-encoded or JSON POST webhooks signed with vendor-specific HMAC tokens (e.g., X-Twilio-Signature).
While the payload schemas and cryptographic headers differ slightly between platforms, Layer 1 abstracts away all client differences:
- Security & Signature Verifier: Routes incoming requests through platform-specific signature validators (MetaHmacValidator, RcsJwtValidator, AppleJwsValidator, TwilioHmacValidator).
- Payload Normalization Adapter: Ingests vendor-specific payloads and transforms them into a unified, channel-agnostic canonical 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
Copied
- Sub-150ms Asynchronous Handshake: Queues the canonical message onto Google Cloud Pub/Sub and returns HTTP 200 OK in < 150ms.
This ensures the rest of your system—Session Store (Layer 2) and Agent Swarm (Layer 3)—operates on clean, normalized data without caring whether the shopper sent their message from an iPhone, an Android RCS chat, or a WhatsApp thread.
Messaging platforms are completely stateless: every text message arrives as an isolated webhook event. To provide a coherent shopping experience, the system must instantly reconstruct the customer's full context without having to re-read or re-parse days of chat logs on every turn.
We use Cloud Memorystore (Redis) keyed by the customer's verified phone number (session:{phone_number}) to store active state:
- The Asynchronous Shopping Trip (Pause & Resume): In real life, people don't complete an order in one sitting. A customer might text at 9:00 AM: "Add a gallon of milk to my cart", close WhatsApp to catch a train, and reply at 2:00 PM: "Also add bread and eggs, and what's my total?" Because session state is cached in Redis, the system instantly restores their active cart ([Milk]) and recalculates totals without losing a beat.
- Contextual Guardrails (Dietary Preferences & Addresses): If a user's profile specifies dietary_preferences: ["Nut-Free"], the Inventory Agent automatically respects that preference when recommending movie night snacks—without the user having to re-declare their allergy in every text message.
- Safe, Tokenized Checkout: The customer profile holds tokenized payment references (customer_id: "cust_9821", payment_ref: "Visa ending in 4242"). The LLM only ever reads the friendly masked label, while the transaction tools pass the tokenized customer ID to Stripe. Raw credit card numbers never touch chat prompts or logs.
A common question in conversational commerce is whether state lookups will introduce latency. Here is what the baseline infrastructure supports:
- Raw Throughput: A single, baseline Google Cloud Memorystore for Redis instance (Basic Tier, 1–5 GB memory) easily delivers 10,000 to 50,000+ QPS with sub-millisecond (< 1ms) read/write latencies.
- Conversational QPS: Each chat turn requires 2–3 Redis operations (session fetch, cart mutation, state commit). This means a single standard Redis instance comfortably handles 3,000 to 15,000 active conversational turns per second before needing Redis Cluster horizontal sharding or read replicas.
- Memory Capacity: An active session payload (profile + active cart + preferences) takes ~2 KB to 5 KB of serialized JSON. A modest 5 GB Redis instance can hold over 1,000,000 concurrent active shopping carts in memory simultaneously.
Key Takeaway: In practice, the Session & Cart Layer will never be the bottleneck—it easily outpaces upstream webhook inference and downstream LLM inference.
Rather than relying on a single monolithic prompt to handle every task, we separate reasoning into a swarm of specialized agents:
- Supervisor (Gemini Flash): Acts as a low-latency router, classifying the customer's request and delegating it to the right specialist.
- Profile Specialist: Retrieves personal preferences and issues secure single-use auth links for new buyers.
- Inventory Specialist (Gemini Pro): Bound to warehouse tools; queries stock, recommends pairing items, and mutates the cart.
- Transaction Specialist (Gemini Pro): Bound to payment tools; calculates totals deterministically and initiates charges.
Now that we have seen the high-level design of the three layers, let's look at the actual Python prototype code that brings each layer to life.
The goal of this service is simple: verify authenticity across any inbound channel, normalize the payload into a canonical schema, and get off the critical path in < 150ms.
Here is our implementation showing how multiple messaging channels (WhatsApp, RCS, Apple Messages, SMS) are handled by pluggable adapters:
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
By immediately returning an HTTP 200 OK and handing the payload off to Pub/Sub, we completely decouple message intake from downstream AI reasoning time across all messaging ecosystems.
- Deploy the Ingress Service (Cloud Run or 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
Plain text
- Live Runtime Verification (Deployed on GCP xxxxxx): We deployed the Ingress container directly to Cloud Run and verified the live HTTP endpoint:
- Service Name: happydelivery-ingress
- Live Service URL: https://happydelivery-ingress-xxxxxx.us-central1.run.app
- Pub/Sub Topic Attached: projects/xxxxxx/topics/inbound-messages
- Subscription Attached: projects/xxxxxx/subscriptions/inbound-messages-sub
We benchmarked the live Cloud Run HTTPS webhook intake across warm invocations:
# 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
This empirically proves our architectural claim: the Ingress service validates the cryptographic signature and immediately commits the eve
Auto-fetched and translated by AI Atlas. All rights belong to the original author.