3.3 工作流状态管理挑战
挑战描述： LangGraph中的状态管理机制与Agno智能体的状态管理存在差异，需要确保状态转换的正确性和一致性。
解决方案：
class WorkflowStateMigrationAdapter:
"""工作流状态迁移适配器"""
def __init__(self):
self.state_mappings = {}
self.transition_rules = {}
self.validation_rules = {}
self.rollback_states = {}
def analyze_langgraph_state(self, langgraph_state: Dict[str, Any]) -> Dict[str, Any]:
"""分析LangGraph状态结构"""
analysis = {
"state_type": self._identify_state_type(langgraph_state),
"fields": self._extract_state_fields(langgraph_state),
"transitions": self._analyze_transitions(langgraph_state),
"validation_rules": self._extract_validation_rules(langgraph_state),
"persistence_requirements": self._analyze_persistence(langgraph_state)
}
return analysis
def _identify_state_type(self, state: Dict[str, Any]) -> str:
"""识别状态类型"""
# 基于状态字段和结构识别类型
if "messages" in state and "current_node" in state:
return "conversation_state"
elif "workflow_data" in state and "execution_stack" in state:
return "workflow_execution_state"
elif "agent_states" in state and "shared_memory" in state:
return "multi_agent_state"
elif "context" in state and "history" in state:
return "contextual_state"
else:
return "generic_state"
def _extract_state_fields(self, state: Dict[str, Any]) -> Dict[str, Any]:
"""提取状态字段信息"""
fields = {}
for key, value in state.items():
field_info = {
"type": type(value).__name__,
"nullable": value is None,
"mutable": self._is_field_mutable(key, value),
"validation_rules": self._extract_field_validation(key, value),
"default_value": self._get_default_value(value)
}
fields[key] = field_info
return fields
def _is_field_mutable(self, field_name: str, field_value: Any) -> bool:
"""判断字段是否可变"""
# 基于字段名和值判断可变性
immutable_patterns = ["id", "created_at", "uuid", "hash"]
return not any(pattern in field_name.lower() for pattern in immutable_patterns)
def _extract_field_validation(self, field_name: str, field_value: Any) -> List[Dict[str, Any]]:
"""提取字段验证规则"""
validations = []
# 基于字段值类型推断验证规则
if isinstance(field_value, str):
validations.append({
"type": "string_length",
"min_length": 1,
"max_length": 1000
})
elif isinstance(field_value, int):
validations.append({
"type": "numeric_range",
"min_value": 0,
"max_value": 1000000
})
elif isinstance(field_value, list):
validations.append({
"type": "array_size",
"min_size": 0,
"max_size": 1000
})
return validations
def _get_default_value(self, value: Any) -> Any:
"""获取默认值"""
if value is None:
return None
elif isinstance(value, str):
return ""
elif isinstance(value, int):
return 0
elif isinstance(value, float):
return 0.0
elif isinstance(value, bool):
return False
elif isinstance(value, list):
return []
elif isinstance(value, dict):
return {}
else:
return None
def _analyze_transitions(self, state: Dict[str, Any]) -> List[Dict[str, Any]]:
"""分析状态转换"""
transitions = []
# 基于历史信息分析转换模式
if "transition_history" in state:
history = state["transition_history"]
for i in range(len(history) - 1):
transition = {
"from_state": history[i].get("state_name", f"state_{i}"),
"to_state": history[i + 1].get("state_name", f"state_{i + 1}"),
"trigger": history[i].get("trigger", "unknown"),
"conditions": history[i].get("conditions", []),
"side_effects": history[i].get("side_effects", [])
}
transitions.append(transition)
return transitions
def _extract_validation_rules(self, state: Dict[str, Any]) -> Dict[str, Any]:
"""提取验证规则"""
rules = {
"state_validation": [],
"transition_validation": [],
"field_validation": {}
}
# 基于状态内容推断验证规则
for key, value in state.items():
field_rules = self._extract_field_validation(key, value)
if field_rules:
rules["field_validation"][key] = field_rules
return rules
def _analyze_persistence(self, state: Dict[str, Any]) -> Dict[str, Any]:
"""分析持久化需求"""
persistence = {
"needs_persistence": False,
"persistence_level": "none",
"backup_frequency": "never",
"retention_policy": "temporary"
}
# 基于状态内容判断持久化需求
if "critical_data" in state or "user_data" in state:
persistence["needs_persistence"] = True
persistence["persistence_level"] = "high"
persistence["backup_frequency"] = "real_time"
persistence["retention_policy"] = "permanent"
elif "workflow_progress" in state:
persistence["needs_persistence"] = True
persistence["persistence_level"] = "medium"
persistence["backup_frequency"] = "checkpoint"
persistence["retention_policy"] = "session_based"
return persistence
def create_agno_state_model(self, langgraph_analysis: Dict[str, Any]) -> Dict[str, Any]:
"""创建Agno状态模型"""
agno_model = {
"state_class": self._generate_state_class(langgraph_analysis),
"validation_class": self._generate_validation_class(langgraph_analysis),
"transition_class": self._generate_transition_class(langgraph_analysis),
"persistence_config": self._generate_persistence_config(langgraph_analysis)
}
return agno_model
def _generate_state_class(self, analysis: Dict[str, Any]) -> str:
"""生成状态类代码"""
state_type = analysis["state_type"]
fields = analysis["fields"]
class_code = f"""
from dataclasses import dataclass
from typing import Optional, List, Dict, Any
from datetime import datetime
@dataclass
class {state_type.title().replace('_', '')}State:
"""{state_type.replace('_', ' ').title()}状态类"""
"""
# 生成字段定义
for field_name, field_info in fields.items():
field_type = self._map_to_agno_type(field_info["type"])
nullable = "Optional[" if field_info["nullable"] else ""
nullable_end = "]" if field_info["nullable"] else ""
default_value = self._get_agno_default_value(field_info["default_value"])
class_code += f" {field_name}: {nullable}{field_type}{nullable_end} = {default_value}\n"
# 添加元数据字段
class_code += """
# 元数据字段
created_at: datetime = field(default_factory=datetime.now)
updated_at: datetime = field(default_factory=datetime.now)
version: int = 1
def update_timestamp(self):
"""更新时间戳"""
self.updated_at = datetime.now()
self.version += 1
"""
return class_code
def _map_to_agno_type(self, langgraph_type: str) -> str:
"""映射类型到Agno"""
type_mappings = {
"str": "str",
"int": "int",
"float": "float",
"bool": "bool",
"list": "List[Any]",
"dict": "Dict[str, Any]",
"datetime": "datetime"
}
return type_mappings.get(langgraph_type, "Any")
def _get_agno_default_value(self, default_value: Any) -> str:
"""获取Agno默认值"""
if default_value is None:
return "None"
elif isinstance(default_value, str):
return f'"{default_value}"'
elif isinstance(default_value, (int, float, bool)):
return str(default_value)
elif isinstance(default_value, list):
return "field(default_factory=list)"
elif isinstance(default_value, dict):
return "field(default_factory=dict)"
else:
return "None"
def _generate_validation_class(self, analysis: Dict[str, Any]) -> str:
"""生成验证类代码"""
validation_rules = analysis["validation_rules"]
class_code = """
from typing import Dict, Any, List
from dataclasses import dataclass
@dataclass
class StateValidator:
"""状态验证器"""
def validate_state(self, state: Any) -> Dict[str, Any]:
"""验证状态"""
errors = []
warnings = []
# 字段验证
for field_name, field_rules in validation_rules.get("field_validation", {}).items():
field_value = getattr(state, field_name, None)
field_errors = self._validate_field(field_name, field_value, field_rules)
errors.extend(field_errors)
return {
"is_valid": len(errors) == 0,
"errors": errors,
"warnings": warnings
}
def _validate_field(self, field_name: str, field_value: Any, rules: List[Dict[str, Any]]) -> List[str]:
"""验证字段"""
errors = []
for rule in rules:
if rule["type"] == "string_length":
if not isinstance(field_value, str):
errors.append(f"字段 {field_name} 必须是字符串")
elif len(field_value) < rule.get("min_length", 0):
errors.append(f"字段 {field_name} 长度不能小于 {rule.get('min_length', 0)}")
elif len(field_value) > rule.get("max_length", 1000):
errors.append(f"字段 {field_name} 长度不能超过 {rule.get('max_length', 1000)}")
elif rule["type"] == "numeric_range":
if not isinstance(field_value, (int, float)):
errors.append(f"字段 {field_name} 必须是数字")
elif field_value < rule.get("min_value", 0):
errors.append(f"字段 {field_name} 不能小于 {rule.get('min_value', 0)}")
elif field_value > rule.get("max_value", 1000000):
errors.append(f"字段 {field_name} 不能超过 {rule.get('max_value', 1000000)}")
return errors
"""
return class_code
def _generate_transition_class(self, analysis: Dict[str, Any]) -> str:
"""生成转换类代码"""
transitions = analysis["transitions"]
class_code = """
from typing import Dict, Any, List, Optional
from dataclasses import dataclass
from enum import Enum
class TransitionStatus(Enum):
PENDING = "pending"
IN_PROGRESS = "in_progress"
COMPLETED = "completed"
FAILED = "failed"
ROLLBACK = "rollback"
@dataclass
class StateTransition:
"""状态转换"""
from_state: str
to_state: str
trigger: str
conditions: List[str]
side_effects: List[str]
status: TransitionStatus = TransitionStatus.PENDING
error_message: Optional[str] = None
def can_execute(self, current_state: Any, context: Dict[str, Any]) -> bool:
"""检查是否可以执行转换"""
# 检查条件
for condition in self.conditions:
if not self._evaluate_condition(condition, current_state, context):
return False
return True
def _evaluate_condition(self, condition: str, current_state: Any, context: Dict[str, Any]) -> bool:
"""评估条件"""
# 简化实现：基于条件字符串评估
if "state." in condition:
# 从状态中取值
field_name = condition.replace("state.", "")
field_value = getattr(current_state, field_name, None)
# 简单的布尔评估
if field_value is None:
return False
elif isinstance(field_value, bool):
return field_value
elif isinstance(field_value, (int, float)):
return field_value > 0
else:
return bool(field_value)
return True
def execute_side_effects(self, current_state: Any, context: Dict[str, Any]) -> Dict[str, Any]:
"""执行副作用"""
results = {}
for effect in self.side_effects:
result = self._execute_side_effect(effect, current_state, context)
results[effect] = result
return results
def _execute_side_effect(self, effect: str, current_state: Any, context: Dict[str, Any]) -> Any:
"""执行单个副作用"""
# 简化实现：基于效果字符串执行
if "log." in effect:
# 记录日志
message = effect.replace("log.", "")
print(f"状态转换日志: {message}")
return {"logged": message}
elif "update." in effect:
# 更新状态
field_update = effect.replace("update.", "")
field_name, field_value = field_update.split("=")
setattr(current_state, field_name.strip(), field_value.strip())
return {"updated": field_name.strip()}
return None
"""
return class_code
def _generate_persistence_config(self, analysis: Dict[str, Any]) -> Dict[str, Any]:
"""生成持久化配置"""
persistence = analysis["persistence_requirements"]
config = {
"enabled": persistence["needs_persistence"],
"level": persistence["persistence_level"],
"backup_frequency": persistence["backup_frequency"],
"retention_policy": persistence["retention_policy"],
"storage_backend": self._select_storage_backend(persistence),
"backup_strategy": self._select_backup_strategy(persistence)
}
return config
def _select_storage_backend(self, persistence: Dict[str, Any]) -> str:
"""选择存储后端"""
level = persistence["persistence_level"]
if level == "high":
return "redis_cluster"
elif level == "medium":
return "postgresql"
else:
return "sqlite"
def _select_backup_strategy(self, persistence: Dict[str, Any]) -> str:
"""选择备份策略"""
frequency = persistence["backup_frequency"]
if frequency == "real_time":
return "synchronous_replication"
elif frequency == "checkpoint":
return "asynchronous_backup"
else:
return "periodic_snapshot"
def validate_migration(self, langgraph_state: Dict[str, Any],
agno_state: Any) -> Dict[str, Any]:
"""验证迁移结果"""
validation = {
"is_valid": True,
"errors": [],
"warnings": [],
"compatibility_score": 0.0
}
# 字段一致性检查
field_validation = self._validate_field_consistency(langgraph_state, agno_state)
validation["errors"].extend(field_validation["errors"])
validation["warnings"].extend(field_validation["warnings"])
# 状态完整性检查
integrity_validation = self._validate_state_integrity(agno_state)
validation["errors"].extend(integrity_validation["errors"])
# 计算兼容性分数
total_checks = len(field_validation["errors"]) + len(field_validation["warnings"]) + len(integrity_validation["errors"])
if total_checks == 0:
validation["compatibility_score"] = 1.0
else:
passed_checks = total_checks - len(validation["errors"])
validation["compatibility_score"] = passed_checks / total_checks
validation["is_valid"] = len(validation["errors"]) == 0
return validation
def _validate_field_consistency(self, langgraph_state: Dict[str, Any], agno_state: Any) -> Dict[str, Any]:
"""验证字段一致性"""
errors = []
warnings = []
for key, expected_value in langgraph_state.items():
actual_value = getattr(agno_state, key, None)
# 检查字段存在性
if actual_value is None and expected_value is not None:
errors.append(f"字段 {key} 缺失")
continue
# 检查类型一致性
if type(expected_value) != type(actual_value):
warnings.append(f"字段 {key} 类型不一致: 期望 {type(expected_value)}, 实际 {type(actual_value)}")
# 检查值一致性（对于简单类型）
if isinstance(expected_value, (str, int, float, bool)):
if expected_value != actual_value:
warnings.append(f"字段 {key} 值不一致: 期望 {expected_value}, 实际 {actual_value}")
return {"errors": errors, "warnings": warnings}
def _validate_state_integrity(self, agno_state: Any) -> Dict[str, Any]:
"""验证状态完整性"""
errors = []
# 检查必需字段
required_fields = ["created_at", "updated_at", "version"]
for field in required_fields:
if not hasattr(agno_state, field):
errors.append(f"必需字段 {field} 缺失")
# 检查时间戳
if hasattr(agno_state, "created_at") and hasattr(agno_state, "updated_at"):
if agno_state.created_at > agno_state.updated_at:
errors.append("创建时间不能晚于更新时间")
# 检查版本号
if hasattr(agno_state, "version") and agno_state.version < 1:
errors.append("版本号必须大于等于1")
return {"errors": errors}
class StateMigrationRollbackManager:
"""状态迁移回滚管理器"""
def __init__(self):
self.rollback_points = {}
self.backup_states = {}
self.migration_history = []
def create_rollback_point(self, migration_id: str, original_state: Dict[str, Any]) -> str:
"""创建回滚点"""
rollback_id = f"rollback_{migration_id}_{int(time.time())}"
self.rollback_points[rollback_id] = {
"migration_id": migration_id,
"original_state": copy.deepcopy(original_state),
"created_at": time.time(),
"status": "active"
}
return rollback_id
def rollback_to_point(self, rollback_id: str) -> Dict[str, Any]:
"""回滚到指定点"""
if rollback_id not in self.rollback_points:
return {
"success": False,
"error": f"回滚点 {rollback_id} 不存在"
}
rollback_point = self.rollback_points[rollback_id]
if rollback_point["status"] != "active":
return {
"success": False,
"error": f"回滚点 {rollback_id} 已失效"
}
# 执行回滚
original_state = rollback_point["original_state"]
# 标记回滚点已使用
rollback_point["status"] = "used"
rollback_point["used_at"] = time.time()
return {
"success": True,
"original_state": original_state,
"rollback_id": rollback_id
}
def cleanup_rollback_points(self, max_age: int = 86400) -> int:
"""清理旧的回滚点"""
current_time = time.time()
cleaned_count = 0
rollback_ids_to_remove = []
for rollback_id, rollback_point in self.rollback_points.items():
if current_time - rollback_point["created_at"] > max_age:
rollback_ids_to_remove.append(rollback_id)
for rollback_id in rollback_ids_to_remove:
del self.rollback_points[rollback_id]
cleaned_count += 1
return cleaned_count
### 3.4 智能体通信机制挑战
**挑战描述：**
LangGraph中的节点通信机制与Agno智能体的消息传递机制存在差异，需要确保通信的可靠性和一致性。
**解决方案：**
class MessagePriority(Enum): LOW = 1 NORMAL = 2 HIGH = 3 CRITICAL = 4
class MessageStatus(Enum): PENDING = "pending" SENT = "sent" DELIVERED = "delivered" FAILED = "failed" RETRY = "retry"
@dataclass class BaseMessage: """基础消息类""" id: str sender: str receiver: str type: str content: Dict[str, Any] timestamp: datetime priority: MessagePriority = MessagePriority.NORMAL status: MessageStatus = MessageStatus.PENDING retry_count: int = 0 max_retries: int = 3 metadata: Dict[str, Any] = None def __post_init__(self): if self.metadata is None: self.metadata = {} def can_retry(self) -> bool: """检查是否可以重试""" return self.retry_count < self.max_retries def increment_retry(self): """增加重试次数""" self.retry_count += 1 def to_dict(self) -> Dict[str, Any]: """转换为字典""" return { "id": self.id, "sender": self.sender, "receiver": self.receiver, "type": self.type, "content": self.content, "timestamp": self.timestamp.isoformat(), "priority": self.priority.value, "status": self.status.value, "retry_count": self.retry_count, "max_retries": self.max_retries, "metadata": self.metadata } """ # 为每种消息类型生成特定的类 for msg_type, type_info in message_types.items(): class_name = f"{msg_type.title().replace('_', '')}Message" class_code += f"""
@dataclass class {class_name}(BaseMessage): """{msg_type.replace('_', ' ').title()}消息类""" def __post_init__(self): super().__post_init__() self.type = "{msg_type}" """ return class_code def _generate_communication_protocol(self, analysis: Dict[str, Any]) -> str: """生成通信协议代码""" patterns = analysis["communication_patterns"] protocol_code = """ from typing import Dict, Any, List, Optional, Callable from abc import ABC, abstractmethod import asyncio import json
class CommunicationProtocol(ABC): """通信协议抽象基类""" @abstractmethod async def send_message(self, message: BaseMessage) -> bool: """发送消息""" pass @abstractmethod async def receive_message(self, timeout: float = 30.0) -> Optional[BaseMessage]: """接收消息""" pass @abstractmethod async def broadcast_message(self, message: BaseMessage, recipients: List[str]) -> Dict[str, bool]: """广播消息""" pass
class ReliableCommunicationProtocol(CommunicationProtocol): """可靠通信协议""" def __init__(self, retry_config: Dict[str, Any] = None): self.retry_config = retry_config or { "max_retries": 3, "retry_delay": 1.0, "exponential_backoff": True, "timeout": 30.0 } self.pending_messages = {} self.message_handlers = {} async def send_message(self, message: BaseMessage) -> bool: """发送消息（带重试机制）""" try: # 尝试发送消息 success = await self._attempt_send(message) if success: message.status = MessageStatus.SENT return True else: # 处理发送失败 return await self._handle_send_failure(message) except Exception as e: # 记录错误并尝试重试 message.metadata["error"] = str(e) return await self._handle_send_failure(message) async def _attempt_send(self, message: BaseMessage) -> bool: """尝试发送消息""" # 这里实现实际的发送逻辑 # 简化实现：模拟发送 await asyncio.sleep(0.1) # 模拟网络延迟 return True # 假设发送成功 async def _handle_send_failure(self, message: BaseMessage) -> bool: """处理发送失败""" if message.can_retry(): message.increment_retry() # 计算重试延迟 delay = self.retry_config["retry_delay"] if self.retry_config.get("exponential_backoff"): delay *= (2 ** message.retry_count) # 等待后重试 await asyncio.sleep(delay) return await self.send_message(message) else: message.status = MessageStatus.FAILED return False async def receive_message(self, timeout: float = 30.0) -> Optional[BaseMessage]: """接收消息""" # 简化实现：返回模拟消息 await asyncio.sleep(0.1) return BaseMessage( id="test_message", sender="test_sender", receiver="test_receiver", type="test", content={"test": "data"}, timestamp=datetime.now() ) async def broadcast_message(self, message: BaseMessage, recipients: List[str]) -> Dict[str, bool]: """广播消息""" results = {} # 并发发送给所有接收者 tasks = [] for recipient in recipients: recipient_message = BaseMessage( id=f"{message.id}_{recipient}", sender=message.sender, receiver=recipient, type=message.type, content=message.content.copy(), timestamp=message.timestamp, priority=message.priority ) tasks.append(self.send_message(recipient_message)) # 等待所有发送完成 send_results = await asyncio.gather(*tasks, return_exceptions=True) for i, recipient in enumerate(recipients): result = send_results[i] if isinstance(result, Exception): results[recipient] = False else: results[recipient] = result return results """ return protocol_code def _generate_routing_system(self, analysis: Dict[str, Any]) -> str: """生成路由系统代码""" routing = analysis["routing_mechanisms"] routing_code = """ from typing import Dict, Any, List, Optional, Callable from abc import ABC, abstractmethod import re
class MessageRouter(ABC): """消息路由器抽象基类""" @abstractmethod def route_message(self, message: BaseMessage) -> List[str]: """路由消息""" pass
class RuleBasedMessageRouter(MessageRouter): """基于规则的消息路由器""" def __init__(self): self.routing_rules = [] self.agent_registry = {} def add_routing_rule(self, rule: Dict[str, Any]): """添加路由规则""" self.routing_rules.append(rule) def register_agent(self, agent_id: str, capabilities: List[str], metadata: Dict[str, Any] = None): """注册智能体""" self.agent_registry[agent_id] = { "capabilities": capabilities, "metadata": metadata or {}, "status": "active" } def route_message(self, message: BaseMessage) -> List[str]: """基于规则路由消息""" suitable_agents = [] # 遍历所有注册的智能体 for agent_id, agent_info in self.agent_registry.items(): if agent_info["status"] != "active": continue # 检查智能体能力是否匹配消息需求 if self._agent_can_handle_message(agent_id, agent_info, message): suitable_agents.append(agent_id) # 应用路由规则 final_agents = self._apply_routing_rules(suitable_agents, message) return final_agents def _agent_can_handle_message(self, agent_id: str, agent_info: Dict[str, Any], message: BaseMessage) -> bool: """检查智能体是否能处理消息""" # 基于消息类型和能力匹配 message_type = message.type capabilities = agent_info["capabilities"] # 简单的关键字匹配 for capability in capabilities: if capability.lower() in message_type.lower() or message_type.lower() in capability.lower(): return True return False def _apply_routing_rules(self, agents: List[str], message: BaseMessage) -> List[str]: """应用路由规则""" final_agents = agents.copy() for rule in self.routing_rules: rule_type = rule.get("type", "filter") if rule_type == "filter": final_agents = self._apply_filter_rule(final_agents, message, rule) elif rule_type == "priority": final_agents = self._apply_priority_rule(final_agents, message, rule) elif rule_type == "load_balance": final_agents = self._apply_load_balance_rule(final_agents, message, rule) return final_agents def _apply_filter_rule(self, agents: List[str], message: BaseMessage, rule: Dict[str, Any]) -> List[str]: """应用过滤规则""" filtered_agents = [] for agent in agents: # 检查过滤条件 if self._meets_filter_conditions(agent, message, rule.get("conditions", [])): filtered_agents.append(agent) return filtered_agents def _apply_priority_rule(self, agents: List[str], message: BaseMessage, rule: Dict[str, Any]) -> List[str]: """应用优先级规则""" if not agents: return agents # 基于优先级排序 priority_agents = sorted(agents, key=lambda agent: self._get_agent_priority(agent, rule)) # 返回前N个优先级最高的智能体 max_agents = rule.get("max_agents", 1) return priority_agents[:max_agents] def _apply_load_balance_rule(self, agents: List[str], message: BaseMessage, rule: Dict[str, Any]) -> List[str]: """应用负载均衡规则""" if not agents: return agents # 基于负载情况选择智能体 load_threshold = rule.get("load_threshold", 0.8) selected_agents = [] for agent in agents: agent_load = self._get_agent_load(agent) if agent_load < load_threshold: selected_agents.append(agent) # 如果没有智能体低于阈值，选择负载最低的 if not selected_agents and agents: selected_agents = [min(agents, key=self._get_agent_load)] return selected_agents def _meets_filter_conditions(self, agent: str, message: BaseMessage, conditions: List[Dict[str, Any]]) -> bool: """检查是否满足过滤条件""" for condition in conditions: if not self._evaluate_condition(agent, message, condition): return False return True def _evaluate_condition(self, agent: str, message: BaseMessage, condition: Dict[str, Any]) -> bool: """评估条件""" condition_type = condition.get("type", "field_match") if condition_type == "field_match": field = condition.get("field") value = condition.get("value") if field == "message_type": return message.type == value elif field == "sender": return message.sender == value elif field == "priority": return message.priority.value >= value elif condition_type == "capability_match": required_capability = condition.get("capability") agent_capabilities = self.agent_registry.get(agent, {}).get("capabilities", []) return required_capability in agent_capabilities return True def _get_agent_priority(self, agent: str, rule: Dict[str, Any]) -> int: """获取智能体优先级""" priority_config = rule.get("priority_config", {}) # 基于能力匹配度计算优先级 agent_capabilities = self.agent_registry.get(agent, {}).get("capabilities", []) priority_capabilities = priority_config.get("capabilities", []) match_score = 0 for cap in agent_capabilities: if cap in priority_capabilities: match_score += priority_capabilities[cap] return -match_score # 负值用于排序（分数越高优先级越高） def _get_agent_load(self, agent: str) -> float: """获取智能体负载""" # 简化实现：返回模拟负载 # 实际实现应该查询智能体的实际负载情况 return 0.5 # 假设负载为50% """ return routing_code def _generate_error_handling(self, analysis: Dict[str, Any]) -> str: """生成错误处理代码""" error_analysis = analysis["error_handling"] error_code = f""" from typing import Dict, Any, List, Optional, Callable from abc import ABC, abstractmethod import logging import traceback
class CommunicationErrorHandler(ABC): """通信错误处理器抽象基类""" @abstractmethod async def handle_error(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]: """处理错误""" pass
class DefaultCommunicationErrorHandler(CommunicationErrorHandler): """默认通信错误处理器""" def __init__(self): self.error_strategies = {{ "timeout": self._handle_timeout_error, "connection": self._handle_connection_error, "validation": self._handle_validation_error, "processing": self._handle_processing_error, "unknown": self._handle_unknown_error }} self.logger = logging.getLogger(__name__) async def handle_error(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]: """处理错误""" error_type = self._classify_error(error) handler = self.error_strategies.get(error_type, self._handle_unknown_error) try: result = await handler(error, context) return result except Exception as handler_error: self.logger.error(f"错误处理器失败: {{handler_error}}") return {{ "success": False, "error": "错误处理失败", "original_error": str(error), "handler_error": str(handler_error) }} def _classify_error(self, error: Exception) -> str: """分类错误""" error_message = str(error).lower() if any(keyword in error_message for keyword in ["timeout", "timed out", "time out"]): return "timeout" elif any(keyword in error_message for keyword in ["connection", "connect", "network"]): return "connection" elif any(keyword in error_message for keyword in ["validation", "invalid", "format"]): return "validation" elif any(keyword in error_message for keyword in ["processing", "process", "execute"]): return "processing" else: return "unknown" async def _handle_timeout_error(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]: """处理超时错误""" self.logger.warning(f"处理超时错误: {{error}}") # 检查是否可以重试 message = context.get("message") if message and message.can_retry(): message.increment_retry() return {{ "success": False, "error": "timeout", "retry_suggested": True, "retry_delay": 2 ** message.retry_count, "message": str(error) }} else: return {{ "success": False, "error": "timeout", "retry_suggested": False, "message": str(error) }} async def _handle_connection_error(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]: """处理连接错误""" self.logger.warning(f"处理连接错误: {{error}}") return {{ "success": False, "error": "connection", "retry_suggested": True, "retry_delay": 5, "circuit_breaker_suggested": True, "message": str(error) }} async def _handle_validation_error(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]: """处理验证错误""" self.logger.warning(f"处理验证错误: {{error}}") return {{ "success": False, "error": "validation", "retry_suggested": False, "fix_required": True, "message": str(error) }} async def _handle_processing_error(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]: """处理处理错误""" self.logger.error(f"处理处理错误: {{error}}") return {{ "success": False, "error": "processing", "retry_suggested": False, "rollback_suggested": True, "message": str(error), "stack_trace": traceback.format_exc() }} async def _handle_unknown_error(self, error: Exception, context: Dict[str, Any]) -> Dict[str, Any]: """处理未知错误""" self.logger.error(f"处理未知错误: {{error}}") return {{ "success": False, "error": "unknown", "retry_suggested": False, "investigation_required": True, "message": str(error), "stack_trace": traceback.format_exc() }} """ return error_code def _generate_monitoring_system(self, analysis: Dict[str, Any]) -> str: """生成监控系统代码""" performance = analysis["performance_characteristics"] monitoring_code = f""" from typing import Dict, Any, List, Optional from dataclasses import dataclass from datetime import datetime import time import json
@dataclass class CommunicationMetrics: """通信指标""" messages_sent: int = 0 messages_received: int = 0 messages_failed: int = 0 average_latency: float = 0.0 throughput: float = 0.0 error_rate: float = 0.0 timestamp: datetime = None def __post_init__(self): if self.timestamp is None: self.timestamp = datetime.now()
class CommunicationMonitor: """通信监控器""" def __init__(self): self.metrics_history = [] self.current_metrics = CommunicationMetrics() self.alerts = [] self.thresholds = {{ "error_rate": 0.05, # 5%错误率 "latency": 1.0, # 1秒延迟 "throughput": 100 # 每秒100条消息 }} def record_message_sent(self, message: BaseMessage): """记录消息发送""" self.current_metrics.messages_sent += 1 self._check_thresholds() def record_message_received(self, message: BaseMessage, latency: float): """记录消息接收""" self.current_metrics.messages_received += 1 # 更新平均延迟 if self.current_metrics.messages_received == 1: self.current_metrics.average_latency = latency else: self.current_metrics.average_latency = ( (self.current_metrics.average_latency * (self.current_metrics.messages_received - 1) + latency) / self.current_metrics.messages_received ) self._check_thresholds() def record_message_failed(self, message: BaseMessage, error_type: str): """记录消息失败""" self.current_metrics.messages_failed += 1 self._update_error_rate() self._check_thresholds() # 记录警报 self._create_alert(f"消息失败: {{error_type}}", "error", {{ "message_id": message.id, "error_type": error_type, "retry_count": message.retry_count }}) def _update_error_rate(self): """更新错误率""" total_messages = (self.current_metrics.messages_sent + self.current_metrics.messages_received) if total_messages > 0: self.current_metrics.error_rate = (self.current_metrics.messages_failed / total_messages) def _check_thresholds(self): """检查阈值""" # 检查错误率 if self.current_metrics.error_rate > self.thresholds["error_rate"]: self._create_alert(f"错误率过高: {{self.current_metrics.error_rate:.2%}}", "warning", {{"error_rate": self.current_metrics.error_rate}}) # 检查延迟 if self.current_metrics.average_latency > self.thresholds["latency"]: self._create_alert(f"延迟过高: {{self.current_metrics.average_latency:.2f}}s", "warning", {{"latency": self.current_metrics.average_latency}}) def _create_alert(self, message: str, severity: str, data: Dict[str, Any]): """创建警报""" alert = {{ "message": message, "severity": severity, "data": data, "timestamp": datetime.now(), "acknowledged": False }} self.alerts.append(alert) # 保持警报历史 if len(self.alerts) > 1000: self.alerts = self.alerts[-1000:] def get_snapshot(self) -> Dict[str, Any]: """获取监控快照""" return {{ "current_metrics": self.current_metrics.__dict__, "alert_count": len([a for a in self.alerts if not a["acknowledged"]]), "recent_alerts": self.alerts[-10:], "status": self._determine_status() }} def _determine_status(self) -> str: """确定系统状态""" if self.current_metrics.error_rate > 0.1: # 10%错误率 return "critical" elif self.current_metrics.error_rate > 0.05: # 5%错误率 return "warning" elif self.current_metrics.average_latency > 2.0: # 2秒延迟 return "degraded" else: return "healthy" def reset_metrics(self): """重置指标""" # 保存当前指标到历史 self.metrics_history.append(self.current_metrics) # 保持历史记录 if len(self.metrics_history) > 1000: self.metrics_history = self.metrics_history[-1000:] # 创建新的指标对象 self.current_metrics = CommunicationMetrics() def get_performance_report(self) -> Dict[str, Any]: """获取性能报告""" if not self.metrics_history: return {{"error": "没有足够的历史数据"}} # 计算趋势 recent_metrics = self.metrics_history[-10:] error_rates = [m.error_rate for m in recent_metrics] latencies = [m.average_latency for m in recent_metrics] throughputs = [m.throughput for m in recent_metrics] return {{ "period": "最近10个指标周期", "error_rate_trend": {{ "current": self.current_metrics.error_rate, "average": sum(error_rates) / len(error_rates), "trend": "increasing" if self.current_metrics.error_rate > sum(error_rates) / len(error_rates) else "decreasing" }}, "latency_trend": {{ "current": self.current_metrics.average_latency, "average": sum(latencies) / len(latencies), "trend": "increasing" if self.current_metrics.average_latency > sum(latencies) / len(latencies) else "decreasing" }}, "throughput_trend": {{ "current": self.current_metrics.throughput, "average": sum(throughputs) / len(throughputs), "trend": "increasing" if self.current_metrics.throughput > sum(throughputs) / len(throughputs) else "decreasing" }}, "recommendations": self._generate_recommendations() }} def _generate_recommendations(self) -> List[str]: """生成建议""" recommendations = [] if self.current_metrics.error_rate > 0.05: recommendations.append("错误率较高，建议检查通信链路") if self.current_metrics.average_latency > 1.0: recommendations.append("延迟较高，建议优化网络配置") if self.current_metrics.messages_failed > 10: recommendations.append("失败消息较多，建议检查消息格式和处理逻辑") return recommendations """ return monitoring_code # 基于不一致类型生成建议