過去 29 天，我們從最基礎的 API 呼叫，一路打通了長文本處理、強制結構化輸出、多智能體協作，再到結合 Telegram 的人機協作 (HITL)。今天，我們迎來最終章。
在企業級軟體工程中，單一技術點的突破只是起點，「系統整合 (System Integration)」與「商業價值」才是決勝的關鍵。我們不再把大語言模型當作全能的魔法師，而是將其降級為「只負責填寫 JSON 的決策元件 (Jev)」，所有的邏輯驗證、防呆與外部呼叫，全部交由確定的 Python 程式碼 (Harness) 接管。
這套被我們反覆打磨的底層引擎，就是 JEV+ 完整體。
一、 JEV+ 完整體：Micro-SaaS 後端引擎程式碼
這是你這 30 天的心血結晶，也是一人企業 (Solopreneur) 的核心印鈔機。請將這段程式碼視為你的「數位軟體公司」，它包含了 PM、導演與行銷三大部門，且具備絕對的防錯與審核機制。
import os
import logging
from typing import List, Optional, Literal
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel, Field, model_validator, ValidationError
import telebot
from telebot.types import InlineKeyboardMarkup, InlineKeyboardButton
# 匯入 Google ADK 核心元件
from google.adk import Agent, Workflow
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)
# ==========================================
# [基礎設施] Telegram 與 FastAPI 設定
# ==========================================
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN", "YOUR_TG_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID", "YOUR_ADMIN_CHAT_ID")
ALLOWED_TG_USERS = [123456789] # 白名單機制，落實零信任資安
bot = telebot.TeleBot(TELEGRAM_BOT_TOKEN)
app_api = FastAPI(title="Video Micro-SaaS Engine")
# ==========================================
# [J: JSON/Typed] 狀態層與強型別定義
# 限制 Jev (Agent) 只能填寫這些明確的決策格子
# ==========================================
class PMDecision(BaseModel):
target_audience: str = Field(description="目標受眾")
core_pain_point: str = Field(description="核心痛點")
tone: str = Field(description="視覺與文案調性")
recommended_duration_sec: int = Field(description="建議總秒數 (如 15, 30)")
class Scene(BaseModel):
scene_number: int
time_sec: int = Field(description="分鏡秒數")
camera: str = Field(description="運鏡")
visual: str = Field(description="畫面描述，限 30 字內")
audio: str = Field(description="旁白")
class DirectorDecision(BaseModel):
target_duration: int
scenes: List[Scene]
# [V: Validation] Harness (Python) 執行的絕對判斷
@model_validator(mode='after')
def validate_duration(self) -> 'DirectorDecision':
total = sum(s.time_sec for s in self.scenes)
if total != self.target_duration:
# Python 攔截 LLM 的數學錯誤，直接拋出例外
raise ValueError(f"【數學錯誤】分鏡秒數加總為 {total}s，但目標時長為 {self.target_duration}s！請重新分配。")
return self
class MarketingDecision(BaseModel):
fb_hook: str = Field(description="Facebook 爆款開頭")
threads_hook: str = Field(description="Threads 碎碎念開頭")
hashtags: List[str]
# 全局白板 (Shared State)
class CampaignState(BaseModel):
client_brief: str = ""
pm_spec: Optional[PMDecision] = None
director_storyboard: Optional[DirectorDecision] = None
marketing_assets: Optional[MarketingDecision] = None
is_approved_by_human: bool = False
status: Literal["INIT", "PROCESSING", "PENDING_APPROVAL", "PUBLISHED", "FAILED"] = "INIT"
# ==========================================
# [決策層] 定義 Jev Agents
# ==========================================
pm_agent = Agent(
name="pm_agent",
model="gemini-2.5-flash",
instruction="你是廣告 PM，請根據客戶的一句話 Brief，輸出嚴謹的 JSON 規格。",
response_schema=PMDecision
)
director_agent = Agent(
name="director_agent",
model="gemini-2.5-pro",
instruction="你是導演，請根據規格規劃分鏡，務必確保 scenes 的 time_sec 加總等於 target_duration。",
response_schema=DirectorDecision
)
marketing_agent = Agent(
name="marketing_agent",
model="gemini-2.5-pro",
instruction="你是社群行銷，請為這支影片撰寫各平台的專屬文案，必須包含極具吸引力的 Hook。",
response_schema=MarketingDecision
)
# ==========================================
# [執行層] ADK Harness 節點操作與錯誤迴圈
# ==========================================
def run_pm(state: CampaignState) -> CampaignState:
logger.info("🧠 [PM 決策器] 進行發散與收斂...")
state.pm_spec = pm_agent.invoke(state.client_brief)
return state
def run_director(state: CampaignState) -> CampaignState:
logger.info("🎬 [Director 決策器] 規劃分鏡中...")
prompt = f"受眾: {state.pm_spec.target_audience}, 秒數: {state.pm_spec.recommended_duration_sec}s"
# [E: Error-Driven] Jev 的自我修正迴圈
for attempt in range(3):
try:
state.director_storyboard = director_agent.invoke(prompt)
logger.info("✅ 數學驗證通過！")
return state
except ValidationError as e:
logger.warning(f"⚠️️ [Harness 攔截] 發現數學錯誤，強制退回重算 (第 {attempt+1} 次)...")
prompt += f"\n\n系統退件原因：{e}。請重新計算所有分鏡的時間！"
state.status = "FAILED"
raise RuntimeError("Director 決策器經過 3 次修正仍無法滿足數學限制。")
def run_marketing(state: CampaignState) -> CampaignState:
logger.info("🔥 [Marketing 決策器] 產出跨平台社群包裝...")
state.marketing_assets = marketing_agent.invoke(str(state.director_storyboard))
state.status = "PENDING_APPROVAL"
return state
def publish_node(state: CampaignState) -> CampaignState:
if state.is_approved_by_human:
logger.info("🚀 [Harness 執行] 收到授權，正式呼叫外部 API 上架與發文！")
state.status = "PUBLISHED"
else:
logger.warning("❌ 提案被人類退回。")
return state
# ==========================================
# [路由層] 編織 Workflow 與中斷點
# ==========================================
workflow = Workflow(name="MicroSaaS_Video_Factory", state_schema=CampaignState)
workflow.add_sequence(["START", run_pm, run_director, run_marketing, publish_node, "END"])
# [+ : Human-in-the-loop] 在高風險動作前強制凍結
app = workflow.compile(interrupt_before=["publish_node"])
thread_config = {"configurable": {"thread_id": "saas_order_001"}}
# ==========================================
# [控制層] Telegram HITL 審核與 FastAPI
# ==========================================
def request_approval(state: CampaignState):
markup = InlineKeyboardMarkup()
markup.add(
InlineKeyboardButton("👍 核准發布並扣款", callback_data="approve"),
InlineKeyboardButton("👎 退回專案", callback_data="reject")
)
msg = f"🛑 *【客戶專案審核】*\n\n👤 *受眾:* {state.pm_spec.target_audience}\n⏱️ *總時長:* {state.director_storyboard.target_duration} 秒\n🔥 *FB Hook:* {state.marketing_assets.fb_hook}\n\n是否執行呼叫外部服務？"
bot.send_message(TELEGRAM_CHAT_ID, msg, parse_mode="Markdown", reply_markup=markup)
@bot.callback_query_handler(func=lambda call: True)
def handle_approval(call):
if call.from_user.id not in ALLOWED_TG_USERS:
bot.answer_callback_query(call.id, "❌ 存取被拒：無審核權限！", show_alert=True)
return
is_approved = (call.data == "approve")
bot.edit_message_text(f"{call.message.text}\n\n✅ *決策：{'已核准' if is_approved else '已退回'}*", call.message.chat.id, call.message.message_id, parse_mode="Markdown")
# 覆寫凍結狀態並喚醒工作流
app.update_state(thread_config, {"is_approved_by_human": is_approved})
app.run(None, config=thread_config)
def background_adk_runner(brief: str):
initial_state = CampaignState(client_brief=brief)
frozen_state = app.run(initial_state, config=thread_config)
if app.get_state(thread_config).next == ("publish_node",):
request_approval(frozen_state)
@app_api.post("/api/v1/generate_campaign", status_code=202)
async def create_campaign(req: dict, background_tasks: BackgroundTasks):
background_tasks.add_task(background_adk_runner, req.get("brief"))
return {"status": "Processing", "message": "專案立項，背景處理中..."}
二、 程式碼深度解析：JEV+ 是如何運作的？
在這份程式碼中，大語言模型不再是一個可以隨意聊天的機器人，它被降級為數位生產線上的一個個「決策元件」。
J (JSON/Typed) 決策邊界：
Agent（如 director_agent）沒有自由發揮的空間。它的輸出來自 DirectorDecision 的 Pydantic 類別。它必須精準給出 target_duration 與 scenes 陣列。在這裡，LLM 負責創意與邏輯的「決策」。
V (Validation) 系統防線：
LLM 的數學能力不穩定，如果它規劃的秒數加起來不等於總長度怎麼辦？我們在 DirectorDecision 內使用 @model_validator。這就是 Harness (系統控制) 的力量。Python 程式碼會無情地攔截運算，若發現 sum(time_sec) 錯誤，直接引爆 ValueError，絕對不讓錯誤資料流入下一關。
E (Error-Driven) 自我修正迴圈：
在 run_director 的 try...except 迴圈中，我們捕捉 Python 拋出的數學錯誤，將其轉換為指令（如 系統退件原因：... 請重新計算）塞回給 LLM 重新生成。如果修正 3 次依然失敗，系統強行中斷。這防止了 API 成本的無限消耗。
三、 一人企業的無限可能：SaaS 商業化展望
這套 JEV+ 架構，就是你打造一人企業最強大的護城河。未來的 SaaS 商業化路徑，關鍵在於「極度垂直」與「成效計價」。
鎖定極度垂直的微型 SaaS (Micro-SaaS)
不要做「通用型 AI 助手」，去解決特定族群最痛苦的繁瑣流程。例如上述的「廣告代理商無頭企劃引擎」，廣告公司只需透過 Webhook 把客戶 Brief 打過來，你的系統在背景執行 Jev 決策迴圈與秒數驗證，最後吐出完美的 JSON 分鏡表與社群文案供其直接使用。
API-First 的無頭微服務 (Headless API)
許多中小企業已經有了 ERP 或 CRM，他們需要的是「大腦」而不是另一個「介面」。你可以將這份 FastAPI 封裝成一個訂閱制的 API 服務。客戶打資料過來，你的虛擬團隊在雲端背景跑完運算、通過你的 Telegram 白名單授權後，再把完美結構化的結果推回去。
從「訂閱制」走向「成效制 (Work-as-a-Service)」
Agentic AI 顛覆了軟體的定價邏輯。客戶買的不再是「工具的使用權」，而是「數位員工的勞動力」。你可以按照產出計價：例如「每次成功產出一份技術工作坊企劃與推廣包，收費 5 美元」。透過 ADK 的 Trace 系統，精準監控每一次的 Token 成本，輕鬆計算出豐厚的毛利。
30 天的挑戰正式畫下句點。AI 不會取代軟體工程師，但懂得運用 Agent 建立自動化系統的工程師，將會成為下一個時代的超級個體。帶上這批虛擬員工，去開創屬於你的一人企業版圖吧！
iThome鐵人賽
看影片追技術