From aca1903592a5f24c2c9c2512364d29350c4a78be Mon Sep 17 00:00:00 2001 From: MarkLo Date: Tue, 25 Nov 2025 21:16:11 +0800 Subject: [PATCH] --- README.md | 2 + build/lib/cli/__init__.py | 2 - build/lib/cli/main.py | 1189 ------------------------------------- build/lib/cli/models.py | 24 - build/lib/cli/utils.py | 609 ------------------- 5 files changed, 2 insertions(+), 1824 deletions(-) delete mode 100644 build/lib/cli/__init__.py delete mode 100644 build/lib/cli/main.py delete mode 100644 build/lib/cli/models.py delete mode 100644 build/lib/cli/utils.py diff --git a/README.md b/README.md index be144c2c..79897f47 100644 --- a/README.md +++ b/README.md @@ -32,6 +32,8 @@ - 🔌 **RESTful API** - 完整的後端 API 支援 - 🐳 **一鍵部署** - 支援 Docker Compose 部署 - 🔑 **BYOK (Bring Your Own Key)** - 使用者自帶 API 金鑰，保障隱私與成本控制 +- 📄 **JSON 格式報告** - 完整的 JSON 結構化分析報告，便於程式化處理與整合 +- ⬇️ **一鍵下載報告** - 支援將分析結果匯出為 JSON 檔案，方便保存與分享 --- diff --git a/build/lib/cli/__init__.py b/build/lib/cli/__init__.py deleted file mode 100644 index 37597c3c..00000000 --- a/build/lib/cli/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -# -*- coding: utf-8 -*- -# 這個檔案的存在是為了讓 Python 將 `cli` 目錄視為一個套件。 diff --git a/build/lib/cli/main.py b/build/lib/cli/main.py deleted file mode 100644 index 9e947c12..00000000 --- a/build/lib/cli/main.py +++ /dev/null @@ -1,1189 +0,0 @@ -# 匯入必要的模組 -from typing import Optional -import datetime -import typer -from pathlib import Path -from functools import wraps -from rich.console import Console -from dotenv import load_dotenv - -# 從 .env 檔案載入環境變數（強制覆蓋系統環境變數） -load_dotenv(override=True) -from rich.panel import Panel -from rich.spinner import Spinner -from rich.live import Live -from rich.columns import Columns -from rich.markdown import Markdown -from rich.layout import Layout -from rich.text import Text -from rich.table import Table -from collections import deque -import time -from rich.tree import Tree -from rich import box -from rich.align import Align -from rich.rule import Rule - -# 匯入專案內的模組 -from tradingagents.graph.trading_graph import TradingAgentsXGraph -from tradingagents.default_config import DEFAULT_CONFIG -from cli.models import AnalystType -from cli.utils import * - -# 初始化 rich Console -console = Console() - -# 建立 Typer 應用程式 -app = typer.Typer( - name="TradingAgentsX", - help="TradingAgentsX CLI：多代理 LLM 金融交易框架", - add_completion=True, # 啟用 shell 自動補全 -) - - -# 建立一個 deque 來儲存最近的訊息，並設定最大長度 -class MessageBuffer: - """ - 用於儲存和管理應用程式訊息、工具呼叫和報告狀態的緩衝區。 - """ - def __init__(self, max_length=100): - # 使用 deque 儲存帶有時間戳的訊息，以實現高效的 append 和 pop 操作 - self.messages = deque(maxlen=max_length) - self.tool_calls = deque(maxlen=max_length) - self.current_report = None # 當前顯示的報告部分 - self.final_report = None # 儲存完整的最終報告 - # 代理狀態字典，追蹤每個代理的進度 - self.agent_status = { - # 分析師團隊 - "Market Analyst": "pending", - "Social Analyst": "pending", - "News Analyst": "pending", - "Fundamentals Analyst": "pending", - # 研究團隊 - "Bull Researcher": "pending", - "Bear Researcher": "pending", - "Research Manager": "pending", - # 交易團隊 - "Trader": "pending", - # 風險管理團隊 - "Risky Analyst": "pending", - "Neutral Analyst": "pending", - "Safe Analyst": "pending", - # 投資組合管理團隊 - "Portfolio Manager": "pending", - } - self.current_agent = None # 當前正在執行的代理 - # 報告區塊字典，儲存分析過程中的各個報告 - self.report_sections = { - "market_report": None, - "sentiment_report": None, - "news_report": None, - "fundamentals_report": None, - "investment_plan": None, - "trader_investment_plan": None, - "final_trade_decision": None, - } - - def add_message(self, message_type, content): - """新增一條訊息到緩衝區。""" - timestamp = datetime.datetime.now().strftime("%H:%M:%S") - self.messages.append((timestamp, message_type, content)) - - def add_tool_call(self, tool_name, args): - """新增一條工具呼叫記錄到緩衝區。""" - timestamp = datetime.datetime.now().strftime("%H:%M:%S") - self.tool_calls.append((timestamp, tool_name, args)) - - def update_agent_status(self, agent, status): - """更新代理的狀態。""" - if agent in self.agent_status: - self.agent_status[agent] = status - self.current_agent = agent - - def update_report_section(self, section_name, content): - """更新報告的特定區塊。""" - if section_name in self.report_sections: - self.report_sections[section_name] = content - self._update_current_report() - - def _update_current_report(self): - """更新當前用於顯示的報告。""" - # 為了面板顯示，只顯示最近更新的部分 - latest_section = None - latest_content = None - - # 找到最近更新的部分 - for section, content in self.report_sections.items(): - if content is not None: - latest_section = section - latest_content = content - - if latest_section and latest_content: - # 格式化當前部分以供顯示 - section_titles = { - "market_report": "市場分析", - "sentiment_report": "社群情緒", - "news_report": "新聞分析", - "fundamentals_report": "基本面分析", - "investment_plan": "研究團隊決策", - "trader_investment_plan": "交易團隊計畫", - "final_trade_decision": "投資組合管理決策", - } - self.current_report = ( - f"### {section_titles[latest_section]}\n{latest_content}" - ) - - # 更新完整的最終報告 - self._update_final_report() - - def _update_final_report(self): - """更新完整的最終報告。""" - report_parts = [] - - # 分析師團隊報告 - if any( - self.report_sections[section] - for section in [ - "market_report", - "sentiment_report", - "news_report", - "fundamentals_report", - ] - ): - report_parts.append("## 分析師團隊報告") - if self.report_sections["market_report"]: - report_parts.append( - f"### 市場分析\n{self.report_sections['market_report']}" - ) - if self.report_sections["sentiment_report"]: - report_parts.append( - f"### 社群情緒\n{self.report_sections['sentiment_report']}" - ) - if self.report_sections["news_report"]: - report_parts.append( - f"### 新聞分析\n{self.report_sections['news_report']}" - ) - if self.report_sections["fundamentals_report"]: - report_parts.append( - f"### 基本面分析\n{self.report_sections['fundamentals_report']}" - ) - - # 研究團隊報告 - if self.report_sections["investment_plan"]: - report_parts.append("## 研究團隊決策") - report_parts.append(f"{self.report_sections['investment_plan']}") - - # 交易團隊報告 - if self.report_sections["trader_investment_plan"]: - report_parts.append("## 交易團隊計畫") - report_parts.append(f"{self.report_sections['trader_investment_plan']}") - - # 投資組合管理決策 - if self.report_sections["final_trade_decision"]: - report_parts.append("## 投資組合管理決策") - report_parts.append(f"{self.report_sections['final_trade_decision']}") - - self.final_report = "\n\n".join(report_parts) if report_parts else None - - -# 實例化訊息緩衝區 -message_buffer = MessageBuffer() - - -def create_layout(): - """建立 CLI 的版面配置。""" - layout = Layout() - layout.split_column( - Layout(name="header", size=3), - Layout(name="main"), - Layout(name="footer", size=3), - ) - layout["main"].split_column( - Layout(name="upper", ratio=3), Layout(name="analysis", ratio=5) - ) - layout["upper"].split_row( - Layout(name="progress", ratio=2), Layout(name="messages", ratio=3) - ) - return layout - - -def update_display(layout, spinner_text=None): - """更新 rich 即時顯示的內容。""" - # 包含歡迎訊息的頁首 - layout["header"].update( - Panel( - "[bold green]歡迎使用 TradingAgentsX CLI[/bold green]\n" - "[dim]© [Tauric Research](https://github.com/TauricResearch)[/dim]", - title="歡迎使用 TradingAgentsX", - border_style="green", - padding=(1, 2), - expand=True, - ) - ) - - # 顯示代理狀態的進度面板 - progress_table = Table( - show_header=True, - header_style="bold magenta", - show_footer=False, - box=box.SIMPLE_HEAD, # 使用帶有水平線的簡單頁首 - title=None, # 移除多餘的進度標題 - padding=(0, 2), # 新增水平內邊距 - expand=True, # 使表格擴展以填滿可用空間 - ) - progress_table.add_column("團隊", style="cyan", justify="center", width=20) - progress_table.add_column("代理", style="green", justify="center", width=20) - progress_table.add_column("狀態", style="yellow", justify="center", width=20) - - # 按團隊對代理進行分組 - teams = { - "分析師團隊": [ - "Market Analyst", - "Social Analyst", - "News Analyst", - "Fundamentals Analyst", - ], - "研究團隊": ["Bull Researcher", "Bear Researcher", "Research Manager"], - "交易團隊": ["Trader"], - "風險管理": ["Risky Analyst", "Neutral Analyst", "Safe Analyst"], - "投資組合管理": ["Portfolio Manager"], - } - - for team, agents in teams.items(): - # 新增帶有團隊名稱的第一個代理 - first_agent = agents[0] - status = message_buffer.agent_status[first_agent] - if status == "in_progress": - spinner = Spinner( - "dots", text="[blue]進行中[/blue]", style="bold cyan" - ) - status_cell = spinner - else: - status_color = { - "pending": "yellow", - "completed": "green", - "error": "red", - }.get(status, "white") - status_cell = f"[{status_color}]{status}[/{status_color}]" - progress_table.add_row(team, first_agent, status_cell) - - # 新增團隊中的其餘代理 - for agent in agents[1:]: - status = message_buffer.agent_status[agent] - if status == "in_progress": - spinner = Spinner( - "dots", text="[blue]進行中[/blue]", style="bold cyan" - ) - status_cell = spinner - else: - status_color = { - "pending": "yellow", - "completed": "green", - "error": "red", - }.get(status, "white") - status_cell = f"[{status_color}]{status}[/{status_color}]" - progress_table.add_row("", agent, status_cell) - - # 在每個團隊後新增水平線 - progress_table.add_row("─" * 20, "─" * 20, "─" * 20, style="dim") - - layout["progress"].update( - Panel(progress_table, title="進度", border_style="cyan", padding=(1, 2)) - ) - - # 顯示最近訊息和工具呼叫的訊息面板 - messages_table = Table( - show_header=True, - header_style="bold magenta", - show_footer=False, - expand=True, # 使表格擴展以填滿可用空間 - box=box.MINIMAL, # 使用最小化的框線樣式以獲得更輕量的外觀 - show_lines=True, # 保留水平線 - padding=(0, 1), # 在列之間新增一些內邊距 - ) - messages_table.add_column("時間", style="cyan", width=8, justify="center") - messages_table.add_column("類型", style="green", width=10, justify="center") - messages_table.add_column( - "內容", style="white", no_wrap=False, ratio=1 - ) # 使內容列擴展 - - # 合併工具呼叫和訊息 - all_messages = [] - - # 新增工具呼叫 - for timestamp, tool_name, args in message_buffer.tool_calls: - # 如果工具呼叫參數過長，則截斷 - if isinstance(args, str) and len(args) > 100: - args = args[:97] + "..." - all_messages.append((timestamp, "工具", f"{tool_name}: {args}")) - - # 新增常規訊息 - for timestamp, msg_type, content in message_buffer.messages: - # 如果內容不是字串，則轉換為字串 - content_str = content - if isinstance(content, list): - # 處理內容區塊列表（Anthropic 格式） - text_parts = [] - for item in content: - if isinstance(item, dict): - if item.get('type') == 'text': - text_parts.append(item.get('text', '')) - elif item.get('type') == 'tool_use': - text_parts.append(f"[工具: {item.get('name', 'unknown')}]") - else: - text_parts.append(str(item)) - content_str = ' '.join(text_parts) - elif not isinstance(content_str, str): - content_str = str(content) - - # 如果訊息內容過長，則截斷 - if len(content_str) > 200: - content_str = content_str[:197] + "..." - all_messages.append((timestamp, msg_type, content_str)) - - # 按時間戳排序 - all_messages.sort(key=lambda x: x[0]) - - # 根據可用空間計算可以顯示多少條訊息 - # 從一個合理的數字開始，並根據內容長度進行調整 - max_messages = 12 # 從 8 增加到 12 以更好地填滿空間 - - # 獲取將適合面板的最後 N 條訊息 - recent_messages = all_messages[-max_messages:] - - # 將訊息新增到表格中 - for timestamp, msg_type, content in recent_messages: - # 格式化內容並自動換行 - wrapped_content = Text(content, overflow="fold") - messages_table.add_row(timestamp, msg_type, wrapped_content) - - if spinner_text: - messages_table.add_row("", "Spinner", spinner_text) - - # 如果訊息被截斷，則新增頁尾以指示 - if len(all_messages) > max_messages: - messages_table.footer = ( - f"[dim]顯示最近 {max_messages} 條訊息，共 {len(all_messages)} 條[/dim]" - ) - - layout["messages"].update( - Panel( - messages_table, - title="訊息與工具", - border_style="blue", - padding=(1, 2), - ) - ) - - # 顯示當前報告的分析面板 - if message_buffer.current_report: - layout["analysis"].update( - Panel( - Markdown(message_buffer.current_report), - title="當前報告", - border_style="green", - padding=(1, 2), - ) - ) - else: - layout["analysis"].update( - Panel( - "[italic]等待分析報告...[/italic]", - title="當前報告", - border_style="green", - padding=(1, 2), - ) - ) - - # 包含統計資訊的頁尾 - tool_calls_count = len(message_buffer.tool_calls) - llm_calls_count = sum( - 1 for _, msg_type, _ in message_buffer.messages if msg_type == "Reasoning" - ) - reports_count = sum( - 1 for content in message_buffer.report_sections.values() if content is not None - ) - - stats_table = Table(show_header=False, box=None, padding=(0, 2), expand=True) - stats_table.add_column("統計", justify="center") - stats_table.add_row( - f"工具呼叫: {tool_calls_count} | LLM 呼叫: {llm_calls_count} | 已生成報告: {reports_count}" - ) - - layout["footer"].update(Panel(stats_table, border_style="grey50")) - - -def get_user_selections(): - """在開始分析顯示之前獲取所有使用者選擇。""" - # 顯示 ASCII 藝術歡迎訊息 - with open("./cli/static/welcome.txt", "r") as f: - welcome_ascii = f.read() - - # 建立歡迎框內容 - welcome_content = f"{welcome_ascii}\n" - welcome_content += "[bold green]TradingAgentsX：多代理 LLM 金融交易框架 - CLI[/bold green]\n\n" - welcome_content += "[bold]工作流程步驟：[/bold]\n" - welcome_content += "I. 分析師團隊 → II. 研究團隊 → III. 交易員 → IV. 風險管理 → V. 投資組合管理\n\n" - welcome_content += ( - "[dim]由 [Tauric Research](https://github.com/TauricResearch) 開發[/dim]" - ) - - # 建立並置中歡迎框 - welcome_box = Panel( - welcome_content, - border_style="green", - padding=(1, 2), - title="歡迎使用 TradingAgentsX", - subtitle="多代理 LLM 金融交易框架", - ) - console.print(Align.center(welcome_box)) - console.print() # 在歡迎框後新增一個空行 - - # 為每個步驟建立一個帶框的問卷 - def create_question_box(title, prompt, default=None): - box_content = f"[bold]{title}[/bold]\n" - box_content += f"[dim]{prompt}[/dim]" - if default: - box_content += f"\n[dim]預設值: {default}[/dim]" - return Panel(box_content, border_style="blue", padding=(1, 2)) - - # 步驟 1：股票代碼 - console.print( - create_question_box( - "步驟 1：股票代碼", "輸入要分析的股票代碼", "SPY" - ) - ) - selected_ticker = get_ticker() - - # 步驟 2：分析日期 - default_date = datetime.datetime.now().strftime("%Y-%m-%d") - console.print( - create_question_box( - "步驟 2：分析日期", - "輸入分析日期 (YYYY-MM-DD)", - default_date, - ) - ) - analysis_date = get_analysis_date() - - # 步驟 3：選擇分析師 - console.print( - create_question_box( - "步驟 3：分析師團隊", "為分析選擇您的 LLM 分析師代理" - ) - ) - selected_analysts = select_analysts() - console.print( - f"[green]已選分析師：[/green] {', '.join(analyst.value for analyst in selected_analysts)}" - ) - - # 步驟 4：研究深度 - console.print( - create_question_box( - "步驟 4：研究深度", "選擇您的研究深度等級" - ) - ) - selected_research_depth = select_research_depth() - - # 步驟 5：LLM 供應商 - console.print( - create_question_box( - "步驟 5：LLM 供應商", "選擇要對話的服務" - ) - ) - selected_llm_provider, backend_url = select_llm_provider() - - # 步驟 6：思維代理 - console.print( - create_question_box( - "步驟 6：思維代理", "為分析選擇您的思維代理" - ) - ) - selected_shallow_thinker = select_shallow_thinking_agent(selected_llm_provider) - selected_deep_thinker = select_deep_thinking_agent(selected_llm_provider) - - # 步驟 7：嵌入模型供應商 - console.print( - create_question_box( - "步驟 7：嵌入模型供應商", "選擇嵌入模型服務（用於記憶體系統）" - ) - ) - embedding_provider, embedding_url = select_embedding_provider() - - # 步驟 8：API Keys - console.print( - create_question_box( - "步驟 8：API Keys", "輸入 API Keys（可留空使用 .env 中的設定）" - ) - ) - - import os - - # 從 .env 讀取 API Key - default_openai_key = os.getenv("OPENAI_API_KEY") - - # 快速思維模型 API Key - quick_think_api_key = get_api_key("快速思維模型", default_openai_key) - - # 深度思維模型 API Key - deep_think_api_key = get_api_key("深度思維模型", default_openai_key) - - # 嵌入模型 API Key - embedding_api_key = get_api_key("嵌入模型", default_openai_key) - - # Alpha Vantage API Key（必填） - alpha_vantage_key = os.getenv("ALPHA_VANTAGE_API_KEY") - if not alpha_vantage_key: - console.print("\n[yellow]未在 .env 中找到 ALPHA_VANTAGE_API_KEY[/yellow]") - alpha_vantage_key = get_api_key("Alpha Vantage", None) - else: - console.print(f"\n[green]✓ 使用 .env 中的 ALPHA_VANTAGE_API_KEY[/green]") - - return { - "ticker": selected_ticker, - "analysis_date": analysis_date, - "analysts": selected_analysts, - "research_depth": selected_research_depth, - "llm_provider": selected_llm_provider.lower(), - "backend_url": backend_url, - "shallow_thinker": selected_shallow_thinker, - "deep_thinker": selected_deep_thinker, - "embedding_provider": embedding_provider, - "embedding_url": embedding_url, - "quick_think_api_key": quick_think_api_key, - "deep_think_api_key": deep_think_api_key, - "embedding_api_key": embedding_api_key, - "alpha_vantage_api_key": alpha_vantage_key, - } - - -def get_ticker(): - """從使用者輸入中獲取股票代碼。""" - return typer.prompt("", default="SPY") - - -def get_analysis_date(): - """從使用者輸入中獲取分析日期。""" - while True: - date_str = typer.prompt( - "", default=datetime.datetime.now().strftime("%Y-%m-%d") - ) - try: - # 驗證日期格式並確保不是未來日期 - analysis_date = datetime.datetime.strptime(date_str, "%Y-%m-%d") - if analysis_date.date() > datetime.datetime.now().date(): - console.print("[red]錯誤：分析日期不能是未來日期[/red]") - continue - return date_str - except ValueError: - console.print( - "[red]錯誤：日期格式無效。請使用 YYYY-MM-DD[/red]" - ) - - -def display_complete_report(final_state): - """顯示包含基於團隊的面板的完整分析報告。""" - console.print("\n[bold green]完整分析報告[/bold green]\n") - - # I. 分析師團隊報告 - analyst_reports = [] - - # 市場分析師報告 - if final_state.get("market_report"): - analyst_reports.append( - Panel( - Markdown(final_state["market_report"]), - title="市場分析師", - border_style="blue", - padding=(1, 2), - ) - ) - - # 社群分析師報告 - if final_state.get("sentiment_report"): - analyst_reports.append( - Panel( - Markdown(final_state["sentiment_report"]), - title="社群分析師", - border_style="blue", - padding=(1, 2), - ) - ) - - # 新聞分析師報告 - if final_state.get("news_report"): - analyst_reports.append( - Panel( - Markdown(final_state["news_report"]), - title="新聞分析師", - border_style="blue", - padding=(1, 2), - ) - ) - - # 基本面分析師報告 - if final_state.get("fundamentals_report"): - analyst_reports.append( - Panel( - Markdown(final_state["fundamentals_report"]), - title="基本面分析師", - border_style="blue", - padding=(1, 2), - ) - ) - - if analyst_reports: - console.print( - Panel( - Columns(analyst_reports, equal=True, expand=True), - title="I. 分析師團隊報告", - border_style="cyan", - padding=(1, 2), - ) - ) - - # II. 研究團隊報告 - if final_state.get("investment_debate_state"): - research_reports = [] - debate_state = final_state["investment_debate_state"] - - # 看漲研究員分析 - if debate_state.get("bull_history"): - research_reports.append( - Panel( - Markdown(debate_state["bull_history"]), - title="看漲研究員", - border_style="blue", - padding=(1, 2), - ) - ) - - # 看跌研究員分析 - if debate_state.get("bear_history"): - research_reports.append( - Panel( - Markdown(debate_state["bear_history"]), - title="看跌研究員", - border_style="blue", - padding=(1, 2), - ) - ) - - # 研究經理決策 - if debate_state.get("judge_decision"): - research_reports.append( - Panel( - Markdown(debate_state["judge_decision"]), - title="研究經理", - border_style="blue", - padding=(1, 2), - ) - ) - - if research_reports: - console.print( - Panel( - Columns(research_reports, equal=True, expand=True), - title="II. 研究團隊決策", - border_style="magenta", - padding=(1, 2), - ) - ) - - # III. 交易團隊報告 - if final_state.get("trader_investment_plan"): - console.print( - Panel( - Panel( - Markdown(final_state["trader_investment_plan"]), - title="交易員", - border_style="blue", - padding=(1, 2), - ), - title="III. 交易團隊計畫", - border_style="yellow", - padding=(1, 2), - ) - ) - - # IV. 風險管理團隊報告 - if final_state.get("risk_debate_state"): - risk_reports = [] - risk_state = final_state["risk_debate_state"] - - # 激進（風險）分析師分析 - if risk_state.get("risky_history"): - risk_reports.append( - Panel( - Markdown(risk_state["risky_history"]), - title="激進分析師", - border_style="blue", - padding=(1, 2), - ) - ) - - # 保守（安全）分析師分析 - if risk_state.get("safe_history"): - risk_reports.append( - Panel( - Markdown(risk_state["safe_history"]), - title="保守分析師", - border_style="blue", - padding=(1, 2), - ) - ) - - # 中立分析師分析 - if risk_state.get("neutral_history"): - risk_reports.append( - Panel( - Markdown(risk_state["neutral_history"]), - title="中立分析師", - border_style="blue", - padding=(1, 2), - ) - ) - - if risk_reports: - console.print( - Panel( - Columns(risk_reports, equal=True, expand=True), - title="IV. 風險管理團隊決策", - border_style="red", - padding=(1, 2), - ) - ) - - # V. 投資組合經理決策 - if risk_state.get("judge_decision"): - console.print( - Panel( - Panel( - Markdown(risk_state["judge_decision"]), - title="投資組合經理", - border_style="blue", - padding=(1, 2), - ), - title="V. 投資組合經理決策", - border_style="green", - padding=(1, 2), - ) - ) - - -def update_research_team_status(status): - """更新所有研究團隊成員和交易員的狀態。""" - research_team = ["Bull Researcher", "Bear Researcher", "Research Manager", "Trader"] - for agent in research_team: - message_buffer.update_agent_status(agent, status) - -def extract_content_string(content): - """從各種訊息格式中提取字串內容。""" - if isinstance(content, str): - return content - elif isinstance(content, list): - # 處理 Anthropic 的列表格式 - text_parts = [] - for item in content: - if isinstance(item, dict): - if item.get('type') == 'text': - text_parts.append(item.get('text', '')) - elif item.get('type') == 'tool_use': - text_parts.append(f"[工具: {item.get('name', 'unknown')}]") - else: - text_parts.append(str(item)) - return ' '.join(text_parts) - else: - return str(content) - -def run_analysis(): - """執行完整的分析流程。""" - # 首先獲取所有使用者選擇 - selections = get_user_selections() - - # 使用選擇的研究深度建立設定 - config = DEFAULT_CONFIG.copy() - config["max_debate_rounds"] = selections["research_depth"] - config["max_risk_discuss_rounds"] = selections["research_depth"] - config["quick_think_llm"] = selections["shallow_thinker"] - config["deep_think_llm"] = selections["deep_thinker"] - config["backend_url"] = selections["backend_url"] - config["llm_provider"] = selections["llm_provider"].lower() - - # 添加 API Keys 到配置 - config["quick_think_api_key"] = selections["quick_think_api_key"] - config["deep_think_api_key"] = selections["deep_think_api_key"] - config["embedding_api_key"] = selections["embedding_api_key"] - config["embedding_base_url"] = selections["embedding_url"] - - # 設置環境變數（某些工具可能需要） - import os - os.environ["OPENAI_API_KEY"] = selections["quick_think_api_key"] - os.environ["ALPHA_VANTAGE_API_KEY"] = selections["alpha_vantage_api_key"] - - # 初始化圖 - graph = TradingAgentsXGraph( - [analyst.value for analyst in selections["analysts"]], config=config, debug=True - ) - - # 建立結果目錄 - results_dir = Path(config["results_dir"]) / selections["ticker"] / selections["analysis_date"] - results_dir.mkdir(parents=True, exist_ok=True) - report_dir = results_dir / "reports" - report_dir.mkdir(parents=True, exist_ok=True) - log_file = results_dir / "message_tool.log" - log_file.touch(exist_ok=True) - - # 裝飾器，用於儲存訊息 - def save_message_decorator(obj, func_name): - func = getattr(obj, func_name) - @wraps(func) - def wrapper(*args, **kwargs): - func(*args, **kwargs) - timestamp, message_type, content = obj.messages[-1] - content = content.replace("\n", " ") # 將換行符替換為空格 - with open(log_file, "a") as f: - f.write(f"{timestamp} [{message_type}] {content}\n") - return wrapper - - # 裝飾器，用於儲存工具呼叫 - def save_tool_call_decorator(obj, func_name): - func = getattr(obj, func_name) - @wraps(func) - def wrapper(*args, **kwargs): - func(*args, **kwargs) - timestamp, tool_name, args = obj.tool_calls[-1] - args_str = ", ".join(f"{k}={v}" for k, v in args.items()) - with open(log_file, "a") as f: - f.write(f"{timestamp} [工具呼叫] {tool_name}({args_str})\n") - return wrapper - - # 裝飾器，用於儲存報告區塊 - def save_report_section_decorator(obj, func_name): - func = getattr(obj, func_name) - @wraps(func) - def wrapper(section_name, content): - func(section_name, content) - if section_name in obj.report_sections and obj.report_sections[section_name] is not None: - content = obj.report_sections[section_name] - if content: - file_name = f"{section_name}.md" - with open(report_dir / file_name, "w") as f: - f.write(content) - return wrapper - - # 應用裝飾器 - message_buffer.add_message = save_message_decorator(message_buffer, "add_message") - message_buffer.add_tool_call = save_tool_call_decorator(message_buffer, "add_tool_call") - message_buffer.update_report_section = save_report_section_decorator(message_buffer, "update_report_section") - - # 現在開始顯示版面配置 - layout = create_layout() - - with Live(layout, refresh_per_second=4) as live: - # 初始顯示 - update_display(layout) - - # 新增初始訊息 - message_buffer.add_message("系統", f"選擇的股票代碼: {selections['ticker']}") - message_buffer.add_message( - "系統", f"分析日期: {selections['analysis_date']}" - ) - message_buffer.add_message( - "系統", - f"選擇的分析師: {', '.join(analyst.value for analyst in selections['analysts'])}", - ) - update_display(layout) - - # 重設代理狀態 - for agent in message_buffer.agent_status: - message_buffer.update_agent_status(agent, "pending") - - # 重設報告區塊 - for section in message_buffer.report_sections: - message_buffer.report_sections[section] = None - message_buffer.current_report = None - message_buffer.final_report = None - - # 將第一個分析師的代理狀態更新為進行中 - first_analyst = f"{selections['analysts'][0].value.capitalize()} Analyst" - message_buffer.update_agent_status(first_analyst, "in_progress") - update_display(layout) - - # 建立 spinner 文字 - spinner_text = ( - f"正在分析 {selections['ticker']} 於 {selections['analysis_date']}..." - ) - update_display(layout, spinner_text) - - # 初始化狀態並獲取圖參數 - init_agent_state = graph.propagator.create_initial_state( - selections["ticker"], selections["analysis_date"] - ) - args = graph.propagator.get_graph_args() - - # 串流分析 - trace = [] - for chunk in graph.graph.stream(init_agent_state, **args): - if len(chunk["messages"]) > 0: - # 獲取區塊中的最後一條訊息 - last_message = chunk["messages"][-1] - - # 提取訊息內容和類型 - if hasattr(last_message, "content"): - content = extract_content_string(last_message.content) # 使用輔助函式 - msg_type = "推理" - else: - content = str(last_message) - msg_type = "系統" - - # 將訊息新增到緩衝區 - message_buffer.add_message(msg_type, content) - - # 如果是工具呼叫，則將其新增到工具呼叫中 - if hasattr(last_message, "tool_calls"): - for tool_call in last_message.tool_calls: - # 處理字典和物件兩種工具呼叫格式 - if isinstance(tool_call, dict): - message_buffer.add_tool_call( - tool_call["name"], tool_call["args"] - ) - else: - message_buffer.add_tool_call(tool_call.name, tool_call.args) - - # 根據區塊內容更新報告和代理狀態 - # 分析師團隊報告 - if "market_report" in chunk and chunk["market_report"]: - message_buffer.update_report_section( - "market_report", chunk["market_report"] - ) - message_buffer.update_agent_status("Market Analyst", "completed") - # 將下一個分析師設定為進行中 - if "social" in selections["analysts"]: - message_buffer.update_agent_status( - "Social Analyst", "in_progress" - ) - - if "sentiment_report" in chunk and chunk["sentiment_report"]: - message_buffer.update_report_section( - "sentiment_report", chunk["sentiment_report"] - ) - message_buffer.update_agent_status("Social Analyst", "completed") - # 將下一個分析師設定為進行中 - if "news" in selections["analysts"]: - message_buffer.update_agent_status( - "News Analyst", "in_progress" - ) - - if "news_report" in chunk and chunk["news_report"]: - message_buffer.update_report_section( - "news_report", chunk["news_report"] - ) - message_buffer.update_agent_status("News Analyst", "completed") - # 將下一個分析師設定為進行中 - if "fundamentals" in selections["analysts"]: - message_buffer.update_agent_status( - "Fundamentals Analyst", "in_progress" - ) - - if "fundamentals_report" in chunk and chunk["fundamentals_report"]: - message_buffer.update_report_section( - "fundamentals_report", chunk["fundamentals_report"] - ) - message_buffer.update_agent_status( - "Fundamentals Analyst", "completed" - ) - # 將所有研究團隊成員設定為進行中 - update_research_team_status("in_progress") - - # 研究團隊 - 處理投資辯論狀態 - if ( - "investment_debate_state" in chunk - and chunk["investment_debate_state"] - ): - debate_state = chunk["investment_debate_state"] - - # 更新看漲研究員狀態和報告 - if "bull_history" in debate_state and debate_state["bull_history"]: - # 保持所有研究團隊成員為進行中 - update_research_team_status("in_progress") - # 提取最新的看漲回應 - bull_responses = debate_state["bull_history"].split("\n") - latest_bull = bull_responses[-1] if bull_responses else "" - if latest_bull: - message_buffer.add_message("推理", latest_bull) - # 使用看漲研究員的最新分析更新研究報告 - message_buffer.update_report_section( - "investment_plan", - f"### 看漲研究員分析\n{latest_bull}", - ) - - # 更新看跌研究員狀態和報告 - if "bear_history" in debate_state and debate_state["bear_history"]: - # 保持所有研究團隊成員為進行中 - update_research_team_status("in_progress") - # 提取最新的看跌回應 - bear_responses = debate_state["bear_history"].split("\n") - latest_bear = bear_responses[-1] if bear_responses else "" - if latest_bear: - message_buffer.add_message("推理", latest_bear) - # 使用看跌研究員的最新分析更新研究報告 - message_buffer.update_report_section( - "investment_plan", - f"{message_buffer.report_sections['investment_plan']}\n\n### 看跌研究員分析\n{latest_bear}", - ) - - # 更新研究經理狀態和最終決策 - if ( - "judge_decision" in debate_state - and debate_state["judge_decision"] - ): - # 在最終決策前保持所有研究團隊成員為進行中 - update_research_team_status("in_progress") - message_buffer.add_message( - "推理", - f"研究經理: {debate_state['judge_decision']}", - ) - # 使用最終決策更新研究報告 - message_buffer.update_report_section( - "investment_plan", - f"{message_buffer.report_sections['investment_plan']}\n\n### 研究經理決策\n{debate_state['judge_decision']}", - ) - # 將所有研究團隊成員標記為已完成 - update_research_team_status("completed") - # 將第一個風險分析師設定為進行中 - message_buffer.update_agent_status( - "Risky Analyst", "in_progress" - ) - - # 交易團隊 - if ( - "trader_investment_plan" in chunk - and chunk["trader_investment_plan"] - ): - message_buffer.update_report_section( - "trader_investment_plan", chunk["trader_investment_plan"] - ) - # 將第一個風險分析師設定為進行中 - message_buffer.update_agent_status("Risky Analyst", "in_progress") - - # 風險管理團隊 - 處理風險辯論狀態 - if "risk_debate_state" in chunk and chunk["risk_debate_state"]: - risk_state = chunk["risk_debate_state"] - - # 更新風險分析師狀態和報告 - if ( - "current_risky_response" in risk_state - and risk_state["current_risky_response"] - ): - message_buffer.update_agent_status( - "Risky Analyst", "in_progress" - ) - message_buffer.add_message( - "推理", - f"風險分析師: {risk_state['current_risky_response']}", - ) - # 僅使用風險分析師的最新分析更新風險報告 - message_buffer.update_report_section( - "final_trade_decision", - f"### 風險分析師分析\n{risk_state['current_risky_response']}", - ) - - # 更新安全分析師狀態和報告 - if ( - "current_safe_response" in risk_state - and risk_state["current_safe_response"] - ): - message_buffer.update_agent_status( - "Safe Analyst", "in_progress" - ) - message_buffer.add_message( - "推理", - f"安全分析師: {risk_state['current_safe_response']}", - ) - # 僅使用安全分析師的最新分析更新風險報告 - message_buffer.update_report_section( - "final_trade_decision", - f"### 安全分析師分析\n{risk_state['current_safe_response']}", - ) - - # 更新中立分析師狀態和報告 - if ( - "current_neutral_response" in risk_state - and risk_state["current_neutral_response"] - ): - message_buffer.update_agent_status( - "Neutral Analyst", "in_progress" - ) - message_buffer.add_message( - "推理", - f"中立分析師: {risk_state['current_neutral_response']}", - ) - # 僅使用中立分析師的最新分析更新風險報告 - message_buffer.update_report_section( - "final_trade_decision", - f"### 中立分析師分析\n{risk_state['current_neutral_response']}", - ) - - # 更新投資組合經理狀態和最終決策 - if "judge_decision" in risk_state and risk_state["judge_decision"]: - message_buffer.update_agent_status( - "Portfolio Manager", "in_progress" - ) - message_buffer.add_message( - "推理", - f"投資組合經理: {risk_state['judge_decision']}", - ) - # 僅使用最終決策更新風險報告 - message_buffer.update_report_section( - "final_trade_decision", - f"### 投資組合經理決策\n{risk_state['judge_decision']}", - ) - # 將風險分析師標記為已完成 - message_buffer.update_agent_status("Risky Analyst", "completed") - message_buffer.update_agent_status("Safe Analyst", "completed") - message_buffer.update_agent_status( - "Neutral Analyst", "completed" - ) - message_buffer.update_agent_status( - "Portfolio Manager", "completed" - ) - - # 更新顯示 - update_display(layout) - - trace.append(chunk) - - # 獲取最終狀態和決策 - final_state = trace[-1] - decision = graph.process_signal(final_state["final_trade_decision"]) - - # 將所有代理狀態更新為已完成 - for agent in message_buffer.agent_status: - message_buffer.update_agent_status(agent, "completed") - - message_buffer.add_message( - "分析", f"已完成 {selections['analysis_date']} 的分析" - ) - - # 更新最終報告區塊 - for section in message_buffer.report_sections.keys(): - if section in final_state: - message_buffer.update_report_section(section, final_state[section]) - - # 顯示完整的最終報告 - display_complete_report(final_state) - - update_display(layout) - - -@app.command() -def analyze(): - """ - 執行分析。 - """ - run_analysis() - - -if __name__ == "__main__": - app() \ No newline at end of file diff --git a/build/lib/cli/models.py b/build/lib/cli/models.py deleted file mode 100644 index 92516cb8..00000000 --- a/build/lib/cli/models.py +++ /dev/null @@ -1,24 +0,0 @@ -# 匯入 Enum 模組，用於建立列舉類型 -from enum import Enum -# 匯入 List, Optional, Dict 類型提示，用於更清晰地定義資料結構 -from typing import List, Optional, Dict -# 匯入 BaseModel，用於建立資料模型 -from pydantic import BaseModel - - -# 定義分析師類型的列舉 -class AnalystType(str, Enum): - """ - AnalystType 是一個列舉 (Enum)，定義了不同類型的分析師。 - 這有助於標準化和限制分析師的角色，確保程式碼的一致性和可讀性。 - - 屬性: - MARKET (str): 市場分析師，專注於市場趨勢和價格行為。 - SOCIAL (str): 社交媒體分析師，監控和分析社交媒體上的情緒和討論。 - NEWS (str): 新聞分析師，分析新聞事件對市場的影響。 - FUNDAMENTALS (str): 基本面分析師，研究公司的財務狀況和健康狀況。 - """ - MARKET = "market" # 市場分析師 - SOCIAL = "social" # 社交媒體分析師 - NEWS = "news" # 新聞分析師 - FUNDAMENTALS = "fundamentals" # 基本面分析師 \ No newline at end of file diff --git a/build/lib/cli/utils.py b/build/lib/cli/utils.py deleted file mode 100644 index cb70e469..00000000 --- a/build/lib/cli/utils.py +++ /dev/null @@ -1,609 +0,0 @@ -# 匯入 questionary 套件，用於建立互動式命令列提示 -import questionary -# 匯入類型提示，用於更清晰地定義函式簽名 -from typing import List, Optional, Tuple, Dict -# 匯入 rich.console 用於美化輸出 -from rich.console import Console - -# 從 cli.models 模組匯入 AnalystType 列舉 -from cli.models import AnalystType - -# 初始化 console -console = Console() - -# 定義分析師的順序和對應的類型 -ANALYST_ORDER = [ - ("市場分析師", AnalystType.MARKET), - ("社群媒體分析師", AnalystType.SOCIAL), - ("新聞分析師", AnalystType.NEWS), - ("基本面分析師", AnalystType.FUNDAMENTALS), -] - - -def get_ticker() -> str: - """ - 提示使用者輸入股票代碼。 - - 返回: - str: 使用者輸入的股票代碼，已轉換為大寫並去除頭尾空格。 - """ - ticker = questionary.text( - "請輸入要分析的股票代碼：", - # 驗證輸入是否為空 - validate=lambda x: len(x.strip()) > 0 or "請輸入有效的股票代碼。", - # 設定提示的樣式 - style=questionary.Style( - [ - ("text", "fg:green"), - ("highlighted", "noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有輸入，則退出程式 - if not ticker: - console.print("\n[red]未提供股票代碼。正在結束程式...[/red]") - exit(1) - - # 返回處理過的股票代碼 - return ticker.strip().upper() - - -def get_analysis_date() -> str: - """ - 提示使用者輸入 YYYY-MM-DD 格式的日期。 - - 返回: - str: 使用者輸入的日期字串。 - """ - import re - from datetime import datetime - - def validate_date(date_str: str) -> bool: - """驗證日期字串是否為 YYYY-MM-DD 格式""" - # 使用正規表示式檢查格式 - if not re.match(r"^\d{4}-\d{2}-\d{2}$", date_str): - return False - try: - # 嘗試將字串解析為日期物件 - datetime.strptime(date_str, "%Y-%m-%d") - return True - except ValueError: - return False - - date = questionary.text( - "請輸入分析日期 (YYYY-MM-DD)：", - # 驗證日期格式是否正確 - validate=lambda x: validate_date(x.strip()) - or "請輸入有效的 YYYY-MM-DD 格式日期。", - # 設定提示的樣式 - style=questionary.Style( - [ - ("text", "fg:green"), - ("highlighted", "noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有輸入，則退出程式 - if not date: - console.print("\n[red]未提供日期。正在結束程式...[/red]") - exit(1) - - # 返回處理過的日期字串 - return date.strip() - - -def select_analysts() -> List[AnalystType]: - """ - 使用互動式核取方塊選擇分析師。 - - 返回: - List[AnalystType]: 使用者選擇的分析師類型列表。 - """ - choices = questionary.checkbox( - "選擇您的 [分析師團隊]：", - # 設定可選項 - choices=[ - questionary.Choice(display, value=value) for display, value in ANALYST_ORDER - ], - # 提供操作說明 - instruction="\n- 按下空白鍵選擇/取消選擇分析師\n- 按下 'a' 鍵選擇/取消選擇所有\n- 完成後按下 Enter 鍵", - # 驗證至少選擇一位分析師 - validate=lambda x: len(x) > 0 or "您必須至少選擇一位分析師。", - # 設定提示的樣式 - style=questionary.Style( - [ - ("checkbox-selected", "fg:green"), - ("selected", "fg:green noinherit"), - ("highlighted", "noinherit"), - ("pointer", "noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有選擇，則退出程式 - if not choices: - console.print("\n[red]未選擇任何分析師。正在結束程式...[/red]") - exit(1) - - # 返回選擇的分析師列表 - return choices - - -def select_research_depth() -> int: - """ - 使用互動式選單選擇研究深度。 - - 返回: - int: 代表研究深度的整數。 - """ - - # 定義研究深度的選項及其對應值 - DEPTH_OPTIONS = [ - ("淺層 - 快速研究，較少的辯論和策略討論", 1), - ("中等 - 中等程度，適度的辯論和策略討論", 3), - ("深層 - 全面研究，深入的辯論和策略討論", 5), - ] - - choice = questionary.select( - "選擇您的 [研究深度]：", - # 設定可選項 - choices=[ - questionary.Choice(display, value=value) for display, value in DEPTH_OPTIONS - ], - # 提供操作說明 - instruction="\n- 使用方向鍵導覽\n- 按下 Enter 鍵選擇", - # 設定提示的樣式 - style=questionary.Style( - [ - ("selected", "fg:yellow noinherit"), - ("highlighted", "fg:yellow noinherit"), - ("pointer", "fg:yellow noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有選擇，則退出程式 - if choice is None: - console.print("\n[red]未選擇研究深度。正在結束程式...[/red]") - exit(1) - - # 返回選擇的研究深度 - return choice - - -def select_shallow_thinking_agent(provider=None) -> str: - """ - 使用互動式選單選擇淺層思維的 LLM 引擎。 - - 參數: - provider (str, optional): LLM 供應商的名稱（已廢棄，不再使用）。 - - 返回: - str: 選擇的 LLM 模型的名稱。 - """ - - # 定義不同供應商的淺層思維 LLM 引擎選項 - SHALLOW_AGENT_OPTIONS = { - "OpenAI": [ - ("GPT-5.1", "gpt-5.1-2025-11-13"), - ("GPT-5-mini","gpt-5-mini-2025-08-07"), - ("GPT-5-nano","gpt-5-nano-2025-08-07"), - ("GPT-4.1-mini", "gpt-4.1-mini"), - ("GPT-4.1-nano", "gpt-4.1-nano"), - ("o4-mini", "o4-mini-2025-04-16"), - ], - "Anthropic": [ - ("Claude Haiku 4.5", "claude-haiku-4-5-20251001"), - ("Claude Sonnet 4.5", "claude-sonnet-4-5-20250929"), - ("Claude Sonnet 4", "claude-sonnet-4-0"), - ("Claude Haiku 3.5", "claude-3-5-haiku-20241022"), - ("Claude Haiku 3", "claude-3-haiku-20240307"), - ], - "Google": [ - ("Gemini 2.5 Pro", "gemini-2.5-pro"), - ("Gemini 2.5 Flash", "gemini-2.5-flash"), - ("Gemini 2.5 Flash Lite", "gemini-2.5-flash-lite"), - ("Gemini 2.0 Flash", "gemini-2.0-flash"), - ("Gemini 2.0 Flash-Lite", "gemini-2.0-flash-lite"), - ], - "Grok":[ - ("Grok 4.1 Fast Reasoning","grok-4-1-fast-reasoning"), - ("Grok 4.1 Fast Non Reasoning","grok-4-1-fast-non-reasoning"), - ("Grok 4 Fast Reasoning","grok-4-fast-reasoning"), - ("Grok 4 Fast Non Reasoning","grok-4-fast-non-reasoning"), - ("Grok 4","grok-4-0709"), - ("Grok 3","grok-3"), - ("Grok 3 Mini","grok-3-mini"), - ], - "DeepSeek": [ - ("DeepSeek Reasoner","deepseek-reasoner"), - ("DeepSeek Chat","deepseek-chat"), - ], - "Qwen":[ - ("Qwen 3 Max", "qwen3-max"), - ("Qwen Plus", "qwen-plus"), - ("Qwen Flash", "qwen-flash"), - ] - } - - # 第一步：選擇供應商 - provider_choice = questionary.select( - "選擇 [快速思維] 模型供應商：", - choices=[ - questionary.Choice(provider_name, value=provider_name) - for provider_name in SHALLOW_AGENT_OPTIONS.keys() - ], - instruction="\n- 使用方向鍵導覽\n- 按下 Enter 鍵選擇", - style=questionary.Style( - [ - ("selected", "fg:cyan noinherit"), - ("highlighted", "fg:cyan noinherit"), - ("pointer", "fg:cyan noinherit"), - ] - ), - ).ask() - - if provider_choice is None: - console.print("\n[red]未選擇供應商。正在結束程式...[/red]") - exit(1) - - # 第二步：根據選擇的供應商顯示模型列表 - model_choice = questionary.select( - f"選擇 [{provider_choice}] 的快速思維模型：", - choices=[ - questionary.Choice(display, value=value) - for display, value in SHALLOW_AGENT_OPTIONS[provider_choice] - ], - instruction="\n- 使用方向鍵導覽\n- 按下 Enter 鍵選擇", - style=questionary.Style( - [ - ("selected", "fg:magenta noinherit"), - ("highlighted", "fg:magenta noinherit"), - ("pointer", "fg:magenta noinherit"), - ] - ), - ).ask() - - if model_choice is None: - console.print( - "\n[red]未選擇快速思維 LLM 引擎。正在結束程式...[/red]" - ) - exit(1) - - # 如果選擇自訂，提示輸入模型名稱 - if model_choice == "custom": - model_name = questionary.text( - "請輸入快速思維 LLM 模型名稱：", - validate=lambda x: len(x.strip()) > 0 or "請輸入有效的模型名稱。", - style=questionary.Style( - [ - ("text", "fg:green"), - ("highlighted", "noinherit"), - ] - ), - ).ask() - - if not model_name: - console.print( - "\n[red]未提供模型名稱。正在結束程式...[/red]" - ) - exit(1) - - return model_name.strip() - - # 返回選擇的 LLM 模型 - return model_choice - - -def select_deep_thinking_agent(provider=None) -> str: - """ - 使用互動式選單選擇深層思維的 LLM 引擎。 - - 參數: - provider (str, optional): LLM 供應商的名稱（已廢棄，不再使用）。 - - 返回: - str: 選擇的 LLM 模型的名稱。 - """ - - # 定義不同供應商的深層思維 LLM 引擎選項 - DEEP_AGENT_OPTIONS = { - "OpenAI": [ - ("GPT-5.1", "gpt-5.1-2025-11-13"), - ("GPT-5-mini","gpt-5-mini-2025-08-07"), - ("GPT-5-nano","gpt-5-nano-2025-08-07"), - ("GPT-4.1-mini", "gpt-4.1-mini"), - ("GPT-4.1-nano", "gpt-4.1-nano"), - ("o4-mini", "o4-mini-2025-04-16"), - ], - "Anthropic": [ - ("Claude Haiku 4.5", "claude-haiku-4-5-20251001"), - ("Claude Sonnet 4.5", "claude-sonnet-4-5-20250929"), - ("Claude Sonnet 4", "claude-sonnet-4-0"), - ("Claude Haiku 3.5", "claude-3-5-haiku-20241022"), - ("Claude Haiku 3", "claude-3-haiku-20240307"), - ], - "Google": [ - ("Gemini 2.5 Pro", "gemini-2.5-pro"), - ("Gemini 2.5 Flash", "gemini-2.5-flash"), - ("Gemini 2.5 Flash Lite", "gemini-2.5-flash-lite"), - ("Gemini 2.0 Flash", "gemini-2.0-flash"), - ("Gemini 2.0 Flash-Lite", "gemini-2.0-flash-lite"), - ], - "Grok":[ - ("Grok 4.1 Fast Reasoning","grok-4-1-fast-reasoning"), - ("Grok 4.1 Fast Non Reasoning","grok-4-1-fast-non-reasoning"), - ("Grok 4 Fast Reasoning","grok-4-fast-reasoning"), - ("Grok 4 Fast Non Reasoning","grok-4-fast-non-reasoning"), - ("Grok 4","grok-4-0709"), - ("Grok 3","grok-3"), - ("Grok 3 Mini","grok-3-mini"), - ], - "DeepSeek":[ - ("DeepSeek Reasoner","deepseek-reasoner"), - ("DeepSeek Chat","deepseek-chat"), - ], - "Qwen":[ - ("Qwen 3 Max", "qwen3-max"), - ("Qwen Plus", "qwen-plus"), - ("Qwen Flash", "qwen-flash"), - ] - } - - # 第一步：選擇供應商 - provider_choice = questionary.select( - "選擇 [深度思維] 模型供應商：", - choices=[ - questionary.Choice(provider_name, value=provider_name) - for provider_name in DEEP_AGENT_OPTIONS.keys() - ], - instruction="\n- 使用方向鍵導覽\n- 按下 Enter 鍵選擇", - style=questionary.Style( - [ - ("selected", "fg:cyan noinherit"), - ("highlighted", "fg:cyan noinherit"), - ("pointer", "fg:cyan noinherit"), - ] - ), - ).ask() - - if provider_choice is None: - console.print("\n[red]未選擇供應商。正在結束程式...[/red]") - exit(1) - - # 第二步：根據選擇的供應商顯示模型列表 - model_choice = questionary.select( - f"選擇 [{provider_choice}] 的深度思維模型：", - choices=[ - questionary.Choice(display, value=value) - for display, value in DEEP_AGENT_OPTIONS[provider_choice] - ], - instruction="\n- 使用方向鍵導覽\n- 按下 Enter 鍵選擇", - style=questionary.Style( - [ - ("selected", "fg:magenta noinherit"), - ("highlighted", "fg:magenta noinherit"), - ("pointer", "fg:magenta noinherit"), - ] - ), - ).ask() - - if model_choice is None: - console.print("\n[red]未選擇深度思維 LLM 引擎。正在結束程式...[/red]") - exit(1) - - # 如果選擇自訂，提示輸入模型名稱 - if model_choice == "custom": - model_name = questionary.text( - "請輸入深度思維 LLM 模型名稱：", - validate=lambda x: len(x.strip()) > 0 or "請輸入有效的模型名稱。", - style=questionary.Style( - [ - ("text", "fg:green"), - ("highlighted", "noinherit"), - ] - ), - ).ask() - - if not model_name: - console.print( - "\n[red]未提供模型名稱。正在結束程式...[/red]" - ) - exit(1) - - return model_name.strip() - - # 返回選擇的 LLM 模型 - return model_choice - -def select_llm_provider() -> tuple[str, str]: - """ - 使用互動式選單選擇 LLM 供應商。 - - 返回: - tuple[str, str]: 包含供應商顯示名稱和 API 基礎 URL 的元組。 - """ - # 定義 LLM 供應商及其 API 基礎 URL - BASE_URLS = [ - ("OpenAI", "https://api.openai.com/v1"), - ("Anthropic", "https://api.anthropic.com/v1"), - ("Google", "https://generativelanguage.googleapis.com/v1beta/openai"), - ("Grok", "https://api.x.ai/v1"), - ("DeepSeek", "https://api.deepseek.com/v1"), - ("Qwen", "https://dashscope-intl.aliyuncs.com/compatible-mode/v1"), - ("自訂 URL", "custom") # 新增自訂 URL 選項 - ] - - choice = questionary.select( - "選擇您的 LLM 供應商：", - # 設定可選項 - choices=[ - questionary.Choice(display, value=(display, value)) - for display, value in BASE_URLS - ], - # 提供操作說明 - instruction="\n- 使用方向鍵導覽\n- 按下 Enter 鍵選擇", - # 設定提示的樣式 - style=questionary.Style( - [ - ("selected", "fg:magenta noinherit"), - ("highlighted", "fg:magenta noinherit"), - ("pointer", "fg:magenta noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有選擇，則退出程式 - if choice is None: - console.print("\n[red]未選擇 LLM 後端。正在結束程式...[/red]") - exit(1) - - # 解構選擇的元組 - display_name, url = choice - - # 如果使用者選擇自訂 URL，提示輸入 - if url == "custom": - custom_url = questionary.text( - "請輸入自訂的 Base URL：", - # 驗證 URL 格式 - validate=lambda x: (x.strip().startswith("http://") or x.strip().startswith("https://")) - or "請輸入有效的 URL（必須以 http:// 或 https:// 開頭）", - # 設定提示的樣式 - style=questionary.Style( - [ - ("text", "fg:green"), - ("highlighted", "noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有輸入，則退出程式 - if not custom_url: - console.print("\n[red]未提供 Base URL。正在結束程式...[/red]") - exit(1) - - url = custom_url.strip() - display_name = "自訂供應商" - - # 印出使用者的選擇 - print(f"您選擇了：{display_name}\tURL: {url}") - - # 返回供應商名稱和 URL - return display_name, url - - -def select_embedding_provider() -> tuple[str, str]: - """ - 使用互動式選單選擇嵌入模型供應商。 - - 返回: - tuple[str, str]: 包含供應商名稱和 API 基礎 URL 的元組。 - """ - # 定義嵌入模型供應商（只有 OpenAI 和自訂） - EMBEDDING_PROVIDERS = [ - ("OpenAI", "https://api.openai.com/v1"), - ("自訂 URL", "custom") - ] - - choice = questionary.select( - "選擇您的嵌入模型供應商：", - # 設定可選項 - choices=[ - questionary.Choice(display, value=(display, value)) - for display, value in EMBEDDING_PROVIDERS - ], - # 提供操作說明 - instruction="\n- 使用方向鍵導覽\n- 按下 Enter 鍵選擇", - # 設定提示的樣式 - style=questionary.Style( - [ - ("selected", "fg:cyan noinherit"), - ("highlighted", "fg:cyan noinherit"), - ("pointer", "fg:cyan noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有選擇，則退出程式 - if choice is None: - console.print("\n[red]未選擇嵌入模型供應商。正在結束程式...[/red]") - exit(1) - - # 解構選擇的元組 - display_name, url = choice - - # 如果選擇自訂 URL，提示使用者輸入 - if url == "custom": - custom_url = questionary.text( - "請輸入自訂的 Base URL：", - validate=lambda x: (x.startswith("http://") or x.startswith("https://")) or "URL 必須以 http:// 或 https:// 開頭", - style=questionary.Style( - [ - ("text", "fg:green"), - ("highlighted", "noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有輸入，則退出程式 - if not custom_url: - console.print("\n[red]未提供 Base URL。正在結束程式...[/red]") - exit(1) - - url = custom_url.strip() - display_name = "自訂供應商" - - # 印出使用者的選擇 - print(f"您選擇了嵌入模型：{display_name}\tURL: {url}") - - # 返回供應商名稱和 URL - return display_name, url - - -def get_api_key(model_type: str, default_key: Optional[str] = None) -> str: - """ - 提示使用者輸入 API Key，如果留空則使用預設值。 - - 參數: - model_type (str): 模型類型（例如：「快速思維」、「深度思維」、「嵌入模型」） - default_key (Optional[str]): 從 .env 文件讀取的預設 API Key - - 返回: - str: 使用者輸入的 API Key 或預設值 - """ - import os - from rich.console import Console - - console = Console() - - # 顯示提示訊息 - if default_key: - hint = f"[dim]（留空使用 .env 中的 API Key: {default_key[:10]}...{default_key[-4:]}）[/dim]" - else: - hint = "[dim]（必填）[/dim]" - - console.print(f"\n[cyan]{model_type} API Key {hint}[/cyan]") - - api_key = questionary.password( - f"請輸入 {model_type} 的 API Key：", - style=questionary.Style( - [ - ("text", "fg:green"), - ("highlighted", "noinherit"), - ] - ), - ).ask() - - # 如果使用者沒有輸入，使用預設值 - if not api_key or api_key.strip() == "": - if default_key: - console.print(f"[green]✓ 使用 .env 中的 API Key[/green]") - return default_key - else: - console.print(f"\n[red]未提供 {model_type} API Key。正在結束程式...[/red]") - exit(1) - - console.print(f"[green]✓ API Key 已設定[/green]") - return api_key.strip() \ No newline at end of file