iT邦幫忙

2026 iThome 鐵人賽

DAY 15
0
Build on Google AI

打造零成本企業級 AI Agent:以 Gemini 2.5 Flash 構建金融分析助手與維運實戰系列 第 15

【Day 15】核心管道重構:三階段分析管道(Stage 1-3 Pipeline)設計與實作 (daily_analysis.py)

  • 分享至 

  • xImage
  •  

「要讓 LLM 產出的金融分析報告具備高度專業性與邏輯一致性,不能寄望於單一 Prompt,而是要將分析流程拆解為具備強約束力的三階段資料管道(Stage 1-3 Pipeline)。」

在先前,我們完成了基礎設施的容器化與自動化部署。進入後半段,我們將重點轉向 Angelina Agent 分析品質的進化!

在原始版本的 daily_analysis.py 中,盤後分析採用單一 Prompt 處理,經常出現「前面看空、後面卻推薦買進策略」的盤態矛盾。為了徹底解決這個問題,我們在最新的 daily_analysis.py 中將 generate_report() 重構為 「三階段分析管道(Stage 1-3 Pipeline)」,透過階段性推理與 MANDATORY 方向約束,確保產出邏輯嚴密且專業的金融報告。

本篇重點摘要

  1. 剖析傳統單一 Prompt 的痛點:盤態矛盾與 Context 浪費。
  2. 拆解 Stage 1 結構化盤態評估(JSON 格式:bullish / bearish / neutral)。
  3. 實作 Stage 2 本地 RAGEngine 針對性策略檢索(0 API 額外消耗)。
  4. 實作 Stage 3 MANDATORY 強制方向約束與最終報告生成。

一、三階段分析管道(Stage 1-3 Pipeline)運作架構

全新重構的 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 瓶頸約束,生成最終一致性報告
└─────────────────────────────────┘

二、Stage 1:結構化盤態評估 (JSON Assessment)

第一階段僅將 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 2:本地針對性知識庫檢索 (Targeted KB Search)

解析出 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 3:MANDATORY 方向約束與報告生成

第三階段根據 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)的重構,我們徹底解決了舊版腳本的兩大痛點:

  1. 觀點絕對一致:Stage 3 中的 MANDATORY 指令強行約束了 AI 的情緒基調,大漲日絕不可能產出極度悲觀的矛盾結論。
  2. 0 API 額外開銷:Stage 2 的策略檢索完全在本地 ChromaDB 進行,保持了高響應度與低配額消耗。

明天(Day 16)我們將深入解析 daily_analysis.py 中 7 條 Core Execution Rules、金額兆/億元前處理格式化與週末自動週報切換機制!

明日預告:【Day 16】分析品質提升:Core Execution Rules 與金額兆/億元自動格式化


上一篇
【Day 14】自動化部署管線:deploy.sh 腳本實作與 CI/CD 流程整合
系列文
打造零成本企業級 AI Agent:以 Gemini 2.5 Flash 構建金融分析助手與維運實戰15
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言