实现双系统Agent 模式

.gitignore

@@ -167,4 +167,9 @@ cython_debug/
 *.db
 
 *.pyc
-__pycache__/
+__pycache__/
+
+# Visual Studio Code 
+.vscode/
+
+tests/test_agentuniverse/unit/agent/custom_key.toml

agentuniverse/agent/default/dual_system_agent/__init__.py

@@ -0,0 +1,6 @@
+# !/usr/bin/env python3
+# -*- coding:utf-8 -*-
+
+# @Time    : 2024/1/22 16:00
+# @Author  : [Your Name]
+# @FileName: __init__.py

agentuniverse/agent/default/dual_system_agent/complexity_evaluator.py

@@ -0,0 +1,126 @@
+#!/usr/bin/env python3
+# -*- coding:utf-8 -*-
+
+"""Default complexity evaluator implementation."""
+
+import re
+from typing import Set, List, Dict
+from agentuniverse.agent.default.dual_system_agent.constants import (
+    ComplexityScore, ComplexityLevel, ComplexityFactor
+)
+from agentuniverse.agent.default.dual_system_agent.evaluator.base_evaluator import BaseEvaluator
+
+class DefaultComplexityEvaluator(BaseEvaluator):
+    """Default implementation of complexity evaluator."""
+
+    def __init__(self) -> None:
+        """Initialize the evaluator with configuration."""
+        # 简单问题的关键词集合
+        self._simple_keywords: Set[str] = {
+            '什么是', '定义', '解释', '加', '减', '乘', 
+            'what is', 'define', 'explain', 'add', 'subtract', 'multiply'
+        }
+
+        # 复杂问题的关键词集合
+        self._complex_keywords: Set[str] = {
+            '为什么', '如何', '分析', '比较', '评估', '预测', '优化',
+            'why', 'how', 'analyze', 'compare', 'evaluate', 'predict', 'optimize'
+        }
+
+        # 上下文依赖词集合
+        self._context_keywords: Set[str] = {
+            '这个', '那个', '它', '他们', '之前', '刚才',
+            'this', 'that', 'it', 'they', 'previous', 'before'
+        }
+
+    def evaluate_text_length(self, text: str) -> float:
+        """Evaluate complexity based on text length."""
+        length = len(text)
+        if length <= 10:
+            return 0.2
+        elif length <= 30:
+            return 0.5
+        elif length <= 100:
+            return 0.8
+        return 1.0
+
+    def evaluate_question_type(self, text: str) -> float:
+        """Evaluate complexity based on question type."""
+        text = text.lower()
+
+        # 检查简单关键词
+        for keyword in self._simple_keywords:
+            if keyword.lower() in text:
+                return 0.3
+
+        # 检查复杂关键词
+        for keyword in self._complex_keywords:
+            if keyword.lower() in text:
+                return 0.8
+
+        return 0.5  # 默认中等复杂度
+
+    def evaluate_keyword_complexity(self, text: str) -> float:
+        """Evaluate complexity based on keyword analysis."""
+        words = set(re.split(r'\s+|[,.，。]', text.lower()))
+
+        # 计算复杂词汇的比例
+        complex_word_count = len([w for w in words if len(w) > 4])
+        if not words:
+            return 0.5
+
+        return min(1.0, complex_word_count / len(words))
+
+    def evaluate_context_dependency(self, text: str) -> float:
+        """Evaluate complexity based on context dependency."""
+        text = text.lower()
+        context_keyword_count = sum(1 for keyword in self._context_keywords 
+                                  if keyword.lower() in text)
+
+        if context_keyword_count == 0:
+            return 0.2
+        elif context_keyword_count == 1:
+            return 0.5
+        elif context_keyword_count == 2:
+            return 0.8
+        return 1.0
+
+    def evaluate(self, text: str) -> ComplexityScore:
+        """Evaluate overall input complexity."""
+        # 计算各维度分数
+        text_length_score = self.evaluate_text_length(text)
+        question_type_score = self.evaluate_question_type(text)
+        keyword_complexity_score = self.evaluate_keyword_complexity(text)
+        context_dependency_score = self.evaluate_context_dependency(text)
+
+        # 计算总体分数（可以调整权重）
+        weights = {
+            'text_length': 0.2,
+            'question_type': 0.3,
+            'keyword_complexity': 0.3,
+            'context_dependency': 0.2
+        }
+
+        overall_score = (
+            text_length_score * weights['text_length'] +
+            question_type_score * weights['question_type'] +
+            keyword_complexity_score * weights['keyword_complexity'] +
+            context_dependency_score * weights['context_dependency']
+        )
+
+        # 确定复杂度级别
+        if overall_score < 0.4:
+            level = ComplexityLevel.SIMPLE
+        elif overall_score < 0.8:
+            level = ComplexityLevel.MODERATE
+        else:
+            level = ComplexityLevel.COMPLEX
+
+        return ComplexityScore(
+            text_length_score=text_length_score,
+            question_type_score=question_type_score,
+            keyword_complexity_score=keyword_complexity_score,
+            context_dependency_score=context_dependency_score,
+            overall_score=overall_score,
+            level=level
+        )

