【實戰】Day 25：Mode B：用 LangGraph 打造三角辯論 Agent

ERP-AI-Agent/backend/
├── app/
│   ├── agents/
│   │   ├── order_agent.py       # Mode A（Day 24）
│   │   └── debate_graph.py      # Mode B（Day 25，新增）← NEW
│   ├── routers/
│   │   ├── orders.py
│   │   ├── agent.py             # /agent/query
│   │   └── debate.py            # /agent/debate ← NEW
│   ├── tools/
│   │   └── order_tools.py       # 新增 BRAND_COST_FACTOR ← UPDATED
│   ├── context/
│   │   └── user_context.py
│   └── main.py                  # 加入 debate router ← UPDATED
├── data/
│   └── *.json
└── requirements.txt             # 加入 langgraph ← UPDATED

CrewAI Hierarchical：
  Manager → (黑盒分配) → Workers → Manager 彙整
  條件分支？放在 prompt 裡，不保證執行

LangGraph StateGraph：
  節點 A → 節點 B → 條件路由函數 → 節點 C 或節點 D
  條件分支是程式碼，不是 prompt

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class DebateState(TypedDict):
    question: str
    sales_view: str      # 業務觀點（節點 A 寫入）
    production_view: str # 生管觀點（節點 B 寫入）
    finance_view: str    # 財務觀點（節點 C 寫入）
    finance_veto: bool   # 否決旗標（節點 C 寫入，路由函數讀取）
    final_decision: str  # 總裁決（節點 D 寫入）

async def sales_node(state: DebateState) -> dict:
    result = await sales_agent.run(state["question"])
    return {"sales_view": result.data}  # 只回傳要更新的欄位

def route_after_finance(state: DebateState) -> Literal["veto_path", "normal_path"]:
    return "veto_path" if state.get("finance_veto") else "normal_path"

builder.add_conditional_edges(
    "finance",           # 來源節點
    route_after_finance, # 路由函數
    {"veto_path": "gm", "normal_path": "gm"},  # 兩條路都到 gm，但狀態不同
)

# 各品牌成本係數（模擬不同代工廠的實際成本結構）
BRAND_COST_FACTOR = {
    "nike":          0.35,  # 毛利 65%
    "adidas":        0.38,  # 毛利 62%
    "new balance":   0.40,  # 毛利 60%
    "puma":          0.88,  # 毛利 12% → 低於 15% 紅線 🚨
    "asics":         0.36,  # 毛利 64%
    "under armour":  0.42,  # 毛利 58%
}

# app/agents/debate_graph.py
from pydantic_ai import Agent
from langgraph.graph import StateGraph, START, END

MODEL = os.getenv("LLM_MODEL", "openai:gpt-4o-mini")

sales_agent = Agent(
    MODEL,
    system_prompt=(
        "你是鞋墊廠的業務主任，使用繁體中文回答。\n"
        "分析重點：訂單達標率、客戶關係維護、交期承諾、市場機會。\n"
        "回答格式：3-5 點建議，每點不超過 50 字。\n"
        "結尾加上「業務觀點結論：」一句話。"
    ),
)

production_agent = Agent(
    MODEL,
    system_prompt=(
        "你是鞋墊廠的生產管理主任，使用繁體中文回答。\n"
        "分析重點：廠區稼動率、產能瓶頸、交期可行性、備料風險。\n"
        "回答格式：3-5 點評估，每點不超過 50 字。\n"
        "結尾加上「生管觀點結論：」一句話。"
    ),
)

finance_agent = Agent(
    MODEL,
    system_prompt=(
        "你是鞋墊廠的財務主任，使用繁體中文回答。\n"
        "分析重點：毛利率健全度、現金流、付款條件風險。\n"
        "公司毛利率紅線：15%。低於紅線必須明確反對並說明原因。\n"
        "如有財務否決，最後一行必須以「財務否決：」開頭說明原因。"
    ),
)

gm_agent = Agent(
    MODEL,
    system_prompt=(
        "你是鞋墊廠的總經理，使用繁體中文回答。\n"
        "決策優先序：財務健全 > 產能可行 > 業務機會。\n"
        "回答格式：\n"
        "【最終決策】：接受 / 拒絕 / 條件接受\n"
        "【決策理由】：2-3 句話\n"
        "【行動指示】：具體的下一步（2-3 條）"
    ),
)

async def sales_node(state: DebateState) -> dict:
    summary = get_orders_summary()
    context = (
        f"問題：{state['question']}\n"
        f"目前整體訂單摘要：各品牌營收 {summary['by_brand']}，"
        f"訂單狀態分佈 {summary['by_status']}"
    )
    result = await sales_agent.run(context)
    return {"sales_view": result.data}

async def production_node(state: DebateState) -> dict:
    factories_info = [get_factory_capacity(fc) for fc in ["TW-01", "VN-02", "ID-03"]]
    cap_summary = [
        f"{f['factory_name']}：稼動率 {f['utilization_pct']}%，"
        f"可接單 {f['available_qty']} 雙，交期 {f['lead_time_days']} 天"
        for f in factories_info if f.get("found")
    ]
    context = (
        f"問題：{state['question']}\n"
        f"目前廠區狀況：{' | '.join(cap_summary)}"
    )
    result = await production_agent.run(context)
    return {"production_view": result.data}

