OCT. 27, 2026 (英语).
想象一下，像给朋友发短信一样自然地订购你的周用品："嘿，我们没有牛奶和鸡蛋了。 周五电影之夜你还可以加点零食吗？".
我们分享了对话商业的建筑蓝图，提出三层脱钩设计(入侵、记忆和Cognition)
虽然蓝图建立了纸面架构，但生产准备需要工作软件。 在此文中，我们取出蓝图，用一个功能原型"快乐交付"(Happy Clivery)来使其活出。
使用LangGraph,Gemini,FastAPI，和Google Cloud，我们将走过代码，状态机器，确定性财务工具，以及一个自动验证套件，展示这种多agent 集群的实际操作方式。
为了建立可靠的商业集群，我们实施了三层脱钩，处理每条客户短信的行程：
让我们来检查每一层是如何用自己的专用架构在代码中运行的。
在设计对话商务时一个常见的问题是：客户在智能手机上发送的信息实际上如何降落在地层一中，而这个架构能否支持WhatsApp以外的消息平台？
终端用户不直接与您的 Google Cloud 服务器通信。 相反，客户信息在登陆您的 Webhook 之前会穿过操作员的生态系统：
每个主要的消费者留言平台都运行在完全相同的基本合同上：一个同步的HTTPS POST webhook送出，严格的HTTP确认超时：
-WhatsApp(Meta Cloud API)：需要快速的HTTP 200 OK确认，并使用HMAC-SHA256验证X-Hub-Signature-256头对准您的Meta App Secret. 如果您的服务器超时或返回非200状态，Meta 将发送标记为失败并启动指数回放( 最多7天)，这可以触发重复订单发送。
- Google RCS业务通讯(RBM)：通过Google Cloud Pub/Sub push或通过OAuth2/服务账户JWT无记名标志或有效载荷HMAC签名验证(X-Goog-Signature)认证的HTTPS网络呼号交付。 要求快速返回 HTTP 200 以避免 Webhook 发送重试。
- Apple 业务信息(AMB)：通过Apple 批准的Message Services(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):
{
"通道":"什么",#或"弧","apple 消息","sms"
"sender id":"+1xxxxxxxxxxxxxx",#已验证的E.164电话号码
"正文":"出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出出入".
"时间戳":1773428900
"raw metadata":{...} (中文(简体) ).
}
Python
复制
- 子- 150ms 同步握手：向 Google Cloud Pub/ Sub 发送犬形信息，并返回 HTTP 200 OK in < 150ms。
这保证了系统的其他部分——Session Store(Layer 2)和Agent 集群(Layer 3)——运行在干净的、正常的数据上，而不考虑shop从iPhone、Android RCS聊天或WhatsApp线索发送信息。
消息平台是完全无序的：每个短信都作为孤立的webhook事件而来。 为了提供连贯的购物经验，系统必须立即重建客户的全部上下文，而不必在每个回合重新阅读或重新分拆聊天日志的日子。
我们使用Cloud Memorystore(Redis)按客户已核实的电话号码(会话：{phone number})键来存储活动状态：
- 同步购物旅行(暂停和恢复)：在现实生活中，人们不会在一个座位上完成命令。 一个顾客可能会在早上9点发短信：“给我的推车加一加仑牛奶”，关闭WhatsApp来接火车，下午2: 00回信：“还要加面包和鸡蛋，我的总数是多少？” 由于会话状态被缓存于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实例在需要Redis Cluster水平硬化或读取复制品之前，可以舒适地处理每秒3000到15000个活式对话转折。
- 内存能力：一个活动会话有效载荷(profile + 活性推车 + 偏好)需要~2 KB到5 KB的序列化JSON. 一个谦虚的5GB Redis实例可以同时持有超过1,000,000个同时运行的购物车来存储。
Key Takeaway：在实际操作中，会话和Cart Laila永远不会成为瓶颈——它很容易超越上游的webhook推论和下游的LLM推论。
与其依靠单一的单音速来处理每一项任务，不如将推理分为专用的agents:
- 主管(Gemini Flash)：充当低纬度路由器，对客户的要求进行分类，并将其委托给合适的专家。
- 概况专家：获取个人喜好，并为新买家提供单一用途认证链接。
- 库存专家(Gemini Pro)：与仓库工具相接；查询库存、推荐配对物品并更改车型。
- 交易专家(Gemini Pro)：与支付工具挂钩；按确定性计算总额并开始收费。
现在，我们已经看到三层的高层次设计，让我们看看实际的Python原型代码，每层带来到生命。
这项服务的目标很简单：通过任何入内通道核实真实性，使有效载荷正常化为可控系统，并离开临界路径 < 150ms。
以下是我们的执行显示多条消息通道(WhatsApp, RCS, Apple Messages，短消息)如何由可插接的适配器处理：
导入 hashlib, hmac, json，时间
从打入中导入。
从 pydantic 导入基底模
类常见消息Event( BaseModel):
"""Channel-Annostic canonical 代表入境客户信息"".
频道：str #"Whatsapp","rcs","apple message","sms"
发件人 id: str # E.164 电话号码："+1xxxxxxxxxxx"
文本：str
时间戳：浮动
原始数据：dict[str, Any]
类：
"""Omni-Channel FastAPI入侵带有密码验证和同步排队"".
def init (自，meta 密：str, rcs 密：str="):
自我。 meta 秘密 = meta 秘密
自我。 rcs secret = rcs secret
self.publisher = CloudPubSubpublisher () 维基语录链接：名人名言 - 文学作品 - 谚语 - 谚语 - 谚语
def 校验 whatsapp signition( 自我，有效载荷字节：字节，sig header: str) - > bul:
""用常时比较法验证X-Hub-Signature-256头".
如果不是 sig header 或不是 sig header. starts with ("sha256="):
返回虚假
预期( hash) = hmac. new ()
自行.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 ("changers", [{}]) [0].get ("value", {})
消息 = 更改。 get ("messages", [{}]) [0]
返回常见消息Event ()
频道="什么app"
发送者 id=message.get (" from","")
text=message.get("text", {}).get("body", ""),
timerstamp=float(message.get ("timerstamp","timerstamp","time.time ()") 互联网档案馆的存檔，存档日期2014-12-21.
生 metadata}}“ message id”: message.get ("id" )}
)
elif频道：
# Google RCS 业务通讯结构
返回常见消息Event ()
频道="弧"
发件人 id=raw pailload.get("senderPhone Number",""")
文本=raw pailload.get("文本","")
timtamp=tim.time () 时间
生 metadata}}“ rcs message id ”：生负载。 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" 而非自已。 验证 whatsapp signation(活口-字节，签名):
返回 {"状态":"错误","消息":"无效的 HMAC 签名","http code": 403}
# 2. 规范化为可控计划
事件 = 自我正常化负载(通道、有效载荷)
如果不是事件或没有事件。 发件人( D):
返回{"状态":"错误","消息":"未识别的有效载荷格式","http code":400}
# 3. 从Cloud Pub/Sub队列上卸载的条形事件( < 150ms)
msg id = self.publisher.publish (专题="输入消息")，数据=event.model dump ().
返回{"状态":"接受","消息 id":msg id,"http code":200}
贾森
通过立即返回HTTP 200 OK，并将有效载荷交给Pub/Sub，我们完全解开下游AI推理时间中的信息摄入量，跨越所有信息生态系统。
- 部署入侵事务处(云跑或GKE):
# 提供云泡/子队列
gcloud pubsub 主题创建链接
# 将入侵的容器放入云跑
GCLUUD运行部署快乐的送货夫人
--image=gcr.io/我的-gcp-project/ress:latest=
- set-env-vars=APP SECRET="快乐机密 meta key 2026" (中文(简体) ).
...没有经过认证？
--端口为8080
纯文本
- Live runtime 校验(部署在GCP上xxxxxxx)：我们把入侵者容器直接部署到了Cloud Run并验证了活的HTTP端点：
- 服务名称：快乐的送货女招待
- Live Service URL: https://happy 交付-ingress-xxxxxxxx.us-central1.run.app.
项目/xxxxx/专题/内容
- 订阅附：项目/xxxxx/订阅/入围消息-sub
我们把现场的云跑HTTPS网游摄入量作为基准
# 基准测试：5个连续直播的HTTPS webhook站点到Cloud Run端点
运行 1：状态 200 in 266.84ms - > 消息编号：21184928529503172 (已装入 Pub/Sub)
运行 2：状态 200 in 301.73ms - > 消息ID: 21181303807117012(已装入 Pub/Sub)
运行 3：状态 200 in 300.47ms - > 消息ID: 21179945883786288 (被装入Pub/Sub)
运行 4：状态 200 in 295.14ms - > 消息ID: 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 Gem Box. All rights belong to the original author.