agentuniverse/agent/default/dual_system_agent/constants.py

@@ -0,0 +1,75 @@
+from enum import Enum
+from dataclasses import dataclass
+from typing import Any, Literal, Optional, Dict, TypedDict
+from agentuniverse.base.common.constants import AgentKeys
+
+class DualSystemKeys(str, Enum):
+    """Keys specific to dual system thinking."""
+    SYSTEM_TYPE = 'system_type'
+    CONFIDENCE = 'confidence'
+    THOUGHT = 'thought'
+    RESPONSE = 'response'
+    COMPLEXITY = 'complexity'
+
+class SystemType(str, Enum):
+    """System types for dual system thinking."""
+    FAST = 'fast_thinking'
+    SLOW = 'slow_thinking'
+
+class ThinkingState(str, Enum):
+    """States in the thinking process."""
+    EVALUATING = 'evaluating'
+    FAST_PROCESSING = 'fast_processing'
+    SLOW_PROCESSING = 'slow_processing'
+    COMPLETED = 'completed'
+
+class ComplexityLevel(str, Enum):
+    """Complexity levels for input evaluation."""
+    SIMPLE = 'simple'
+    MODERATE = 'moderate'
+    COMPLEX = 'complex'
+
+class ComplexityFactor(str, Enum):
+    """Factors affecting input complexity."""
+    TEXT_LENGTH = 'text_length'
+    QUESTION_TYPE = 'question_type'
+    KEYWORD_COMPLEXITY = 'keyword_complexity'
+    CONTEXT_DEPENDENCY = 'context_dependency'
+
+@dataclass
+class ThinkingResult:
+    """Data class for thinking results."""
+    system_type: SystemType
+    output: Any
+    confidence: float
+    thought: Optional[str] = None
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary format."""
+        return {
+            AgentKeys.OUTPUT: self.output,
+            DualSystemKeys.SYSTEM_TYPE: self.system_type,
+            DualSystemKeys.CONFIDENCE: self.confidence,
+            DualSystemKeys.THOUGHT: self.thought
+        }
+
+@dataclass
+class ComplexityScore:
+    """Data class for complexity evaluation results."""
+    text_length_score: float
+    question_type_score: float
+    keyword_complexity_score: float
+    context_dependency_score: float
+    overall_score: float
+    level: ComplexityLevel
+
+    def to_dict(self) -> Dict[str, Any]:
+        """Convert to dictionary format."""
+        return {
+            ComplexityFactor.TEXT_LENGTH: self.text_length_score,
+            ComplexityFactor.QUESTION_TYPE: self.question_type_score,
+            ComplexityFactor.KEYWORD_COMPLEXITY: self.keyword_complexity_score,
+            ComplexityFactor.CONTEXT_DEPENDENCY: self.context_dependency_score,
+            'overall_score': self.overall_score,
+            'level': self.level
+        }

agentuniverse/agent/default/dual_system_agent/dual_system_agent.py

@@ -0,0 +1,43 @@
+# !/usr/bin/env python3
+# -*- coding:utf-8 -*-
+
+# @Time    : 2024/1/22 16:00
+# @Author  : [Your Name]
+# @FileName: dual_system_agent.py
+
+from agentuniverse.agent.agent import Agent
+from agentuniverse.agent.input_object import InputObject
+from agentuniverse.agent.output_object import OutputObject
+from .constants import AgentKeys, SystemType
+import json
+import logging
+
+class DualSystemAgent(Agent):
+    """Dual System Agent class implementing fast and slow thinking."""
+
+    def input_keys(self) -> list[str]:
+        """Return the input keys of the Agent."""
+        return [AgentKeys.INPUT]
+
+    def output_keys(self) -> list[str]:
+        """Return the output keys of the Agent."""
+        return [AgentKeys.OUTPUT, AgentKeys.SYSTEM_TYPE, AgentKeys.CONFIDENCE]
+
+    def parse_input(self, input_object: InputObject, agent_input: dict) -> dict:
+        """Agent parameter parsing."""
+        agent_input[AgentKeys.INPUT] = input_object.get_data(AgentKeys.INPUT)
+        return agent_input
+
+    def parse_result(self, planner_result: dict) -> dict:
+        """Planner result parser."""
+        # Extract the system type and confidence from planner result
+        system_type: SystemType = planner_result.get(AgentKeys.SYSTEM_TYPE, 'slow')
+        confidence = planner_result.get(AgentKeys.CONFIDENCE, 0.0)
+        logging.debug(f"planner_result: {json.dumps(planner_result, ensure_ascii=False)}")
+        return {
+                AgentKeys.OUTPUT: planner_result.get(AgentKeys.OUTPUT),
+                AgentKeys.SYSTEM_TYPE: system_type,
+                AgentKeys.CONFIDENCE: confidence
+            }
+
+

agentuniverse/agent/default/dual_system_agent/dual_system_agent.yaml

@@ -0,0 +1,18 @@
+info:
+  name: 'DualSystemAgent'
+  description: 'Agent implementing fast/slow thinking systems'
+plan:
+  planner:
+    name: 'DualSystemPlanner'
+    # System 1 (Fast Thinking) Configuration
+    system1:
+      name: 'FastThinkingAgent'
+      complexity_evaluator: 'default_complexity_evaluator'
+      confidence_threshold: 0.7  # Aligned with planner threshold
+    # System 2 (Slow Thinking) Configuration
+    system2:
+      name: 'PeerAgent'
+metadata:
+  type: 'AGENT'
+  module: 'agentuniverse.agent.default.dual_system_agent.dual_system_agent'
+  class: 'DualSystemAgent'

agentuniverse/agent/default/dual_system_agent/evaluator/__init__.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+# -*- coding:utf-8 -*-
+
+"""Evaluator package initialization."""
+
+from agentuniverse.agent.default.dual_system_agent.evaluator.evaluator_manager import EvaluatorManager
+from agentuniverse.agent.default.dual_system_agent.complexity_evaluator import DefaultComplexityEvaluator
+
+# Register default evaluator
+EvaluatorManager().register("default_complexity_evaluator", DefaultComplexityEvaluator)
+EvaluatorManager().set_default("default_complexity_evaluator")

agentuniverse/agent/default/dual_system_agent/evaluator/base_evaluator.py

@@ -0,0 +1,70 @@
+#!/usr/bin/env python3
+# -*- coding:utf-8 -*-
+
+"""Base evaluator interface for complexity evaluation."""
+
+from abc import ABC, abstractmethod
+from agentuniverse.agent.default.dual_system_agent.constants import ComplexityScore
+
+class BaseEvaluator(ABC):
+    """Base interface for complexity evaluators."""
+
+    @abstractmethod
+    def evaluate(self, text: str) -> ComplexityScore:
+        """Evaluate input complexity.
+
+        Args:
+            text: Input text to evaluate
+
+        Returns:
+            ComplexityScore: Complexity evaluation results
+        """
+        pass
+
+    @abstractmethod
+    def evaluate_text_length(self, text: str) -> float:
+        """Evaluate complexity based on text length.
+
+        Args:
+            text: Input text to evaluate
+
+        Returns:
+            float: Score between 0 and 1, where higher means more complex
+        """
+        pass
+
+    @abstractmethod
+    def evaluate_question_type(self, text: str) -> float:
+        """Evaluate complexity based on question type.
+
+        Args:
+            text: Input text to evaluate
+
+        Returns:
+            float: Score between 0 and 1, where higher means more complex
+        """
+        pass
+
+    @abstractmethod
+    def evaluate_keyword_complexity(self, text: str) -> float:
+        """Evaluate complexity based on keyword analysis.
+
+        Args:
+            text: Input text to evaluate
+
+        Returns:
+            float: Score between 0 and 1, where higher means more complex
+        """
+        pass
+
+    @abstractmethod
+    def evaluate_context_dependency(self, text: str) -> float:
+        """Evaluate complexity based on context dependency.
+
+        Args:
+            text: Input text to evaluate
+
+        Returns:
+            float: Score between 0 and 1, where higher means more complex
+        """
+        pass