用 AI Bot 打造顧問團隊（五）：數位行銷公司實戰案例

1mkdir pixelflow-ai-team && cd pixelflow-ai-team
2
3# 安裝依賴
4pip install langgraph langchain-anthropic langchain-core \
5            pandas openpyxl jinja2 python-pptx httpx
6
7# 目錄結構
8mkdir -p {brands,outputs/{content,reports,slides},prompts,tools}

 1{
 2  "client_id": "client_A",
 3  "brand_name": "綠野良田",
 4  "industry": "有機食品",
 5  "target_audience": {
 6    "primary": "30-45歲，注重健康的都市媽媽",
 7    "secondary": "25-35歲，環保意識強的年輕人"
 8  },
 9  "brand_voice": {
10    "tone": "溫暖、真實、專業但不冷漠",
11    "avoid": ["誇大功效詞彙（例如：神奇、奇蹟）", "過度推銷感的語氣", "英文夾雜"],
12    "keywords": ["自然", "無農藥", "產地直送", "用心", "土地"]
13  },
14  "platforms": ["Facebook", "Instagram", "LINE"],
15  "posting_schedule": {
16    "facebook": 3,
17    "instagram": 5,
18    "line": 2
19  },
20  "content_pillars": [
21    "產品介紹（30%）",
22    "農場故事（25%）",
23    "食譜/料理靈感（25%）",
24    "健康知識（15%）",
25    "促銷活動（5%）"
26  ],
27  "hashtag_strategy": {
28    "brand_tags": ["#綠野良田", "#有機生活"],
29    "content_tags": ["#有機食品", "#產地直送", "#無農藥"],
30    "trend_tags": "每週更新，由 Content Agent 建議"
31  },
32  "past_top_posts": [
33    {
34      "platform": "Instagram",
35      "content_type": "農場故事",
36      "engagement_rate": 0.082,
37      "key_elements": ["農夫主角", "清晨光線照片", "150字以內文案"]
38    }
39  ],
40  "forbidden_topics": ["競爭品牌比較", "政治議題", "未經驗證的健康聲稱"]
41}

 1# state.py
 2from typing import TypedDict, Annotated, List, Optional
 3from langgraph.graph.message import add_messages
 4
 5class MarketingState(TypedDict):
 6    # 基本資訊
 7    client_id: str
 8    task_type: str           # "content_batch" | "monthly_report" | "ad_copy" | "strategy"
 9    period: str              # "2026-04"
