今天這篇文章的 GitHub：A2A Demo
在理解了代理對代理通訊協定（Agent-to-Agent Protocol，A2A）的基礎理論之後，今天我們要動手實作。我們將遵循 Google 官方的 A2A v1.0 協定規範，實際搭建一套具備即時修訂能力（Self-healing Revision Loop）的多代理跨社群發文流水線，會針對目前主流社群平台（X、Linkedin、Threads、Facebook）產生不同風格的文章，和一張對應的圖片。
本專案同時具備本地 A2A 微服務（方案 A）與 Google Cloud Agent Platform 雲端全託管（方案 B）雙軌架構，由四位分工明確的獨立人工智慧代理（Artificial Intelligence Agent，AI Agent）組成：
在真實的社群行銷工作中，一則高品質的宣傳專案往往需要企劃、文案、美編設計以及通路合規等多方協作。如果只使用單一提示詞（Prompt）要求單一模型同時搞定所有任務，往往會面臨以下問題：
透過 A2A 協定，我們將業務解耦為四大獨立微服務。當文案代理產出的 X 初稿超出字數限制時，審核代理會直接在通訊協定層發出 REVISION_REQUESTED 退件請求；文案代理收到後自動重寫、精簡壓縮並二次送審。整個團隊在背景完成自動校正與協商，呈報給人類的便是一套完全合規、圖文兼備的四大社群矩陣成果。
在 A2A 協定規範中，每個微服務都必須在 /.well-known/agent-card.json 公開端點宣告自身身分、技能清單與支援的介面規格：
# src/a2a_protocol.py
from pydantic import BaseModel
from typing import List
class AgentSkill(BaseModel):
id: str
name: str
description: str
class SupportedInterface(BaseModel):
url: str
protocolBinding: str = "JSONRPC"
protocolVersion: str = "2.0"
class AgentCard(BaseModel):
name: str
description: str
version: str = "1.0.0"
url: str
skills: List[AgentSkill] = []
supportedInterfaces: List[SupportedInterface] = []
文案代理負責針對四大社群平台輸出專屬風格，初稿刻意讓 X 貼文字數超過 270 字限制，用以演練通路稽核與自癒迴圈：
# src/agent_engines.py (片段)
class CopywriterAgentEngine:
def __init__(self, model_name: str = "gemini-3.8-flash"):
self.model_name = model_name
self.client = None
def set_up(self):
from google import genai
self.client = genai.Client(api_key=os.getenv("GEMINI_API_KEY"))
def query(self, action: str = "generate", topic: str = "", current_x: str = "", feedback: str = "") -> dict:
if self.client is None:
self.set_up()
if action == "generate":
prompt = f"""你是一位精通社群行銷的專業內容策略總監。請根據企劃主題「{topic}」，為 4 個社群平台量身打造專屬風格的繁體中文貼文：
1. X (Twitter)：明快有力的 Hook 開頭。初稿請詳盡論述（字數約 290 到 330 字，故意略超長以進行自癒修訂展示），文末附上標籤 #A2A #GoogleCloud #AI。
2. LinkedIn：B2B 商業趨勢與洞察。結構包含痛點背景、技術突破、商業效益與專業提問，留白舒適。
3. Threads (脆)：第一人稱生活化碎念（如「今天實測被驚艷到…」），短句斷行，真誠無距離感。
4. Facebook：故事敘事（Storytelling），運用表情符號條列重點，文末具備熱情明確的行動呼籲（CTA）。
請嚴格輸出合法 JSON 格式：{{"x": "...", "linkedin": "...", "threads": "...", "facebook": "..."}}"""
res = self.client.models.generate_content(
model=self.model_name,
contents=prompt,
config={"response_mime_type": "application/json"}
)
return {"status": "DRAFTED", "content": json.loads(res.text)}
elif action == "revise":
prompt = f"以下是超標的 X 貼文：\n{current_x}\n\n審核意見：{feedback}\n請精準壓縮在 250 字內，保留主題標籤 #A2A #GoogleCloud #AI。"
res = self.client.models.generate_content(model=self.model_name, contents=prompt)
return {"status": "REVISED", "revised_x": res.text.strip()}
視覺代理串接 gemini-3-pro-image，在接收到行銷主題後，自動規劃具備未來科技感的視覺提示詞並生成高畫質海報：
# src/agent_engines.py (片段)
class VisualAgentEngine:
def __init__(self, model_name: str = "gemini-3-pro-image"):
self.model_name = model_name
self.client = None
def query(self, topic: str = "") -> dict:
if self.client is None:
self.set_up()
prompt = f"A professional high-tech cinematic marketing poster for '{topic}', cybernetic aesthetics, sleek neon lines, 4k resolution, masterpiece."
from google.genai import types
res = self.client.models.generate_content(
model=self.model_name,
contents=prompt,
config=types.GenerateContentConfig(response_modalities=["IMAGE"])
)
for part in res.candidates[0].content.parts:
if getattr(part, "inline_data", None):
b64 = base64.b64encode(part.inline_data.data).decode("utf-8")
return {"status": "COMPLETED", "poster_url": f"data:image/jpeg;base64,{b64}"}
審核代理嚴格把關 X 通路的發文上限（270 字），當字數超標時直接在協定層退件：
class ReviewerAgentEngine:
def query(self, x: str = "") -> dict:
char_count = len(x)
if char_count > 270:
return {
"status": "REVISION_REQUESTED",
"feedback": f"貼文字數為 {char_count} 字，超出社群最佳限額（270 字），請濃縮精簡至 250 字內並保留主題標籤。",
"char_count": char_count
}
return {
"status": "APPROVED",
"feedback": "符合所有平台限制規範，審核通過！",
"char_count": char_count
}
總監代理在接收前端請求後，發動跨代理任務協調，並將原始協商封包以伺服器推送事件（Server-Sent Events，SSE）即時推播至前端：
# src/agents/director.py (方案 B: Google Cloud Agent Platform 調度核心)
@app.post("/api/start_pipeline")
async def start_pipeline(req: dict):
topic = req.get("topic", "")
async def run_pipeline():
# 1. 探索 Google Cloud Agent Platform 上的代理資源 (DISCOVER_CARD)
for name, eng_id in ENGINE_IDS.items():
await log_event("director", name, "DISCOVER_CARD", {
"platform": "Google Cloud Agent Platform",
"resource": eng_id
})
# 2. 並行派發文案生成與 Nano Banana Pro 產圖
copy_res = await asyncio.to_thread(copy_eng.query, action="generate", topic=topic)
copy_data = copy_res.get("content", {})
vis_res = await asyncio.to_thread(vis_eng.query, topic=topic)
poster_url = vis_res.get("poster_url", "")
# 3. 第一次送審 (Reviewer)
review_res = await asyncio.to_thread(rev_eng.query, x=copy_data.get("x", ""))
iteration_1_x = copy_data.get("x", "")
final_x = iteration_1_x
# 4. 退件自癒迴圈 (Self-healing Revision Loop)
if review_res.get("status") == "REVISION_REQUESTED":
await log_event("reviewer", "copywriter", "REVISION_REQUESTED", review_res)
# 文案代理自主壓縮
revise_res = await asyncio.to_thread(copy_eng.query, action="revise", current_x=iteration_1_x, feedback=review_res.get("feedback"))
final_x = revise_res.get("revised_x", "")
# 二次複審通過
second_review = await asyncio.to_thread(rev_eng.query, x=final_x)
await log_event("reviewer", "director", "APPROVED", second_review)
# 5. 推播四平台成果與海報
await log_event("director", "client", "PIPELINE_COMPLETED", {
"iteration_1_x": iteration_1_x,
"final_x": final_x,
"linkedin": copy_data.get("linkedin", ""),
"threads": copy_data.get("threads", ""),
"facebook": copy_data.get("facebook", ""),
"poster_url": poster_url
})
asyncio.create_task(run_pipeline())
return {"status": "started"}
專案提供智慧啟動腳本，自動偵測環境設定並啟動服務：
python run.py
執行後瀏覽器會自動開啟 http://localhost:8001，進入視覺化操作儀表板。
直接執行自動化部署腳本：
python deploy_agent_engine.py
腳本會自動建立 Google Cloud Storage 暫存貯體（Staging Bucket），並將 a2a-copywriter、a2a-visual、a2a-reviewer 部署為 Google Cloud Agent Platform 全託管實例（Agent Engine），最後將雲端資源識別碼自動寫入 .env，完成雲地混合調度。
在網頁儀表板上，輸入任何企劃大綱並啟動流水線後，可以完整觀測到以下流程：
透過 A2A 協定與 Google Cloud Agent Platform，我們成功把傳統依賴人工不斷微調的繁瑣流程，轉化為具備「自主分工、同儕審核、退件自癒」的高韌性多代理流水線，真正落實端到端自動化。
iThome鐵人賽
看影片追技術