90 lines
2.9 KiB
Python
90 lines
2.9 KiB
Python
"""
|
||
配置管理模块
|
||
负责加载和管理所有配置项
|
||
"""
|
||
|
||
import os
|
||
from dataclasses import dataclass
|
||
from typing import Optional
|
||
from dotenv import load_dotenv
|
||
|
||
|
||
@dataclass
|
||
class Config:
|
||
"""Agent配置类"""
|
||
|
||
# LiteLLM Gateway配置
|
||
litellm_gateway_url: str
|
||
litellm_api_key: str
|
||
litellm_model: str = "taiji/gpt-4o-mini"
|
||
|
||
# Facebook RapidAPI配置
|
||
facebook_api_host: str = "facebook-scraper3.p.rapidapi.com"
|
||
facebook_api_key: str = "34225c5924msh453fa7aff7a52a9p1d7adfjsn260ba0796c00"
|
||
facebook_mcp_url: str = "https://mcp.rapidapi.com"
|
||
|
||
# Agent配置
|
||
max_results: int = 10
|
||
timeout: int = 30
|
||
|
||
# 可选配置
|
||
log_level: str = "INFO"
|
||
|
||
@staticmethod
|
||
def ensure_model_prefix(model_name: str) -> str:
|
||
"""确保模型名称有 openai: 前缀（pydantic_ai 要求格式为 provider:model_name）"""
|
||
if not model_name:
|
||
return 'openai:taiji/gpt-4o-mini'
|
||
# 如果已经有 provider: 前缀，直接返回
|
||
if ':' in model_name:
|
||
return model_name
|
||
# 否则添加 openai: 前缀
|
||
return f'openai:{model_name}'
|
||
|
||
@classmethod
|
||
def from_env(cls, env_path: Optional[str] = None) -> "Config":
|
||
"""从环境变量加载配置"""
|
||
if env_path:
|
||
load_dotenv(env_path)
|
||
else:
|
||
load_dotenv()
|
||
|
||
# 获取模型名称并确保有 openai: 前缀
|
||
raw_model = os.getenv("LITELLM_MODEL", "taiji/gpt-4o-mini")
|
||
model_name = cls.ensure_model_prefix(raw_model)
|
||
|
||
return cls(
|
||
# LiteLLM Gateway配置
|
||
litellm_gateway_url=os.getenv("LITELLM_GATEWAY_URL", ""),
|
||
litellm_api_key=os.getenv("LITELLM_API_KEY", "sk"),
|
||
litellm_model=model_name,
|
||
|
||
# Facebook RapidAPI配置
|
||
facebook_api_host=os.getenv("FACEBOOK_API_HOST", "facebook-scraper3.p.rapidapi.com"),
|
||
facebook_api_key=os.getenv("FACEBOOK_API_KEY", "34225c5924msh453fa7aff7a52a9p1d7adfjsn260ba0796c00"),
|
||
facebook_mcp_url=os.getenv("FACEBOOK_MCP_URL", "https://mcp.rapidapi.com"),
|
||
|
||
# Agent配置
|
||
max_results=int(os.getenv("MAX_RESULTS", "10")),
|
||
timeout=int(os.getenv("TIMEOUT", "30")),
|
||
|
||
# 可选配置
|
||
log_level=os.getenv("LOG_LEVEL", "INFO")
|
||
)
|
||
|
||
def validate(self) -> bool:
|
||
"""验证配置是否完整"""
|
||
required_fields = [
|
||
("litellm_gateway_url", self.litellm_gateway_url),
|
||
("litellm_api_key", self.litellm_api_key),
|
||
("facebook_api_key", self.facebook_api_key),
|
||
]
|
||
|
||
missing = [name for name, value in required_fields if not value]
|
||
|
||
if missing:
|
||
raise ValueError(f"缺少必要的配置项: {', '.join(missing)}")
|
||
|
||
return True
|
||
|