Cortex Core Framework 規格
概述
Core Framework 是 Cortex 的核心可複用層，設計為 domain-agnostic，可支援投資分析、健康管理等不同領域的 AI Agent。
設計原則: - 高內聚、低耦合 - 可測試性優先 - 配置驅動 - 可擴展性
模組架構
core/
├── llm/ # LLM 整合層
├── agent_framework/ # Agent 框架
├── knowledge/ # 知識管理
├── utils/ # 工具函數
└── models/ # 資料模型
1. LLM 模組 (core/llm/)
1.1 BaseLLM (抽象基類)
檔案: core/llm/base.py
用途: 定義所有 LLM 提供商的統一介面
類別定義:
class BaseLLM(ABC):
"""LLM 抽象基類，支援多種 LLM 提供商"""
@abstractmethod
async def generate(
self,
prompt: str,
system_prompt: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2000,
**kwargs
) -> str:
"""
生成文字回應
Args:
prompt: 用戶提示
system_prompt: 系統提示（可選）
temperature: 溫度參數 (0-1)
max_tokens: 最大 token 數
**kwargs: 其他 LLM 特定參數
Returns:
生成的文字回應
Raises:
LLMError: LLM API 調用失敗
"""
pass
@abstractmethod
async def generate_stream(
self,
prompt: str,
system_prompt: Optional[str] = None,
temperature: float = 0.7,
max_tokens: int = 2000,
**kwargs
) -> AsyncIterator[str]:
"""
串流生成文字回應
Yields:
逐步生成的文字片段
"""
pass
@abstractmethod
async def generate_structured(
self,
prompt: str,
schema: Dict[str, Any],
system_prompt: Optional[str] = None,
**kwargs
) -> Dict[str, Any]:
"""
生成結構化輸出 (JSON)
Args:
schema: 預期的 JSON schema
Returns:
符合 schema 的 JSON 物件
"""
pass
錯誤處理:
class LLMError(Exception):
"""LLM 基礎錯誤"""
pass
class RateLimitError(LLMError):
"""API 速率限制錯誤"""
pass
class InvalidResponseError(LLMError):
"""無效回應錯誤"""
pass
1.2 GeminiLLM (Gemini 實現)
檔案: core/llm/gemini.py
類別定義:
class GeminiLLM(BaseLLM):
"""Google Gemini API 實現"""
def __init__(
self,
api_key: str,
model_name: str = "gemini-1.5-flash", # 或 gemini-1.5-pro
default_temperature: float = 0.7,
max_retries: int = 3
):
"""
初始化 Gemini LLM
Args:
api_key: Google API Key
model_name: 模型名稱
default_temperature: 預設溫度
max_retries: 最大重試次數
"""
pass
實現要求:
- 使用 google-generativeai SDK
- 實現 exponential backoff 重試機制
- 處理 rate limit (每分鐘請求數限制)
- 支援 function calling (用於結構化輸出)
- 錯誤日誌記錄
配置範例:
# config/llm.yaml
gemini:
api_key: ${GOOGLE_API_KEY}
models:
fast: gemini-1.5-flash
pro: gemini-1.5-pro
rate_limits:
requests_per_minute: 60
retry:
max_attempts: 3
backoff_factor: 2
1.3 PromptManager
檔案: core/llm/prompt_manager.py
用途: 管理和渲染 Prompt 模板
類別定義:
class PromptManager:
"""Prompt 模板管理器"""
def __init__(self, templates_dir: str):
"""
初始化 Prompt Manager
Args:
templates_dir: 模板目錄路徑
"""
self.templates: Dict[str, Template] = {}
self.load_templates(templates_dir)
def load_templates(self, templates_dir: str) -> None:
"""載入所有 .txt 模板檔案"""
pass
def render(
self,
template_name: str,
**variables
) -> str:
"""
渲染模板
Args:
template_name: 模板名稱
**variables: 模板變數
Returns:
渲染後的 prompt
Example:
>>> pm = PromptManager("prompts/")
>>> pm.render(
... "analyze_stock",
... ticker="2330.TW",
... data=stock_data
... )
"""
pass
def validate_template(
self,
template_name: str,
required_vars: List[str]
) -> bool:
"""驗證模板是否包含必要變數"""
pass
模板格式 (Jinja2):
{# prompts/analyze_stock.txt #}
You are a professional stock analyst.
Analyze the following stock:
Ticker: {{ ticker }}
Current Price: {{ current_price }}
Historical Data:
{{ historical_data }}
Technical Indicators:
{% for indicator, value in indicators.items() %}
- {{ indicator }}: {{ value }}
{% endfor %}
Provide a comprehensive analysis covering:
1. Trend analysis
2. Support/Resistance levels
3. Risk assessment
2. Agent Framework 模組 (core/agent_framework/)
2.1 BaseAgent
檔案: core/agent_framework/base_agent.py
類別定義:
class BaseAgent(ABC):
"""所有 Agent 的基類"""
def __init__(
self,
name: str,
llm: BaseLLM,
config: Dict[str, Any]
):
"""
初始化 Agent
Args:
name: Agent 名稱
llm: LLM 實例
config: Agent 配置
"""
self.name = name
self.llm = llm
self.config = config
self.logger = get_logger(f"agent.{name}")
@abstractmethod
async def execute(
self,
task: AgentTask
) -> AgentResult:
"""
執行 Agent 任務
Args:
task: Agent 任務物件
Returns:
執行結果
"""
pass
async def _log_execution(
self,
task: AgentTask,
result: AgentResult,
duration_ms: int
) -> None:
"""記錄執行歷史到資料庫"""
pass
2.2 MetaAgent
檔案: core/agent_framework/meta_agent.py
用途: 核心決策中樞，動態生成 VRR 策略
類別定義:
class MetaAgent(BaseAgent):
"""Meta-Agent: 策略生成中樞"""
async def generate_strategy(
self,
task_description: str,
domain: str,
context: Optional[Dict[str, Any]] = None
) -> VRRStrategy:
"""
生成 VRR 策略
Args:
task_description: 任務描述
domain: 領域 (investment, health, etc.)
context: 額外上下文
Returns:
VRRStrategy 物件
Example:
>>> meta = MetaAgent(llm=gemini)
>>> strategy = await meta.generate_strategy(
... task_description="分析台積電技術面",
... domain="investment"
... )
>>> print(strategy.verify_criteria)
["RSI 數值範圍 0-100", "MACD 交叉點識別正確", ...]
"""
pass
async def optimize_strategy(
self,
base_strategy: VRRStrategy,
historical_failures: List[AgentExecution]
) -> VRRStrategy:
"""
基於歷史失敗案例優化策略
Args:
base_strategy: 原始策略
historical_failures: 歷史失敗執行記錄
Returns:
優化後的策略
"""
pass
Prompt 設計:
STRATEGY_GENERATION_PROMPT = """
You are a Meta-Agent that generates verification strategies for AI agents.
Task: {task_description}
Domain: {domain}
Context: {context}
Generate a Verify-Reflect-Refine (VRR) strategy in JSON format:
{{
"verify_criteria": [
"具體可檢查的驗證點1",
"具體可檢查的驗證點2",
...
],
"reflect_focus": [
"如果驗證失敗，應該反思的面向1",
"如果驗證失敗，應該反思的面向2",
...
],
"refine_priority": [
"改進的優先順序1",
"改進的優先順序2",
...
]
}}
Requirements:
1. verify_criteria must be specific and checkable
2. Each criterion should be independent
3. Focus on domain-specific concerns
4. Prioritize common failure modes
"""
2.3 VRREngine
檔案: core/agent_framework/vrr_engine.py
用途: 執行 Verify-Reflect-Refine 循環
類別定義:
class VRREngine:
"""VRR 執行引擎"""
def __init__(
self,
llm: BaseLLM,
max_iterations: int = 3
):
"""
初始化 VRR Engine
Args:
llm: LLM 實例
max_iterations: 最大迭代次數
"""
self.llm = llm
self.max_iterations = max_iterations
async def execute(
self,
initial_result: str,
strategy: VRRStrategy,
task_context: Dict[str, Any]
) -> VRRResult:
"""
執行完整 VRR 流程
Args:
initial_result: 初始 Agent 輸出
strategy: VRR 策略
task_context: 任務上下文
Returns:
VRRResult 包含最終結果和執行歷史
Flow:
1. Verify: 檢查 initial_result
2. [If passed] Return result
3. [If failed] Reflect: 分析問題
4. Refine: 改進結果
5. 回到步驟 1 (最多 max_iterations 次)
"""
pass
async def _verify(
self,
result: str,
criteria: List[str],
context: Dict[str, Any]
) -> VerificationResult:
"""
驗證結果
Returns:
VerificationResult:
- passed: bool
- failed_criteria: List[str]
- details: Dict[str, Any]
"""
pass
async def _reflect(
self,
result: str,
verification: VerificationResult,
focus_areas: List[str]
) -> ReflectionResult:
"""
反思問題
Returns:
ReflectionResult:
- identified_issues: List[str]
- root_causes: List[str]
- improvement_suggestions: List[str]
"""
pass
async def _refine(
self,
original_result: str,
reflection: ReflectionResult,
priorities: List[str]
) -> str:
"""
改進結果
Returns:
改進後的結果
"""
pass
2.4 Coordinator (LangGraph)
檔案: core/agent_framework/coordinator.py
用途: 使用 LangGraph 編排 multi-agent 工作流程
概念:
from langgraph.graph import StateGraph, END
class AgentCoordinator:
"""LangGraph 協調器"""
def __init__(self, agents: Dict[str, BaseAgent]):
"""
初始化協調器
Args:
agents: Agent 字典 {name: agent_instance}
"""
self.agents = agents
self.graph = self._build_graph()
def _build_graph(self) -> StateGraph:
"""
建立 LangGraph 工作流程
Returns:
StateGraph 實例
"""
# 在 INVESTMENT_AGENT_SPEC.md 中詳細定義具體工作流程
pass
async def execute(
self,
initial_state: Dict[str, Any]
) -> Dict[str, Any]:
"""
執行 Agent 工作流程
Args:
initial_state: 初始狀態
Returns:
最終狀態
"""
pass
3. Knowledge 模組 (core/knowledge/)
3.1 VectorStore (Chroma 抽象層)
檔案: core/knowledge/vector_store.py
類別定義:
class VectorStore(ABC):
"""向量資料庫抽象介面"""
@abstractmethod
async def add_documents(
self,
documents: List[Document],
embeddings: Optional[List[List[float]]] = None
) -> List[str]:
"""
新增文件到向量庫
Args:
documents: 文件列表
embeddings: 預先計算的 embeddings（可選）
Returns:
文件 IDs
"""
pass
@abstractmethod
async def search(
self,
query: str,
k: int = 5,
filter: Optional[Dict[str, Any]] = None
) -> List[SearchResult]:
"""
相似度搜尋
Args:
query: 查詢字串
k: 返回結果數量
filter: 過濾條件
Returns:
SearchResult 列表
"""
pass
class ChromaVectorStore(VectorStore):
"""Chroma 實現"""
def __init__(
self,
collection_name: str,
persist_directory: str,
embedding_function: Optional[Callable] = None
):
"""
初始化 Chroma
Args:
collection_name: Collection 名稱
persist_directory: 持久化目錄
embedding_function: Embedding 函數（預設使用 Gemini embeddings）
"""
pass
3.2 RAGService
檔案: core/knowledge/rag_service.py
類別定義:
class RAGService:
"""RAG (Retrieval-Augmented Generation) 服務"""
def __init__(
self,
vector_store: VectorStore,
llm: BaseLLM
):
self.vector_store = vector_store
self.llm = llm
async def query(
self,
question: str,
k: int = 5,
context_filter: Optional[Dict[str, Any]] = None
) -> RAGResponse:
"""
RAG 查詢
Args:
question: 問題
k: 檢索文件數
context_filter: 上下文過濾
Returns:
RAGResponse:
- answer: 回答
- sources: 來源文件
- confidence: 信心分數
"""
# 1. 檢索相關文件
docs = await self.vector_store.search(question, k)
# 2. 組合 context
context = self._format_context(docs)
# 3. LLM 生成回答
prompt = self._build_rag_prompt(question, context)
answer = await self.llm.generate(prompt)
# 4. 返回結果
return RAGResponse(
answer=answer,
sources=docs,
confidence=self._calculate_confidence(docs)
)
async def add_knowledge(
self,
documents: List[Document],
metadata: Optional[Dict[str, Any]] = None
) -> None:
"""新增知識到向量庫"""
pass
3.3 DomainBase
檔案: core/knowledge/domain_base.py
用途: 領域知識基類，供不同應用繼承
類別定義:
class DomainKnowledge(ABC):
"""領域知識基類"""
def __init__(self, rag_service: RAGService):
self.rag_service = rag_service
@abstractmethod
async def initialize(self) -> None:
"""初始化領域知識庫"""
pass
@abstractmethod
async def query(
self,
question: str,
context: Optional[Dict[str, Any]] = None
) -> str:
"""查詢領域知識"""
pass
@abstractmethod
def get_concepts(self) -> List[str]:
"""獲取核心概念列表"""
pass
4. Utils 模組 (core/utils/)
4.1 Config
檔案: core/utils/config.py
類別定義:
class Config:
"""配置管理器"""
def __init__(self, config_dir: str = "config"):
"""
從 YAML 檔案載入配置
支援環境變數替換: ${ENV_VAR}
"""
self.config_dir = config_dir
self.configs: Dict[str, Any] = {}
self.load_all()
def load_all(self) -> None:
"""載入所有 .yaml 配置檔"""
pass
def get(self, key: str, default: Any = None) -> Any:
"""
獲取配置值
支援點號路徑: config.get("llm.gemini.api_key")
"""
pass
def reload(self) -> None:
"""重新載入配置（熱更新）"""
pass
4.2 Logging
檔案: core/utils/logging.py
函數定義:
def get_logger(
name: str,
level: str = "INFO",
log_file: Optional[str] = None
) -> logging.Logger:
"""
獲取 logger 實例
Args:
name: Logger 名稱
level: 日誌等級
log_file: 日誌檔案路徑（可選）
Returns:
配置好的 Logger
Format:
%(asctime)s - %(name)s - %(levelname)s - %(message)s
"""
pass
def setup_logging(config: Dict[str, Any]) -> None:
"""根據配置初始化日誌系統"""
pass
4.3 Monitoring
檔案: core/utils/monitoring.py
類別定義:
class ExecutionMonitor:
"""Agent 執行監控"""
def __init__(self, db_connection):
self.db = db_connection
async def log_execution(
self,
agent_name: str,
task: Dict[str, Any],
result: Dict[str, Any],
duration_ms: int,
success: bool
) -> str:
"""
記錄 Agent 執行到資料庫
Returns:
執行 ID
"""
pass
async def get_success_rate(
self,
agent_name: str,
time_range: Optional[Tuple[datetime, datetime]] = None
) -> float:
"""計算成功率"""
pass
async def get_common_failures(
self,
agent_name: str,
limit: int = 10
) -> List[Dict[str, Any]]:
"""獲取常見失敗模式"""
pass
5. Models 模組 (core/models/)
5.1 Message
檔案: core/models/message.py
from pydantic import BaseModel
from typing import Literal
class Message(BaseModel):
"""訊息模型"""
role: Literal["user", "assistant", "system"]
content: str
timestamp: datetime = Field(default_factory=datetime.now)
metadata: Optional[Dict[str, Any]] = None
class ChatHistory(BaseModel):
"""對話歷史"""
messages: List[Message] = []
session_id: str
def add_message(self, role: str, content: str) -> None:
"""新增訊息"""
pass
def get_context(self, max_messages: int = 10) -> List[Message]:
"""獲取最近 N 條訊息作為上下文"""
pass
5.2 Strategy
檔案: core/models/strategy.py
class VRRStrategy(BaseModel):
"""VRR 策略模型"""
verify_criteria: List[str]
reflect_focus: List[str]
refine_priority: List[str]
domain: str
task_type: str
created_at: datetime = Field(default_factory=datetime.now)
def to_dict(self) -> Dict[str, Any]:
"""轉換為字典"""
pass
@classmethod
def from_llm_response(cls, response: str) -> "VRRStrategy":
"""從 LLM JSON 回應解析"""
pass
class VRRResult(BaseModel):
"""VRR 執行結果"""
final_result: str
iterations: int
verification_history: List[VerificationResult]
reflection_history: List[ReflectionResult]
success: bool
total_duration_ms: int
class VerificationResult(BaseModel):
"""驗證結果"""
passed: bool
failed_criteria: List[str]
details: Dict[str, Any]
timestamp: datetime = Field(default_factory=datetime.now)
class ReflectionResult(BaseModel):
"""反思結果"""
identified_issues: List[str]
root_causes: List[str]
improvement_suggestions: List[str]
5.3 AgentState
檔案: core/models/agent_state.py
class AgentTask(BaseModel):
"""Agent 任務"""
task_id: str = Field(default_factory=lambda: str(uuid.uuid4()))
task_type: str
input_data: Dict[str, Any]
context: Optional[Dict[str, Any]] = None
created_at: datetime = Field(default_factory=datetime.now)
class AgentResult(BaseModel):
"""Agent 執行結果"""
task_id: str
agent_name: str
output: Dict[str, Any]
success: bool
error: Optional[str] = None
duration_ms: int
metadata: Optional[Dict[str, Any]] = None
class AgentExecution(BaseModel):
"""Agent 執行記錄（用於資料庫）"""
id: str = Field(default_factory=lambda: str(uuid.uuid4()))
agent_name: str
task: AgentTask
result: AgentResult
strategy: Optional[VRRStrategy] = None
vrr_result: Optional[VRRResult] = None
created_at: datetime = Field(default_factory=datetime.now)
6. 安裝與依賴
setup.py:
from setuptools import setup, find_packages
setup(
name="cortex-core",
version="0.1.0",
packages=find_packages(),
install_requires=[
"google-generativeai>=0.3.0",
"langchain>=0.1.0",
"langgraph>=0.0.20",
"chromadb>=0.4.0",
"pydantic>=2.0.0",
"pyyaml>=6.0",
"python-dotenv>=1.0.0",
"aiohttp>=3.9.0",
"jinja2>=3.1.0",
],
python_requires=">=3.11",
)
安裝方式:
# 開發模式安裝
cd core/
pip install -e .
7. 測試策略
單元測試結構
core/
└── tests/
├── llm/
│ ├── test_base.py
│ ├── test_gemini.py
│ └── test_prompt_manager.py
├── agent_framework/
│ ├── test_base_agent.py
│ ├── test_meta_agent.py
│ └── test_vrr_engine.py
├── knowledge/
│ ├── test_vector_store.py
│ └── test_rag_service.py
└── utils/
├── test_config.py
└── test_logging.py
Mock 策略
- LLM 測試: Mock Gemini API 回應
- VectorStore 測試: 使用 in-memory Chroma
- Agent 測試: Mock 依賴的其他 Agents
覆蓋率目標
- Core framework: 80%+
- 關鍵路徑 (VRR Engine, MetaAgent): 90%+
8. 使用範例
初始化 Core Components
from core.llm import GeminiLLM
from core.agent_framework import MetaAgent, VRREngine
from core.knowledge import ChromaVectorStore, RAGService
from core.utils import Config
# 載入配置
config = Config("config/")
# 初始化 LLM
llm = GeminiLLM(
api_key=config.get("llm.gemini.api_key"),
model_name="gemini-1.5-flash"
)
# 初始化 VRR Engine
vrr_engine = VRREngine(llm=llm, max_iterations=3)
# 初始化 Meta-Agent
meta_agent = MetaAgent(
name="meta",
llm=llm,
config=config.get("agents.meta")
)
# 初始化 RAG
vector_store = ChromaVectorStore(
collection_name="investment_knowledge",
persist_directory="data/chroma"
)
rag_service = RAGService(vector_store=vector_store, llm=llm)
執行 Meta-Agent
# 生成策略
strategy = await meta_agent.generate_strategy(
task_description="分析台積電的技術指標趨勢",
domain="investment",
context={
"ticker": "2330.TW",
"timeframe": "3_months"
}
)
print(strategy.verify_criteria)
# ["RSI 數值應在 0-100 範圍內", "MACD 交叉點識別正確", ...]
執行 VRR
# 假設某個 Agent 生成了初始分析結果
initial_analysis = "台積電目前 RSI 為 65，處於中性偏多..."
# 執行 VRR
vrr_result = await vrr_engine.execute(
initial_result=initial_analysis,
strategy=strategy,
task_context={"ticker": "2330.TW"}
)
if vrr_result.success:
print(f"分析通過驗證！迭代次數: {vrr_result.iterations}")
print(vrr_result.final_result)
else:
print("分析未通過驗證")
print(vrr_result.verification_history[-1].failed_criteria)
9. 配置檔案範例
config/core.yaml:
llm:
gemini:
api_key: ${GOOGLE_API_KEY}
models:
fast: gemini-1.5-flash
pro: gemini-1.5-pro
default_temperature: 0.7
max_tokens: 2000
rate_limits:
requests_per_minute: 60
retry:
max_attempts: 3
backoff_factor: 2
agent_framework:
vrr:
max_iterations: 3
enable_optimization: true
meta_agent:
strategy_cache_ttl: 3600 # seconds
knowledge:
chroma:
persist_directory: data/chroma
embedding_model: gemini # 使用 Gemini embeddings
rag:
top_k: 5
min_relevance_score: 0.7
logging:
level: INFO
format: "%(asctime)s - %(name)s - %(levelname)s - %(message)s"
file: logs/cortex.log
rotation: daily
retention_days: 30
monitoring:
enable_db_logging: true
metrics_interval: 60 # seconds
10. 擴展指南
新增 LLM 提供商
- 在
core/llm/新增{provider}.py - 繼承
BaseLLM - 實現所有抽象方法
- 在
config/core.yaml新增配置
新增 Agent 類型
- 在應用層繼承
BaseAgent - 實現
execute()方法 - 在
AgentCoordinator中註冊
新增領域知識
- 繼承
DomainKnowledge - 準備領域文件
- 實現
initialize()和query()方法
文檔版本: 1.0
最後更新: 2025-02-01