10
11    # 品牌資料
12    brand_dna: dict
13
14    # 各 Agent 產出
15    content_plan: dict       # 本月內容規劃
16    post_drafts: List[dict]  # 社群貼文草稿清單
17    ad_copies: List[dict]    # 廣告文案清單
18    analytics_data: dict     # 廣告數據
19    insights: dict           # 數據洞察
20    report: str              # 月報全文
21    slide_content: dict      # 簡報結構
22
23    # 控制流
24    current_agent: str
25    approved_posts: List[str]   # 人工審核通過的貼文 ID
26    error: Optional[str]
27    messages: Annotated[list, add_messages]

 1# agents/brand_agent.py
 2import json
 3from pathlib import Path
 4from langchain_anthropic import ChatAnthropic
 5from langchain_core.messages import SystemMessage, HumanMessage
 6
 7llm = ChatAnthropic(model="claude-sonnet-4-6", temperature=0.2)
 8
 9BRAND_AGENT_PROMPT = """你是品牌守門員，負責在所有行銷工作開始前，
10確保整個 AI 團隊完全理解這個品牌的核心精神。
11
12你的任務：
131. 讀取品牌 DNA 檔案
142. 產出給其他 Agent 使用的「品牌簡報」
153. 列出 3 個最近成效好的內容類型（供 Content Agent 參考）
164. 提醒本月應該避免的敏感話題
17
18品牌簡報格式（Markdown）：
19## 品牌核心
20## 目標受眾特徵（3 個最具體的描述）
21## 這個月的內容重點（根據季節/節慶/品牌行事曆）
22## 語氣規範（5 個 DO / 5 個 DON'T）
23## 本月禁止話題
24"""
25
26def brand_agent(state: MarketingState) -> MarketingState:
27    brand_path = Path(f"brands/{state['client_id']}.json")
28    brand_dna = json.loads(brand_path.read_text(encoding="utf-8"))
29
30    response = llm.invoke([
31        SystemMessage(content=BRAND_AGENT_PROMPT),
32        HumanMessage(content=f"請為以下品牌產出品牌簡報：\n{json.dumps(brand_dna, ensure_ascii=False, indent=2)}")
33    ])
34
35    return {
36        "brand_dna": brand_dna,
37        "current_agent": "content_and_ad",
38        "messages": [response]
39    }

 1# agents/content_agent.py
 2CONTENT_AGENT_PROMPT = """你是資深社群內容策略師，擅長為品牌規劃有溫度、
 3有轉換率的社群內容。
 4
 5## 你的任務
 6根據品牌 DNA 和本月行事曆，規劃一個月的社群內容，並產出草稿。
 7
 8## 內容規劃原則
 91. 遵守品牌 DNA 中的內容比例（Content Pillars）
102. 每篇貼文要有明確目的：曝光、互動、轉換 選其一
113. 平台差異化：Instagram 視覺導向（短文案）、Facebook 故事導向（長文案）、LINE 直接實用
12
13## 每篇貼文草稿格式
14```json
15{
16  "post_id": "POST-001",
17  "platform": "Instagram",
18  "content_pillar": "農場故事",
19  "goal": "engagement",
20  "post_date": "2026-04-05",
21  "copy": "貼文文案內容",
22  "visual_direction": "建議搭配的視覺方向（給設計師看）",
23  "hashtags": ["#tag1", "#tag2"],
24  "cta": "行動呼籲",
25  "estimated_engagement": "high|medium|low"
26}

 1# graph.py
 2from langgraph.graph import StateGraph, END
 3from langgraph.checkpoint.memory import MemorySaver
 4from state import MarketingState
 5from agents.brand_agent import brand_agent
 6from agents.content_agent import content_agent
 7from agents.ad_copy_agent import ad_copy_agent
 8from agents.analyst_agent import analyst_agent
 9from agents.report_agent import report_agent
10
11
12def parallel_content_and_ads(state: MarketingState):
13    """並行執行 Content Agent 和 Ad Copy Agent"""
14    # LangGraph 支援 Send API 實現並行
15    from langgraph.constants import Send
16    return [
17        Send("content_agent", state),
18        Send("ad_copy_agent", state)
19    ]
20
21
22def route_by_task(state: MarketingState) -> str:
23    """根據任務類型決定流程"""
24    task = state.get("task_type", "monthly_report")
25    if task == "content_batch":
26        return "content_only"
27    elif task == "ad_copy":
28        return "ad_only"
29    return "full_pipeline"   # monthly_report 跑完整流程
30
31
32def build_marketing_graph():
33    graph = StateGraph(MarketingState)
34
35    # 加入節點
36    graph.add_node("brand_agent", brand_agent)
37    graph.add_node("content_agent", content_agent)
38    graph.add_node("ad_copy_agent", ad_copy_agent)
39    graph.add_node("analyst_agent", analyst_agent)
40    graph.add_node("report_agent", report_agent)
41
42    # 入口：Brand Agent 先執行
43    graph.set_entry_point("brand_agent")
44
45    # Brand → 根據任務類型分流
46    graph.add_conditional_edges(
47        "brand_agent",
48        route_by_task,
49        {
50            "full_pipeline": "analyst_agent",      # 月報：先跑分析
51            "content_only": "content_agent",
52            "ad_only": "ad_copy_agent"
53        }
54    )
55
56    # 月報完整流程
57    graph.add_edge("analyst_agent", "content_agent")
58    graph.add_edge("content_agent", "ad_copy_agent")
59    graph.add_edge("ad_copy_agent", "report_agent")
60    graph.add_edge("report_agent", END)
61
62    # 單獨任務直接結束
63    graph.add_edge("content_agent", END)   # content_only 路徑（條件判斷）
64
65    memory = MemorySaver()
66    return graph.compile(checkpointer=memory)

 1# main.py
 2import asyncio
 3from datetime import datetime
 4from graph import build_marketing_graph
 5from langchain_core.messages import HumanMessage
 6
 7app = build_marketing_graph()
 8
 9
10async def run_monthly_report(client_id: str, period: str):
11    """執行月報流程"""
12    print(f"[{datetime.now()}] 開始產出 {client_id} 的 {period} 月報...")
13
14    config = {"configurable": {"thread_id": f"{client_id}-{period}"}}
15    result = await app.ainvoke(
16        {
17            "client_id": client_id,
18            "task_type": "monthly_report",
19            "period": period,
20            "messages": [HumanMessage(content=f"開始 {period} 月報產出流程")]
21        },
22        config=config
23    )
24
25    # 儲存結果
26    report_path = f"outputs/reports/{client_id}_{period}_report.md"
27    with open(report_path, "w", encoding="utf-8") as f:
28        f.write(result["report"])
29
30    print(f"[完成] 月報已儲存至 {report_path}")
31    return result
32
33
34async def run_content_batch(client_id: str, period: str):
35    """只產出社群內容草稿"""
36    config = {"configurable": {"thread_id": f"{client_id}-{period}-content"}}
37    result = await app.ainvoke(
38        {
39            "client_id": client_id,
40            "task_type": "content_batch",
41            "period": period,
42            "messages": [HumanMessage(content="開始內容規劃")]
43        },
44        config=config
45    )
46    return result["post_drafts"]
47
48
49# 批次執行所有客戶的月報
50async def run_all_monthly_reports(period: str):
51    clients = ["client_A", "client_B", "client_C"]  # 從資料庫讀取
52
53    # 並行執行所有客戶
54    tasks = [run_monthly_report(c, period) for c in clients]
55    results = await asyncio.gather(*tasks, return_exceptions=True)
56
57    for client, result in zip(clients, results):
58        if isinstance(result, Exception):
59            print(f"[錯誤] {client}: {result}")
60        else:
61            print(f"[成功] {client} 月報完成，品質分數待審核")
62
63
64if __name__ == "__main__":
65    import sys
66    period = sys.argv[1] if len(sys.argv) > 1 else "2026-04"
67    asyncio.run(run_all_monthly_reports(period))

1# 每月 1 日凌晨 2 點自動產出上個月月報
20 2 1 * * cd /path/to/pixelflow-ai-team && python main.py $(date -v-1m +%Y-%m) >> logs/cron.log 2>&1
3
4# 每週一早上 8 點產出本週內容草稿
50 8 * * 1 cd /path/to/pixelflow-ai-team && python -c "
6import asyncio
7from main import run_content_batch
8asyncio.run(run_content_batch('client_A', '2026-W18'))
9" >> logs/cron.log 2>&1

 1# review_ui.py
 2import streamlit as st
 3import json
 4from pathlib import Path
 5
 6st.title("PixelFlow AI 內容審核台")
 7
 8# 載入待審核的貼文草稿
 9draft_files = list(Path("outputs/content").glob("*.json"))
10selected_file = st.selectbox("選擇客戶草稿", draft_files)
11
12if selected_file:
13    drafts = json.loads(Path(selected_file).read_text())
14
15    st.subheader(f"共 {len(drafts)} 篇草稿待審核")
16
17    approved = []
18    for draft in drafts:
19        with st.expander(f"[{draft['platform']}] {draft['post_date']} - {draft['content_pillar']}"):
20            st.write("**文案：**", draft["copy"])
21            st.write("**視覺方向：**", draft["visual_direction"])
22            st.write("**Hashtags：**", " ".join(draft["hashtags"]))
23            st.write("**CTA：**", draft["cta"])
24
25            col1, col2, col3 = st.columns(3)
26            if col1.button("✅ 核可", key=f"approve_{draft['post_id']}"):
27                approved.append(draft["post_id"])
28            if col2.button("✏️ 修改", key=f"edit_{draft['post_id']}"):
29                new_copy = st.text_area("修改文案", draft["copy"])
30                draft["copy"] = new_copy
31            if col3.button("❌ 退回", key=f"reject_{draft['post_id']}"):
32                st.warning("此貼文已退回，請說明原因（供 AI 學習）")
33
34    if st.button("送出審核結果"):
35        st.success(f"已核可 {len(approved)}/{len(drafts)} 篇，已排程發佈")

 1# 在 Content Agent 輸出後加驗證
 2def validate_post(post: dict) -> dict:
 3    platform_limits = {
 4        "Instagram": {"copy_max": 2200, "hashtag_max": 30},
 5        "Facebook": {"copy_max": 63206, "hashtag_max": 10},
 6        "LINE": {"copy_max": 500, "hashtag_max": 0}
 7    }
 8    platform = post["platform"]
 9    limits = platform_limits.get(platform, {})
10
11    errors = []
12    if len(post["copy"]) > limits.get("copy_max", 9999):
13        errors.append(f"文案超過字數限制（{len(post['copy'])} > {limits['copy_max']}）")
14
15    hashtag_count = len(post.get("hashtags", []))
16    if hashtag_count > limits.get("hashtag_max", 99):
17        errors.append(f"Hashtag 過多（{hashtag_count} > {limits['hashtag_max']}）")
18
19    post["validation_errors"] = errors
20    post["is_valid"] = len(errors) == 0
21    return post

1# analyst_agent.py 中加入防護
2ANALYST_PROMPT += """
3## 嚴格規定
4- 所有數字必須來自傳入的廣告數據，不得自行假設或發明
5- 如果某個指標沒有數據，明確寫「本月無資料」
6- 計算 ROAS 等衍生指標時，用以下公式：ROAS = 營收 / 花費
7- 不確定的部分寫「待確認」，不要猜測
8"""

 1## 2026年4月 行銷月報 - 綠野良田
 2
 3### 本月亮點
 4本月廣告整體表現優異，Meta Ads 的 ROAS 達到 3.8x，超越目標 27%。
 5農場故事系列的貼文互動率達到 8.2%，是一般貼文的 2.3 倍。
 6建議下個月加重農場故事的內容比例，並測試影片格式。
 7
 8### 廣告成效
 9
10| 平台 | 曝光 | 點擊 | ROAS | 花費(TWD) |
11|------|------|------|------|---------|
12| Meta Ads | 125,000 ▲12% | 1,890 ▲8% | 3.8x ▲0.3 | $28,000 |
13| Google Ads | 45,000 ▲5% | 2,100 ▲15% | 4.2x ▲0.5 | $18,000 |
14
15### 下個月建議
161. 🎯 增加農場故事影片（預計提升互動率 30-50%）
172. 💰 Meta Ads 預算提高 20%（ROAS 穩定，值得加碼）
183. 📝 新增「食譜教學」系列，搭配母親節活動