「要讓 LLM 產出的金融分析報告具備高度專業性與邏輯一致性,不能寄望於單一 Prompt,而是要將分析流程拆解為具備強約束力的三階段資料管道(Stage 1-3 Pipeline)。」
在先前,我們完成了基礎設施的容器化與自動化部署。進入後半段,我們將重點轉向 Angelina Agent 分析品質的進化!
在原始版本的 daily_analysis.py 中,盤後分析採用單一 Prompt 處理,經常出現「前面看空、後面卻推薦買進策略」的盤態矛盾。為了徹底解決這個問題,我們在最新的 daily_analysis.py 中將 generate_report() 重構為 「三階段分析管道(Stage 1-3 Pipeline)」,透過階段性推理與 MANDATORY 方向約束,確保產出邏輯嚴密且專業的金融報告。
全新重構的 generate_report() 運算流程如下:
Plaintext
[ 原始 TWSE / 美股數據 ]
│
▼
┌─────────────────────────────────┐
│ Stage 1: Structured Assessment │ ──> 輸出 JSON: market_direction (bullish/bearish/neutral)
└─────────────────────────────────┘
│
▼
┌─────────────────────────────────┐
│ Stage 2: Targeted KB Search │ ──> 呼叫 RAGEngine 本地檢索針對性策略 (0 API Cost)
└─────────────────────────────────┘
│
▼
┌─────────────────────────────────┐
│ Stage 3: Generate Final Report │ ──> 帶入 MANDATORY 瓶頸約束,生成最終一致性報告
└─────────────────────────────────┘
第一階段僅將 TWSE 與美股數據傳入,要求 Gemini 僅輸出 JSON 格式的短評,嚴禁加入個人觀點或投資建議,以獲得極度純粹的市場方向:
Python
# Stage 1: Structured Market Assessment
print(" [Stage 1] Analyzing market data for structured assessment...")
stage1_prompt = (
"You are a market data analyst. Analyze the following raw market data and produce a SHORT "
"structured assessment in JSON format. Do NOT add any opinion or investment advice.\n\n"
f"## Taiwan Market Data\n{tw_section}\n\n"
f"## US Market Data\n{us_section}\n\n"
"## Output (JSON only, no markdown fences):\n"
"{\n"
' "market_direction": "bullish" or "bearish" or "neutral",\n'
' "tw_change_pct": <the percentage change as a number>,\n'
' "key_events": ["observation 1", "observation 2", "observation 3"],\n'
' "institutional_flow": "<net buy/sell summary in one sentence>"\n'
"}\n\n"
"Rules:\n"
"- bullish: index up > +0.5%\n"
"- bearish: index down < -0.5%\n"
"- neutral: between -0.5% and +0.5%\n"
"- Only use the data provided. Do NOT hallucinate numbers.\n"
"- Output ONLY the JSON object, nothing else."
)
stage1_text = await _call_gemini(client, stage1_prompt, temperature=0.2, max_tokens=1024)
stage1_assessment = json.loads(stage1_text)
解析出 Stage 1 的 market_direction(例如 bullish)後,系統直接呼叫本地 RAGEngine,針對該方向檢索特定的交易策略,完全不浪費 Gemini API Token:
Python
# Stage 2: Targeted Knowledge Base Search
print(" [Stage 2] Searching knowledge base based on market direction...")
from app.services.rag_engine import RAGEngine
direction = stage1_assessment.get('market_direction', 'neutral')
if direction == 'bullish':
search_queries = ["逢高調節 順勢操作 倉位控制"]
elif direction == 'bearish':
search_queries = ["停損 風險管理 反彈觀察"]
else:
search_queries = ["區間操作 觀望策略"]
engine = RAGEngine()
await engine.initialise()
all_chunks = []
for query in search_queries:
chunks = await engine.search(query=query, top_k=5)
for chunk in chunks:
all_chunks.append(chunk.text)
stage2_knowledge = "\n\n".join(all_chunks[:8])
第三階段根據 Stage 1 的評估結果動態構建 MANDATORY 方向約束指令,並將 Stage 2 檢索到的策略一併傳入,生成最終報告:
Python
# Stage 3: Generate Final Report with Direction Constraint
direction = stage1_assessment.get('market_direction', 'neutral')
tw_change = stage1_assessment.get('tw_change_pct', 0)
if direction == 'bullish':
mandatory_line = (
f"MANDATORY: 今日為多方格局 ({tw_change:+.2f}%)。"
f"你必須先肯定多方動能與漲勢,再適度帶入逢高順勢調節的風險提醒。"
)
elif direction == 'bearish':
mandatory_line = (
f"MANDATORY: 今日為空方格局 ({tw_change:+.2f}%)。"
f"你必須先說明下跌事實與風險,再帶入停損與反彈觀察建議。"
)
else:
mandatory_line = (
f"MANDATORY: 今日為中性盤整格局 ({tw_change:+.2f}%)。"
f"你必須以區間操作與觀望策略為主軸撰寫報告。"
)
stage3_prompt = (
f"{mandatory_line}\n\n"
"# Role & Tone\n"
"你是一位專業、客觀且具備實戰經驗的台灣股市首席分析師。報告應兼具數據精準度、高可讀性與實用操作指引。\n\n"
f"{stage1_summary}\n"
f"# 市場數據\n{tw_section}\n\n{us_section}\n\n{news_section}\n\n"
f"# 投資知識參考\n{combined_knowledge}\n\n"
"# 輸出格式\n"
"請依序產出:1.台股摘要 2.美股狀況 3.趨勢分析 4.操作建議(含倉位與停損位) 5.免責聲明\n"
)
透過三階段分析管道(Stage 1-3 Pipeline)的重構,我們徹底解決了舊版腳本的兩大痛點:
明天(Day 16)我們將深入解析 daily_analysis.py 中 7 條 Core Execution Rules、金額兆/億元前處理格式化與週末自動週報切換機制!
明日預告:【Day 16】分析品質提升:Core Execution Rules 與金額兆/億元自動格式化