async def finance_node(state: DebateState) -> dict:
    margin_lines = []
    veto_brands = []

    for brand in ["Nike", "Adidas", "New Balance", "Puma", "Asics", "Under Armour"]:
        m = calculate_gross_margin(brand)
        if m.get("found"):
            margin_lines.append(f"{brand}: {m['gross_margin_pct']}%")
            if m["below_redline"]:
                veto_brands.append(f"{brand}({m['gross_margin_pct']}%)")

    veto_triggered = len(veto_brands) > 0
    veto_reason = f"以下品牌毛利率低於 15% 紅線：{', '.join(veto_brands)}" if veto_triggered else ""

    context = (
        f"問題：{state['question']}\n"
        f"各品牌毛利率：{', '.join(margin_lines)}\n"
        + (f"⚠️ 警告：{veto_reason}" if veto_triggered else "所有品牌毛利率均符合紅線要求")
    )
    result = await finance_agent.run(context)
    return {
        "finance_view": result.data,
        "finance_veto": veto_triggered,   # ← 這個旗標會被路由函數讀取
        "veto_reason": veto_reason,
    }

async def gm_node(state: DebateState) -> dict:
    veto_notice = (
        f"\n⚠️ 財務否決已觸發：{state.get('veto_reason', '')}\n"
        "請特別考量是否接受財務否決，或要求財務提出改善條件後重新評估。"
        if state.get("finance_veto")
        else ""
    )
    synthesis = (
        f"問題：{state['question']}\n\n"
        f"業務觀點：\n{state['sales_view']}\n\n"
        f"生管觀點：\n{state['production_view']}\n\n"
        f"財務觀點：\n{state['finance_view']}"
        f"{veto_notice}"
    )
    result = await gm_agent.run(synthesis)
    return {"final_decision": result.data}

def build_debate_graph():
    builder = StateGraph(DebateState)

    # 註冊節點
    builder.add_node("sales", sales_node)
    builder.add_node("production", production_node)
    builder.add_node("finance", finance_node)
    builder.add_node("gm", gm_node)

    # 線性邊：sales → production → finance
    builder.add_edge(START, "sales")
    builder.add_edge("sales", "production")
    builder.add_edge("production", "finance")

    # 條件邊：finance → (veto_path / normal_path) → gm
    builder.add_conditional_edges(
        "finance",
        route_after_finance,                        # 路由函數
        {"veto_path": "gm", "normal_path": "gm"},  # 兩條路都到 gm
    )

    builder.add_edge("gm", END)
    return builder.compile()

debate_graph = build_debate_graph()

# app/routers/debate.py
from fastapi import APIRouter
from pydantic import BaseModel
from app.agents.debate_graph import debate_graph, DebateState

router = APIRouter(prefix="/api/v1", tags=["debate"])

class DebateRequest(BaseModel):
    message: str
    user_id: str = "gm_harry"

class DebateResponse(BaseModel):
    sales_view: str
    production_view: str
    finance_view: str
    finance_veto: bool
    veto_reason: str
    final_decision: str
    user_role: str
    user_name: str

@router.post("/agent/debate", response_model=DebateResponse)
async def agent_debate(request: DebateRequest):
    profile = USER_PROFILES.get(request.user_id, DEFAULT_PROFILE)
    initial_state: DebateState = {
        "question": request.message,
        "user_role": profile["role"],
        "user_name": profile["name"],
        "sales_view": "",
        "production_view": "",
        "finance_view": "",
        "finance_veto": False,
        "veto_reason": "",
        "final_decision": "",
    }
    result = await debate_graph.ainvoke(initial_state)
    return DebateResponse(**{k: result[k] for k in DebateResponse.model_fields}, ...)

from app.routers import orders, agent, debate   # ← 新增 debate
app.include_router(debate.router)

# 在 venv 環境下
pip install langgraph>=0.2.0

pip install -r requirements.txt

cd C:\Users\Harry\Desktop\30-Days\ERP-AI-Agent\backend
.\.venv\Scripts\Activate.ps1
uvicorn app.main:app --reload --port 8000

curl -X POST http://localhost:8000/api/v1/agent/debate \
  -H "Content-Type: application/json" \
  -d '{"message": "我們是否應該增加 Puma 的訂單量？", "user_id": "gm_harry"}'

{
  "sales_view": "1. Puma 去年完成率 100%，客戶關係穩定...\n業務觀點結論：建議積極爭取 Puma 擴單。",
  "production_view": "1. VN-02 目前稼動率 91%，已接近上限...\n生管觀點結論：短期擴單有產能風險。",
  "finance_view": "1. Puma 訂單毛利率 12%，低於公司紅線 15%...\n財務否決：Puma 毛利率不符最低門檻，建議重新議價後再決定。",
  "finance_veto": true,
  "veto_reason": "以下品牌毛利率低於 15% 紅線：Puma(12%)",
  "final_decision": "【最終決策】：條件接受\n【決策理由】：業務端客戶關係良好，但財務紅線不可逾越...\n【行動指示】：\n1. 要求業務在兩週內與 Puma 重新議價，目標毛利率 18%\n2. ...",
  "finance_veto": true,
  "user_role": "general_manager",
  "user_name": "Harry"
}

curl -X POST http://localhost:8000/api/v1/agent/debate \
  -H "Content-Type: application/json" \
  -d '{"message": "Nike Q3 訂單量能否如期交貨？", "user_id": "sales_alice"}'