模块4：状态管理系统迁移方案
1. 现状分析
1.1 当前LangGraph状态管理
# 当前LangGraph状态定义
from typing import TypedDict, Optional, List, Dict, Any
from datetime import datetime
class AgentState(TypedDict):
"""智能体状态定义"""
# 基础信息
messages: List[Dict[str, Any]]
stock_symbol: str
market: str
# 分析结果
fundamentals_analysis: Optional[str]
market_analysis: Optional[str]
news_analysis: Optional[str]
social_media_analysis: Optional[str]
# 研究辩论
bull_argument: Optional[str]
bear_argument: Optional[str]
debate_history: List[Dict[str, Any]]
# 风险管理
risk_assessment: Optional[str]
risk_score: Optional[float]
risk_factors: List[str]
# 交易决策
final_decision: Optional[str]
confidence_score: float
recommended_action: Optional[str]
target_price: Optional[float]
stop_loss: Optional[float]
# 执行状态
execution_status: str
error_message: Optional[str]
execution_times: Dict[str, float]
# 性能指标
memory_usage: Optional[float]
token_usage: Optional[int]
# 时间戳
created_at: str
updated_at: str
# 状态初始化函数
def create_initial_state(stock_symbol: str, market: str = "us") -> AgentState:
"""创建初始状态"""
now = datetime.now().isoformat()
return {
'messages': [],
'stock_symbol': stock_symbol,
'market': market,
'fundamentals_analysis': None,
'market_analysis': None,
'news_analysis': None,
'social_media_analysis': None,
'bull_argument': None,
'bear_argument': None,
'debate_history': [],
'risk_assessment': None,
'risk_score': None,
'risk_factors': [],
'final_decision': None,
'confidence_score': 0.0,
'recommended_action': None,
'target_price': None,
'stop_loss': None,
'execution_status': 'pending',
'error_message': None,
'execution_times': {},
'memory_usage': None,
'token_usage': None,
'created_at': now,
'updated_at': now
}
# 状态更新函数
def update_state(state: AgentState, updates: Dict[str, Any]) -> AgentState:
"""更新状态"""
state.update(updates)
state['updated_at'] = datetime.now().isoformat()
return state
1.2 当前状态管理特点
1. TypedDict定义：使用TypedDict定义状态结构 2. 扁平结构：所有字段都在同一层级 3. 手动更新：需要手动调用update_state函数 4. 无验证机制：没有类型验证和默认值处理 5. 简单时间戳：只有创建和更新时间戳
2. Agno状态管理架构设计
2.1 Pydantic基础状态模型
from pydantic import BaseModel, Field, validator
from typing import Optional, List, Dict, Any
from datetime import datetime
from enum import Enum
class ExecutionStatus(str, Enum):
"""执行状态枚举"""
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
CANCELLED = "cancelled"
class AnalysisResult(BaseModel):
"""分析结果基类"""
content: str = Field(..., description="分析内容")
confidence_score: float = Field(default=0.0, ge=0.0, le=1.0, description="置信度分数")
key_metrics: Dict[str, Any] = Field(default_factory=dict, description="关键指标")
risk_factors: List[str] = Field(default_factory=list, description="风险因素")
timestamp: datetime = Field(default_factory=datetime.now, description="时间戳")
@validator('confidence_score')
def validate_confidence_score(cls, v):
if not 0 <= v <= 1:
raise ValueError('置信度分数必须在0-1之间')
return v
class FundamentalsAnalysis(AnalysisResult):
"""基本面分析结果"""
pe_ratio: Optional[float] = Field(None, description="市盈率")
pb_ratio: Optional[float] = Field(None, description="市净率")
roe: Optional[float] = Field(None, description="净资产收益率")
debt_ratio: Optional[float] = Field(None, description="负债率")
revenue_growth: Optional[float] = Field(None, description="营收增长率")
class Config:
schema_extra = {
"example": {
"content": "公司基本面良好，财务状况稳健...",
"confidence_score": 0.8,
"pe_ratio": 15.2,
"pb_ratio": 2.1,
"roe": 0.15,
"key_metrics": {"market_cap": "1000亿", "dividend_yield": "3.2%"}
}
}
class MarketAnalysis(AnalysisResult):
"""市场分析结果"""
trend_direction: str = Field(default="neutral", description="趋势方向")
support_level: Optional[float] = Field(None, description="支撑位")
resistance_level: Optional[float] = Field(None, description="阻力位")
volume_analysis: str = Field(default="", description="成交量分析")
technical_indicators: Dict[str, Any] = Field(default_factory=dict, description="技术指标")
class RiskAssessment(BaseModel):
"""风险评估结果"""
risk_score: float = Field(..., ge=0.0, le=10.0, description="风险评分")
risk_level: str = Field(..., description="风险等级")
risk_factors: List[str] = Field(default_factory=list, description="风险因素")
mitigation_suggestions: List[str] = Field(default_factory=list, description="缓解建议")
confidence_score: float = Field(default=0.0, ge=0.0, le=1.0, description="置信度")
@validator('risk_level')
def validate_risk_level(cls, v):
valid_levels = ["low", "medium", "high", "extreme"]
if v.lower() not in valid_levels:
raise ValueError(f'风险等级必须是以下之一: {valid_levels}')
return v.lower()
class TradingDecision(BaseModel):
"""交易决策"""
action: str = Field(..., description="交易动作")
confidence_score: float = Field(..., ge=0.0, le=1.0, description="置信度分数")
target_price: Optional[float] = Field(None, description="目标价格")
stop_loss: Optional[float] = Field(None, description="止损价格")
position_size: Optional[float] = Field(None, description="仓位大小")
reasoning: str = Field(default="", description="决策理由")
risk_reward_ratio: Optional[float] = Field(None, description="风险收益比")
@validator('action')
def validate_action(cls, v):
valid_actions = ["buy", "sell", "hold", "wait", "strong_buy", "strong_sell"]
if v.lower() not in valid_actions:
raise ValueError(f'交易动作必须是以下之一: {valid_actions}')
return v.lower()
class PerformanceMetrics(BaseModel):
"""性能指标"""
execution_time: float = Field(..., ge=0.0, description="执行时间（秒）")
memory_usage_start: Optional[float] = Field(None, description="开始内存使用（MB）")
memory_usage_end: Optional[float] = Field(None, description="结束内存使用（MB）")
memory_usage_delta: Optional[float] = Field(None, description="内存使用变化（MB）")
token_usage: Optional[int] = Field(None, description="Token使用量")
api_calls: int = Field(default=0, ge=0, description="API调用次数")
cache_hits: int = Field(default=0, ge=0, description="缓存命中次数")
class ErrorInfo(BaseModel):
"""错误信息"""
error_type: str = Field(..., description="错误类型")
error_message: str = Field(..., description="错误消息")
error_code: Optional[str] = Field(None, description="错误代码")
stack_trace: Optional[str] = Field(None, description="堆栈跟踪")
recovery_suggestion: Optional[str] = Field(None, description="恢复建议")
timestamp: datetime = Field(default_factory=datetime.now, description="错误时间")
2.2 主状态模型
class TradingAgentState(BaseModel):
"""交易智能体状态（Agno版本）"""
# 基础信息
stock_symbol: str = Field(..., description="股票代码", min_length=1, max_length=20)
market: str = Field(default="us", description="市场", regex=r"^(us|hk|cn)$")
company_name: Optional[str] = Field(None, description="公司名称")
# 消息历史
messages: List[Dict[str, Any]] = Field(default_factory=list, description="消息历史")
# 分析结果
fundamentals_analysis: Optional[FundamentalsAnalysis] = Field(None, description="基本面分析")
market_analysis: Optional[MarketAnalysis] = Field(None, description="市场分析")
news_analysis: Optional[AnalysisResult] = Field(None, description="新闻分析")
social_media_analysis: Optional[AnalysisResult] = Field(None, description="社交媒体分析")
# 研究辩论
bull_argument: Optional[AnalysisResult] = Field(None, description="看涨论证")
bear_argument: Optional[AnalysisResult] = Field(None, description="看跌论证")
debate_history: List[Dict[str, Any]] = Field(default_factory=list, description="辩论历史")
# 风险管理
risk_assessment: Optional[RiskAssessment] = Field(None, description="风险评估")
# 交易决策
final_decision: Optional[TradingDecision] = Field(None, description="最终决策")
# 执行状态
execution_status: ExecutionStatus = Field(default=ExecutionStatus.PENDING, description="执行状态")
error_info: Optional[ErrorInfo] = Field(None, description="错误信息")
# 性能指标
performance_metrics: Dict[str, PerformanceMetrics] = Field(default_factory=dict, description="性能指标")
# 元数据
metadata: Dict[str, Any] = Field(default_factory=dict, description="元数据")
# 时间戳
created_at: datetime = Field(default_factory=datetime.now, description="创建时间")
updated_at: datetime = Field(default_factory=datetime.now, description="更新时间")
class Config:
"""Pydantic配置"""
validate_assignment = True # 赋值时验证
use_enum_values = True # 使用枚举值
json_encoders = {
datetime: lambda v: v.isoformat()
}
@validator('stock_symbol')
def validate_stock_symbol(cls, v):
"""验证股票代码"""
if not v or len(v.strip()) == 0:
raise ValueError('股票代码不能为空')
return v.strip().upper()
@validator('updated_at', always=True)
def update_timestamp(cls, v, values):
"""更新时间戳"""
return datetime.now()
def add_message(self, role: str, content: str, metadata: Optional[Dict[str, Any]] = None):
"""添加消息"""
message = {
'role': role,
'content': content,
'timestamp': datetime.now(),
'metadata': metadata or {}
}
self.messages.append(message)
self.updated_at = datetime.now()
def update_analysis(self, analysis_type: str, analysis: AnalysisResult):
"""更新分析结果"""
if analysis_type == "fundamentals":
self.fundamentals_analysis = analysis
elif analysis_type == "market":
self.market_analysis = analysis
elif analysis_type == "news":
self.news_analysis = analysis
elif analysis_type == "social_media":
self.social_media_analysis = analysis
elif analysis_type == "bull":
self.bull_argument = analysis
elif analysis_type == "bear":
self.bear_argument = analysis
else:
raise ValueError(f"未知的分析类型: {analysis_type}")
self.updated_at = datetime.now()
def set_error(self, error_type: str, error_message: str, error_code: Optional[str] = None):
"""设置错误信息"""
self.error_info = ErrorInfo(
error_type=error_type,
error_message=error_message,
error_code=error_code
)
self.execution_status = ExecutionStatus.FAILED
self.updated_at = datetime.now()
def clear_error(self):
"""清除错误信息"""
self.error_info = None
if self.execution_status == ExecutionStatus.FAILED:
self.execution_status = ExecutionStatus.PENDING
self.updated_at = datetime.now()
def add_performance_metric(self, phase: str, metric: PerformanceMetrics):
"""添加性能指标"""
self.performance_metrics[phase] = metric
self.updated_at = datetime.now()
def get_total_execution_time(self) -> float:
"""获取总执行时间"""
return sum(metric.execution_time for metric in self.performance_metrics.values())
def get_average_confidence_score(self) -> float:
"""获取平均置信度分数"""
confidence_scores = []
if self.fundamentals_analysis:
confidence_scores.append(self.fundamentals_analysis.confidence_score)
if self.market_analysis:
confidence_scores.append(self.market_analysis.confidence_score)
if self.news_analysis:
confidence_scores.append(self.news_analysis.confidence_score)
if self.social_media_analysis:
confidence_scores.append(self.social_media_analysis.confidence_score)
if self.final_decision:
confidence_scores.append(self.final_decision.confidence_score)
return sum(confidence_scores) / len(confidence_scores) if confidence_scores else 0.0
def is_complete(self) -> bool:
"""检查是否完成"""
return self.execution_status == ExecutionStatus.COMPLETED
def has_errors(self) -> bool:
"""检查是否有错误"""
return self.error_info is not None
def get_summary(self) -> Dict[str, Any]:
"""获取状态摘要"""
return {
'stock_symbol': self.stock_symbol,
'market': self.market,
'execution_status': self.execution_status.value,
'confidence_score': self.get_average_confidence_score(),
'total_execution_time': self.get_total_execution_time(),
'has_errors': self.has_errors(),
'created_at': self.created_at,
'updated_at': self.updated_at
}
2.3 状态持久化
import json
import redis
import pickle
from typing import Optional, Dict, Any
from datetime import datetime, timedelta
import logging
logger = logging.getLogger(__name__)
class StatePersistenceManager:
"""状态持久化管理器"""
def __init__(self,
redis_client: Optional[redis.Redis] = None,
redis_host: str = "localhost",
redis_port: int = 6379,
redis_db: int = 0,
redis_password: Optional[str] = None,
default_ttl: int = 3600):
self.redis_client = redis_client or redis.Redis(
host=redis_host,
port=redis_port,
db=redis_db,
password=redis_password,
decode_responses=False # 使用二进制序列化
)
self.default_ttl = default_ttl
self.logger = logging.getLogger(__name__)
def _generate_key(self, stock_symbol: str, market: str, session_id: Optional[str] = None) -> str:
"""生成Redis键"""
if session_id:
return f"trading_state:{market}:{stock_symbol}:{session_id}"
else:
return f"trading_state:{market}:{stock_symbol}"
def save_state(self, state: TradingAgentState, session_id: Optional[str] = None, ttl: Optional[int] = None) -> bool:
"""保存状态"""
try:
key = self._generate_key(state.stock_symbol, state.market, session_id)
# 序列化状态
serialized_state = pickle.dumps(state)
# 保存到Redis
ttl = ttl or self.default_ttl
self.redis_client.setex(key, ttl, serialized_state)
self.logger.info(f"状态保存成功: {key}")
return True
except Exception as e:
self.logger.error(f"状态保存失败: {str(e)}")
return False
def load_state(self, stock_symbol: str, market: str, session_id: Optional[str] = None) -> Optional[TradingAgentState]:
"""加载状态"""
try:
key = self._generate_key(stock_symbol, market, session_id)
# 从Redis获取
serialized_state = self.redis_client.get(key)
if not serialized_state:
self.logger.info(f"状态不存在: {key}")
return None
# 反序列化
state = pickle.loads(serialized_state)
self.logger.info(f"状态加载成功: {key}")
return state
except Exception as e:
self.logger.error(f"状态加载失败: {str(e)}")
return None
def delete_state(self, stock_symbol: str, market: str, session_id: Optional[str] = None) -> bool:
"""删除状态"""
try:
key = self._generate_key(stock_symbol, market, session_id)
result = self.redis_client.delete(key)
if result > 0:
self.logger.info(f"状态删除成功: {key}")
return True
else:
self.logger.warning(f"状态不存在，无法删除: {key}")
return False
except Exception as e:
self.logger.error(f"状态删除失败: {str(e)}")
return False
def state_exists(self, stock_symbol: str, market: str, session_id: Optional[str] = None) -> bool:
"""检查状态是否存在"""
try:
key = self._generate_key(stock_symbol, market, session_id)
return self.redis_client.exists(key) > 0
except Exception as e:
self.logger.error(f"检查状态存在性失败: {str(e)}")
return False
def get_state_ttl(self, stock_symbol: str, market: str, session_id: Optional[str] = None) -> Optional[int]:
"""获取状态剩余TTL"""
try:
key = self._generate_key(stock_symbol, market, session_id)
ttl = self.redis_client.ttl(key)
return ttl if ttl >= 0 else None
except Exception as e:
self.logger.error(f"获取状态TTL失败: {str(e)}")
return None
def extend_state_ttl(self, stock_symbol: str, market: str, session_id: Optional[str] = None, additional_ttl: int = 3600) -> bool:
"""延长状态TTL"""
try:
key = self._generate_key(stock_symbol, market, session_id)
# 检查状态是否存在
if not self.state_exists(stock_symbol, market, session_id):
self.logger.warning(f"状态不存在，无法延长TTL: {key}")
return False
# 延长TTL
result = self.redis_client.expire(key, additional_ttl)
if result:
self.logger.info(f"状态TTL延长成功: {key}")
return True
else:
self.logger.error(f"状态TTL延长失败: {key}")
return False
except Exception as e:
self.logger.error(f"延长状态TTL失败: {str(e)}")
return False
def save_state_batch(self, states: Dict[str, TradingAgentState], ttl: Optional[int] = None) -> Dict[str, bool]:
"""批量保存状态"""
results = {}
for key, state in states.items():
try:
# 序列化状态
serialized_state = pickle.dumps(state)
# 保存到Redis
ttl = ttl or self.default_ttl
self.redis_client.setex(key, ttl, serialized_state)
results[key] = True
self.logger.info(f"批量状态保存成功: {key}")
except Exception as e:
results[key] = False
self.logger.error(f"批量状态保存失败 {key}: {str(e)}")
return results
def get_all_states(self, pattern: str = "trading_state:*") -> Dict[str, TradingAgentState]:
"""获取所有匹配的状态"""
states = {}
try:
# 获取所有匹配的键
keys = self.redis_client.keys(pattern)
if not keys:
return states
# 批量获取值
values = self.redis_client.mget(keys)
for key, value in zip(keys, values):
if value:
try:
# 反序列化
state = pickle.loads(value)
states[key.decode() if isinstance(key, bytes) else key] = state
except Exception as e:
self.logger.error(f"反序列化状态失败 {key}: {str(e)}")
self.logger.info(f"获取所有状态成功，共{len(states)}个")
except Exception as e:
self.logger.error(f"获取所有状态失败: {str(e)}")
return states
def cleanup_expired_states(self, pattern: str = "trading_state:*") -> int:
"""清理过期状态"""
try:
# 获取所有匹配的键
keys = self.redis_client.keys(pattern)
if not keys:
return 0
# 检查每个键的TTL
expired_keys = []
for key in keys:
ttl = self.redis_client.ttl(key)
if ttl == -2: # 键不存在
expired_keys.append(key)
# 删除过期键
if expired_keys:
result = self.redis_client.delete(*expired_keys)
self.logger.info(f"清理过期状态成功，共{result}个")
return result
return 0
except Exception as e:
self.logger.error(f"清理过期状态失败: {str(e)}")
return 0
3. 迁移挑战与解决方案
3.1 状态结构迁移
#### 挑战
- LangGraph使用扁平的TypedDict结构
- Agno使用嵌套的Pydantic模型
- 字段映射和类型转换
class StateMigrationConverter:
"""状态迁移转换器"""
@staticmethod
def convert_langgraph_to_agno(langgraph_state: Dict[str, Any]) -> TradingAgentState:
"""转换LangGraph状态到Agno状态"""
try:
# 基础信息
stock_symbol = langgraph_state.get('stock_symbol', '')
market = langgraph_state.get('market', 'us')
# 分析结果转换
fundamentals_analysis = None
if langgraph_state.get('fundamentals_analysis'):
fundamentals_analysis = FundamentalsAnalysis(
content=langgraph_state['fundamentals_analysis'],
confidence_score=langgraph_state.get('fundamentals_confidence', 0.0)
)
market_analysis = None
if langgraph_state.get('market_analysis'):
market_analysis = MarketAnalysis(
content=langgraph_state['market_analysis'],
confidence_score=langgraph_state.get('market_confidence', 0.0),
trend_direction=langgraph_state.get('trend_direction', 'neutral')
)
news_analysis = None
if langgraph_state.get('news_analysis'):
news_analysis = AnalysisResult(
content=langgraph_state['news_analysis'],
confidence_score=langgraph_state.get('news_confidence', 0.0)
)
social_media_analysis = None
if langgraph_state.get('social_media_analysis'):
social_media_analysis = AnalysisResult(
content=langgraph_state['social_media_analysis'],
confidence_score=langgraph_state.get('social_media_confidence', 0.0)
)
# 研究辩论转换
bull_argument = None
if langgraph_state.get('bull_argument'):
bull_argument = AnalysisResult(
content=langgraph_state['bull_argument'],
confidence_score=langgraph_state.get('bull_confidence', 0.0)
)
bear_argument = None
if langgraph_state.get('bear_argument'):
bear_argument = AnalysisResult(
content=langgraph_state['bear_argument'],
confidence_score=langgraph_state.get('bear_confidence', 0.0)
)
# 风险评估转换
risk_assessment = None
if langgraph_state.get('risk_assessment'):
risk_assessment = RiskAssessment(
risk_score=langgraph_state.get('risk_score', 0.0),
risk_level=langgraph_state.get('risk_level', 'medium'),
risk_factors=langgraph_state.get('risk_factors', [])
)
# 交易决策转换
final_decision = None
if langgraph_state.get('final_decision'):
final_decision = TradingDecision(
action=langgraph_state.get('recommended_action', 'hold'),
confidence_score=langgraph_state.get('confidence_score', 0.0),
target_price=langgraph_state.get('target_price'),
stop_loss=langgraph_state.get('stop_loss')
)
# 执行状态转换
execution_status = ExecutionStatus(langgraph_state.get('execution_status', 'pending'))
# 错误信息转换
error_info = None
if langgraph_state.get('error_message'):
error_info = ErrorInfo(
error_type='execution_error',
error_message=langgraph_state['error_message']
)
# 性能指标转换
performance_metrics = {}
execution_times = langgraph_state.get('execution_times', {})
for phase, execution_time in execution_times.items():
performance_metrics[phase] = PerformanceMetrics(
execution_time=execution_time
)
# 创建Agno状态
agno_state = TradingAgentState(
stock_symbol=stock_symbol,
market=market,
messages=langgraph_state.get('messages', []),
fundamentals_analysis=fundamentals_analysis,
market_analysis=market_analysis,
news_analysis=news_analysis,
social_media_analysis=social_media_analysis,
bull_argument=bull_argument,
bear_argument=bear_argument,
debate_history=langgraph_state.get('debate_history', []),
risk_assessment=risk_assessment,
final_decision=final_decision,
execution_status=execution_status,
error_info=error_info,
performance_metrics=performance_metrics,
metadata=langgraph_state.get('metadata', {}),
created_at=datetime.fromisoformat(langgraph_state.get('created_at', datetime.now().isoformat())),
updated_at=datetime.fromisoformat(langgraph_state.get('updated_at', datetime.now().isoformat()))
)
return agno_state
except Exception as e:
logger.error(f"状态转换失败: {str(e)}")
# 返回默认状态
return TradingAgentState(
stock_symbol=langgraph_state.get('stock_symbol', 'UNKNOWN'),
market=langgraph_state.get('market', 'us')
)
@staticmethod
def convert_agno_to_langgraph(agno_state: TradingAgentState) -> Dict[str, Any]:
"""转换Agno状态到LangGraph状态"""
try:
langgraph_state = {
'messages': agno_state.messages,
'stock_symbol': agno_state.stock_symbol,
'market': agno_state.market,
'company_name': agno_state.company_name,
'execution_status': agno_state.execution_status.value,
'created_at': agno_state.created_at.isoformat(),
'updated_at': agno_state.updated_at.isoformat()
}
# 转换分析结果
if agno_state.fundamentals_analysis:
langgraph_state.update({
'fundamentals_analysis': agno_state.fundamentals_analysis.content,
'fundamentals_confidence': agno_state.fundamentals_analysis.confidence_score,
'fundamentals_key_metrics': agno_state.fundamentals_analysis.key_metrics,
'fundamentals_risk_factors': agno_state.fundamentals_analysis.risk_factors
})
if agno_state.market_analysis:
langgraph_state.update({
'market_analysis': agno_state.market_analysis.content,
'market_confidence': agno_state.market_analysis.confidence_score,
'trend_direction': agno_state.market_analysis.trend_direction,
'support_level': agno_state.market_analysis.support_level,
'resistance_level': agno_state.market_analysis.resistance_level,
'volume_analysis': agno_state.market_analysis.volume_analysis
})
if agno_state.news_analysis:
langgraph_state.update({
'news_analysis': agno_state.news_analysis.content,
'news_confidence': agno_state.news_analysis.confidence_score,
'news_risk_factors': agno_state.news_analysis.risk_factors
})
if agno_state.social_media_analysis:
langgraph_state.update({
'social_media_analysis': agno_state.social_media_analysis.content,
'social_media_confidence': agno_state.social_media_analysis.confidence_score,
'social_media_risk_factors': agno_state.social_media_analysis.risk_factors
})
# 转换研究辩论
if agno_state.bull_argument:
langgraph_state.update({
'bull_argument': agno_state.bull_argument.content,
'bull_confidence': agno_state.bull_argument.confidence_score
})
if agno_state.bear_argument:
langgraph_state.update({
'bear_argument': agno_state.bear_argument.content,
'bear_confidence': agno_state.bear_argument.confidence_score
})
langgraph_state['debate_history'] = agno_state.debate_history
# 转换风险评估
if agno_state.risk_assessment:
langgraph_state.update({
'risk_assessment': f"风险评分: {agno_state.risk_assessment.risk_score}, 风险等级: {agno_state.risk_assessment.risk_level}",
'risk_score': agno_state.risk_assessment.risk_score,
'risk_level': agno_state.risk_assessment.risk_level,
'risk_factors': agno_state.risk_assessment.risk_factors,
'mitigation_suggestions': agno_state.risk_assessment.mitigation_suggestions
})
# 转换交易决策
if agno_state.final_decision:
langgraph_state.update({
'final_decision': f"动作: {agno_state.final_decision.action}, 置信度: {agno_state.final_decision.confidence_score}",
'recommended_action': agno_state.final_decision.action,
'confidence_score': agno_state.final_decision.confidence_score,
'target_price': agno_state.final_decision.target_price,
'stop_loss': agno_state.final_decision.stop_loss,
'position_size': agno_state.final_decision.position_size,
'reasoning': agno_state.final_decision.reasoning,
'risk_reward_ratio': agno_state.final_decision.risk_reward_ratio
})
# 转换错误信息
if agno_state.error_info:
langgraph_state.update({
'error_message': agno_state.error_info.error_message,
'error_type': agno_state.error_info.error_type,
'error_code': agno_state.error_info.error_code
})
# 转换性能指标
execution_times = {}
for phase, metrics in agno_state.performance_metrics.items():
execution_times[phase] = metrics.execution_time
langgraph_state['execution_times'] = execution_times
langgraph_state['memory_usage'] = agno_state.get_total_memory_usage()
langgraph_state['token_usage'] = agno_state.get_total_token_usage()
# 转换元数据
langgraph_state['metadata'] = agno_state.metadata
return langgraph_state
except Exception as e:
logger.error(f"状态转换失败: {str(e)}")
# 返回基本状态
return {
'stock_symbol': agno_state.stock_symbol,
'market': agno_state.market,
'execution_status': agno_state.execution_status.value,
'error_message': f"状态转换失败: {str(e)}"
}
3.2 状态验证与清理
#### 挑战
- 状态数据可能不完整或不一致
- 需要验证状态的有效性
- 清理过期或无效的状态
class StateValidator:
"""状态验证器"""
@staticmethod
def validate_trading_state(state: TradingAgentState) -> Dict[str, Any]:
"""验证交易状态"""
validation_result = {
'is_valid': True,
'errors': [],
'warnings': [],
'suggestions': []
}
try:
# 基础信息验证
if not state.stock_symbol:
validation_result['errors'].append("股票代码不能为空")
validation_result['is_valid'] = False
if state.market not in ["us", "hk", "cn"]:
validation_result['errors'].append(f"无效的市场代码: {state.market}")
validation_result['is_valid'] = False
# 分析结果一致性验证
analyses = [
("基本面分析", state.fundamentals_analysis),
("市场分析", state.market_analysis),
("新闻分析", state.news_analysis),
("社交媒体分析", state.social_media_analysis)
]
valid_analyses = [name for name, analysis in analyses if analysis is not None]
if len(valid_analyses) == 0:
validation_result['warnings'].append("没有任何分析结果")
elif len(valid_analyses) < len(analyses):
missing = [name for name, analysis in analyses if analysis is None]
validation_result['warnings'].append(f"缺少分析结果: {', '.join(missing)}")
# 置信度分数验证
for analysis_name, analysis in analyses:
if analysis and analysis.confidence_score > 0:
if analysis.confidence_score < 0.3:
validation_result['warnings'].append(f"{analysis_name}置信度较低: {analysis.confidence_score}")
elif analysis.confidence_score > 0.9:
validation_result['suggestions'].append(f"{analysis_name}置信度很高，可以考虑增加权重")
# 研究辩论验证
if state.bull_argument and state.bear_argument:
bull_confidence = state.bull_argument.confidence_score
bear_confidence = state.bear_argument.confidence_score
if abs(bull_confidence - bear_confidence) < 0.1:
validation_result['warnings'].append("看涨和看跌论证置信度过于接近")
elif bull_confidence > 0.8 and bear_confidence < 0.3:
validation_result['suggestions'].append("看涨论证明显强于看跌论证")
elif bear_confidence > 0.8 and bull_confidence < 0.3:
validation_result['suggestions'].append("看跌论证明显强于看涨论证")
# 风险评估验证
if state.risk_assessment:
if state.risk_assessment.risk_score > 7:
validation_result['warnings'].append("风险评分较高，需要谨慎")
elif state.risk_assessment.risk_score < 3:
validation_result['suggestions'].append("风险评分较低，可以考虑积极策略")
# 交易决策验证
if state.final_decision:
if state.final_decision.confidence_score < 0.5:
validation_result['warnings'].append("交易决策置信度较低")
if state.final_decision.action in ["buy", "sell"] and not state.final_decision.target_price:
validation_result['warnings'].append("买入/卖出决策缺少目标价格")
if state.final_decision.action in ["buy", "sell"] and not state.final_decision.stop_loss:
validation_result['suggestions'].append("建议设置止损价格")
# 执行状态验证
if state.execution_status == ExecutionStatus.COMPLETED:
if not state.final_decision:
validation_result['errors'].append("完成状态但没有最终决策")
validation_result['is_valid'] = False
elif state.execution_status == ExecutionStatus.FAILED:
if not state.error_info:
validation_result['warnings'].append("失败状态但没有错误信息")
# 性能指标验证
if state.performance_metrics:
total_time = state.get_total_execution_time()
if total_time > 300: # 5分钟
validation_result['warnings'].append(f"总执行时间过长: {total_time:.2f}秒")
elif total_time < 10:
validation_result['suggestions'].append("执行时间很短，可能需要增加分析深度")
# 时间戳验证
if state.updated_at < state.created_at:
validation_result['errors'].append("更新时间早于创建时间")
validation_result['is_valid'] = False
# 检查状态是否过期
if datetime.now() - state.updated_at > timedelta(hours=24):
validation_result['warnings'].append("状态已过期（超过24小时）")
return validation_result
except Exception as e:
validation_result['errors'].append(f"验证过程出错: {str(e)}")
validation_result['is_valid'] = False
return validation_result
@staticmethod
def cleanup_state(state: TradingAgentState) -> TradingAgentState:
"""清理状态"""
try:
# 清理空的分析结果
if state.fundamentals_analysis and not state.fundamentals_analysis.content.strip():
state.fundamentals_analysis = None
if state.market_analysis and not state.market_analysis.content.strip():
state.market_analysis = None
if state.news_analysis and not state.news_analysis.content.strip():
state.news_analysis = None
if state.social_media_analysis and not state.social_media_analysis.content.strip():
state.social_media_analysis = None
# 清理空的论证
if state.bull_argument and not state.bull_argument.content.strip():
state.bull_argument = None
if state.bear_argument and not state.bear_argument.content.strip():
state.bear_argument = None
# 清理辩论历史
state.debate_history = [
debate for debate in state.debate_history
if debate.get('content', '').strip()
]
# 清理风险因素
if state.risk_assessment:
state.risk_assessment.risk_factors = [
factor for factor in state.risk_assessment.risk_factors
if factor.strip()
]
# 重置错误状态
if state.execution_status == ExecutionStatus.FAILED and not state.error_info:
state.execution_status = ExecutionStatus.PENDING
# 清理性能指标
state.performance_metrics = {
phase: metrics for phase, metrics in state.performance_metrics.items()
if metrics.execution_time > 0
}
# 清理元数据
state.metadata = {
key: value for key, value in state.metadata.items()
if value is not None
}
return state
except Exception as e:
logger.error(f"状态清理失败: {str(e)}")
return state
3.3 状态版本管理
#### 挑战
- 状态结构可能随时间变化
- 需要向后兼容
- 版本迁移
from typing import Dict, Any, Optional
from datetime import datetime
import json
class StateVersionManager:
"""状态版本管理器"""
CURRENT_VERSION = "2.0"
def __init__(self):
self.version_history = {
"1.0": self._migrate_v1_to_v2,
"1.1": self._migrate_v1_1_to_v2,
}
def detect_version(self, state_data: Dict[str, Any]) -> str:
"""检测状态版本"""
# 检查是否有版本字段
if 'version' in state_data:
return state_data['version']
# 根据结构特征判断版本
if 'execution_status' in state_data and isinstance(state_data['execution_status'], str):
# 新版本特征
return "2.0"
elif 'final_decision' in state_data and isinstance(state_data['final_decision'], dict):
# 中间版本特征
return "1.1"
else:
# 旧版本特征
return "1.0"
def migrate_to_current(self, state_data: Dict[str, Any]) -> Dict[str, Any]:
"""迁移到当前版本"""
current_version = self.detect_version(state_data)
if current_version == self.CURRENT_VERSION:
return state_data
# 逐步迁移
while current_version != self.CURRENT_VERSION:
if current_version in self.version_history:
state_data = self.version_history[current_version](state_data)
current_version = self.detect_version(state_data)
else:
raise ValueError(f"不支持的版本迁移: {current_version}")
# 添加版本信息
state_data['version'] = self.CURRENT_VERSION
state_data['migrated_at'] = datetime.now().isoformat()
return state_data
def _migrate_v1_to_v2(self, state_data: Dict[str, Any]) -> Dict[str, Any]:
"""从v1.0迁移到v2.0"""
# 基础转换
new_state = {
'stock_symbol': state_data.get('stock_symbol', ''),
'market': state_data.get('market', 'us'),
'messages': state_data.get('messages', []),
'execution_status': state_data.get('execution_status', 'pending'),
'created_at': state_data.get('created_at', datetime.now().isoformat()),
'updated_at': state_data.get('updated_at', datetime.now().isoformat())
}
# 转换分析结果
if state_data.get('fundamentals_analysis'):
new_state['fundamentals_analysis'] = {
'content': state_data['fundamentals_analysis'],
'confidence_score': state_data.get('fundamentals_confidence', 0.0),
'key_metrics': {},
'risk_factors': []
}
if state_data.get('market_analysis'):
new_state['market_analysis'] = {
'content': state_data['market_analysis'],
'confidence_score': state_data.get('market_confidence', 0.0),
'trend_direction': 'neutral',
'support_level': None,
'resistance_level': None,
'volume_analysis': '',
'technical_indicators': {}
}
# 转换其他字段...
return new_state
def _migrate_v1_1_to_v2(self, state_data: Dict[str, Any]) -> Dict[str, Any]:
"""从v1.1迁移到v2.0"""
# 这个版本更接近v2.0，迁移更简单
new_state = state_data.copy()
# 添加缺少的字段
if 'performance_metrics' not in new_state:
new_state['performance_metrics'] = {}
if 'metadata' not in new_state:
new_state['metadata'] = {}
return new_state
4. 迁移实施计划
4.1 迁移步骤
1. 状态结构重构
- 将TypedDict转换为Pydantic模型
- 添加验证和默认值
- 实现嵌套结构
- 实现Redis存储
- 添加序列化/反序列化
- 实现批量操作
- 实现状态验证
- 添加清理逻辑
- 版本管理
- 优化序列化性能
- 实现缓存策略
- 监控状态使用
4.2 回滚策略
class StateMigrationRollback:
"""状态迁移回滚管理"""
def __init__(self, backup_manager):
self.backup_manager = backup_manager
def create_backup(self, state: TradingAgentState) -> str:
"""创建状态备份"""
backup_id = f"backup_{datetime.now().strftime('%Y%m%d_%H%M%S')}"
self.backup_manager.save_backup(backup_id, state)
return backup_id
def rollback_to_langgraph(self, backup_id: str) -> Dict[str, Any]:
"""回滚到LangGraph格式"""
agno_state = self.backup_manager.load_backup(backup_id)
if agno_state:
return StateMigrationConverter.convert_agno_to_langgraph(agno_state)
return None
def rollback_to_agno(self, backup_id: str) -> TradingAgentState:
"""回滚到Agno格式"""
return self.backup_manager.load_backup(backup_id)
这个状态管理系统迁移方案提供了从LangGraph到Agno的完整迁移路径，包含详细的状态结构设计、持久化实现、验证机制和版本管理。