multi-agent-memory-framework

問題

マルチエージェントシステムは、非効率な情報共有と調整に悩まされています。エージェントは計算処理を重複させることが多いか、集合知を活用できず終わります。従来のアプローチでは、共有メモリ(調整を可能にする)と個々のエージェント自律性(並列化を可能にする)の間のバランスを効果的に取ることができません。

ソリューション

BMAMを実装します。これは脳に着想を得たメモリフレームワークで、マルチエージェントメモリを階層的に構造化します：

1. ワーキングメモリ: エージェント間で共有される短期的で高容量の状態。即座の調整に使用されます
2. エピソディックメモリ: エージェント相互作用、判断、結果の永続的な記録
3. セマンティックメモリ: チーム全体で共有される抽象的な知識と学習パターン
4. メモリの統合: 関連性と頻度に基づいてメモリタイプ間で情報を移動させるメカニズム

このアプローチは神経科学モデルを反映しながら、効率的なマルチエージェント調整を実現します。

使用する場合

• 共同作業を実行するマルチエージェントシステム(研究、計画、問題解決)
• エージェント専門化を維持しながら共有知識が必要なシナリオ
• 継続的な学習が必要な長時間実行するエージェントシステム
• 調整と並列実行の両方が必要な複雑なタスク
• エージェントが互いの発見に基づいて構築するシステム

使用しない場合

• シングルエージェントシステム(オーバーヘッドが正当化されません)
• 厳しいメモリ/レイテンシの制約があるシナリオ
• 競争的または敵対的なマルチエージェント設定
• 統合なしで即座の応答が必要なタスク

実装

ステップ1: メモリアーキテクチャを設計する

認知科学に着想を得た階層的メモリ構造を作成します。

class BrainInspiredMemory:
    """
    Hierarchical memory system for multi-agent coordination
    Inspired by working, episodic, and semantic memory in neuroscience
    """

    def __init__(self, num_agents):
        # Working Memory: High-capacity, short-lived state
        self.working_memory = {
            "current_observations": {},  # Latest observations from each agent
            "recent_actions": deque(maxlen=100),  # Last 100 actions across agents
            "shared_goals": [],
            "active_subtasks": {}
        }

        # Episodic Memory: Historical records of agent interactions
        self.episodic_memory = {
            "interaction_history": [],  # (agent_a, agent_b, action, outcome)
            "decision_outcomes": [],    # (agent, decision, outcome, success)
            "problem_solutions": {}     # problem_id -> successful_solutions
        }

        # Semantic Memory: Abstracted knowledge and learned patterns
        self.semantic_memory = {
            "agent_capabilities": {},  # agent_id -> capabilities
            "task_strategies": {},     # task_type -> effective_strategies
            "learned_relationships": {},  # entity -> related_entities
            "domain_knowledge": {}     # abstracted domain facts
        }

        self.num_agents = num_agents
        self.consolidation_counter = 0

    def record_observation(self, agent_id, observation):
        """Add agent observation to working memory"""
        self.working_memory["current_observations"][agent_id] = {
            "data": observation,
            "timestamp": time.time()
        }

    def record_action(self, agent_id, action, result):
        """Log action execution"""
        action_record = {
            "agent": agent_id,
            "action": action,
            "result": result,
            "timestamp": time.time()
        }
        self.working_memory["recent_actions"].append(action_record)

    def record_interaction(self, agent_a, agent_b, action, outcome):
        """Log multi-agent interaction"""
        interaction = {
            "agents": (agent_a, agent_b),
            "action": action,
            "outcome": outcome,
            "timestamp": time.time()
        }
        self.episodic_memory["interaction_history"].append(interaction)

    def get_agent_working_context(self, agent_id):
        """Retrieve relevant working memory for an agent"""
        context = {
            "own_observation": self.working_memory["current_observations"].get(agent_id),
            "recent_actions": list(self.working_memory["recent_actions"])[-10:],
            "shared_goals": self.working_memory["shared_goals"],
            "relevant_subtasks": self.working_memory["active_subtasks"].get(agent_id, [])
        }
        return context

ステップ2: メモリ統合を実装する

関連性に基づいてワーキングメモリからエピソディック、セマンティックメモリに情報を移動させます。

class MemoryConsolidation:
    """Manage information movement through memory hierarchy"""

    def consolidate_memory(self, memory_system):
        """
        Periodically consolidate working memory to episodic/semantic
        Similar to sleep-based memory consolidation in brains
        """
        # Step 1: Extract decision patterns from recent actions
        recent_actions = list(memory_system.working_memory["recent_actions"])

        decision_patterns = self.extract_decision_patterns(recent_actions)

        # Step 2: Identify frequently successful action sequences
        successful_sequences = self.identify_frequent_patterns(
            memory_system.episodic_memory["decision_outcomes"],
            min_success_rate=0.7,
            min_frequency=3
        )

        # Step 3: Move successful patterns to semantic memory (learned strategies)
        for sequence, success_rate in successful_sequences:
            task_type = self.infer_task_type(sequence)
            if task_type not in memory_system.semantic_memory["task_strategies"]:
                memory_system.semantic_memory["task_strategies"][task_type] = []

            memory_system.semantic_memory["task_strategies"][task_type].append({
                "strategy": sequence,
                "success_rate": success_rate,
                "learned_at": time.time()
            })

        # Step 4: Extract agent capability profiles
        for agent_id in range(memory_system.num_agents):
            capabilities = self.extract_agent_capabilities(
                memory_system.episodic_memory["decision_outcomes"],
                agent_id
            )
            memory_system.semantic_memory["agent_capabilities"][agent_id] = capabilities

        # Step 5: Prune old entries from working memory
        memory_system.working_memory["recent_actions"] = deque(
            list(memory_system.working_memory["recent_actions"])[-50:],
            maxlen=100
        )

    def extract_decision_patterns(self, actions):
        """Find repeated decision patterns"""
        patterns = {}
        for action in actions:
            action_type = action["action"]["type"]
            if action_type not in patterns:
                patterns[action_type] = 0
            patterns[action_type] += 1

        return sorted(patterns.items(), key=lambda x: x[1], reverse=True)

    def identify_frequent_patterns(self, decision_outcomes, min_success_rate=0.7, min_frequency=3):
        """Extract successful action sequences that should become learned strategies"""
        sequence_success = {}

        for outcome in decision_outcomes:
            decision = outcome["decision"]
            success = outcome["success"]

            # Convert decision to sequence representation
            seq_key = tuple(decision) if isinstance(decision, list) else (decision,)

            if seq_key not in sequence_success:
                sequence_success[seq_key] = {"success": 0, "total": 0}

            sequence_success[seq_key]["total"] += 1
            if success:
                sequence_success[seq_key]["success"] += 1

        # Filter by frequency and success rate
        frequent_patterns = [
            (seq, data["success"] / data["total"])
            for seq, data in sequence_success.items()
            if data["total"] >= min_frequency and (data["success"] / data["total"]) >= min_success_rate
        ]

        return frequent_patterns

    def extract_agent_capabilities(self, decision_outcomes, agent_id):
        """Build capability profile for an agent"""
        agent_outcomes = [
            o for o in decision_outcomes
            if o["agent"] == agent_id
        ]

        capabilities = {}
        for outcome in agent_outcomes:
            task_type = self.infer_task_type(outcome["decision"])
            if task_type not in capabilities:
                capabilities[task_type] = {"success": 0, "total": 0}

            capabilities[task_type]["total"] += 1
            if outcome["success"]:
                capabilities[task_type]["success"] += 1

        # Convert to success rates
        return {
            task: data["success"] / data["total"]
            for task, data in capabilities.items()
            if data["total"] >= 2
        }

ステップ3: 学習知識用のセマンティックメモリを実装する

エージェント全体で学習パターンと関係を保存および取得します。

class SemanticMemoryManager:
    """Manage semantic/abstract knowledge across agents"""

    def __init__(self, semantic_memory):
        self.semantic_memory = semantic_memory

    def get_best_strategy_for_task(self, task_type, agent_capabilities=None):
        """
        Retrieve learned strategy for a task, optionally filtered by agent capability
        """
        if task_type not in self.semantic_memory["task_strategies"]:
            return None

        strategies = self.semantic_memory["task_strategies"][task_type]
        strategies.sort(key=lambda s: s["success_rate"], reverse=True)

        if agent_capabilities:
            # Prefer strategies the agent is good at
            best_fit = None
            for strategy in strategies:
                if agent_capabilities.get(task_type, 0) > 0.5:
                    best_fit = strategy
                    break
            return best_fit or strategies[0]

        return strategies[0]

    def find_capable_agent(self, task_type, min_capability=0.6):
        """Find which agent is best suited for a task"""
        capabilities = self.semantic_memory["agent_capabilities"]

        best_agent = None
        best_score = min_capability

        for agent_id, agent_caps in capabilities.items():
            score = agent_caps.get(task_type, 0)
            if score > best_score:
                best_score = score
                best_agent = agent_id

        return best_agent

    def get_related_knowledge(self, concept):
        """Find related domain knowledge"""
        if concept in self.semantic_memory["learned_relationships"]:
            return self.semantic_memory["learned_relationships"][concept]

        return []

ステップ4: メモリを使用したマルチエージェント動作の調整

メモリを使用してエージェント調整をガイドします。

class MultiAgentCoordinator:
    """Use brain-inspired memory for agent coordination"""

    def __init__(self, memory_system):
        self.memory = memory_system
        self.semantic_manager = SemanticMemoryManager(memory_system.semantic_memory)

    def assign_task_to_capable_agent(self, task_type):
        """
        Find best agent for task using semantic memory
        """
        capable_agent = self.semantic_manager.find_capable_agent(task_type, min_capability=0.5)

        if not capable_agent:
            # Fall back to first available agent
            capable_agent = 0

        return capable_agent

    def get_learned_strategy_for_agent(self, agent_id, task_type):
        """
        Provide agent with learned strategy for task
        """
        agent_capabilities = self.memory.semantic_memory["agent_capabilities"].get(agent_id, {})

        strategy = self.semantic_manager.get_best_strategy_for_task(
            task_type,
            agent_capabilities
        )

        return strategy

    def record_team_decision(self, agents_involved, decision, outcome):
        """
        Log multi-agent collaborative decision
        """
        for agent_id in agents_involved:
            self.memory.record_action(agent_id, decision, outcome)

        # Cross-agent interaction
        if len(agents_involved) > 1:
            for i in range(len(agents_involved) - 1):
                self.memory.record_interaction(
                    agents_involved[i],
                    agents_involved[i+1],
                    decision,
                    outcome
                )

    def coordinate_multi_step_task(self, agents, task_sequence):
        """
        Execute complex task requiring multiple agents
        Use memory for coordination
        """
        results = []

        for step_idx, task in enumerate(task_sequence):
            # Find best agent using semantic memory
            assigned_agent = self.assign_task_to_capable_agent(task["type"])

            # Provide learned strategy if available
            strategy = self.get_learned_strategy_for_agent(assigned_agent, task["type"])

            # Execute with strategy
            outcome = agents[assigned_agent].execute(task, strategy)
            results.append(outcome)

            # Record interaction
            self.record_team_decision([assigned_agent], task, outcome)

            # Consolidate memory periodically
            if step_idx % 10 == 0:
                consolidator = MemoryConsolidation()
                consolidator.consolidate_memory(self.memory)

        return results

神経科学の主要な洞察

• ワーキングメモリ: 即座の調整を可能にします。容量は限定的(約100項目)
• エピソディックメモリ: 何が起こったか、いつ、誰と、を記録します。履歴から学習することを可能にします
• セマンティックメモリ: 抽象的な事実と戦略。効率的な知識表現
• 統合: 頻繁に使用されるエピソディック知識をセマンティックストレージに定期的に移動させます

メリット

• エージェントは直接通信なしで共有メモリを通じて調整できます
• 学習戦略は統合を通じて時間の経過とともに改善されます
• 複数のエージェント間での効率的な知識再利用
• スケーラブル: エージェントを追加しても再訓練が不要です

参考文献

• arXiv:2601.20465: BMAM脳に着想を得たマルチエージェントメモリフレームワーク
• 人間のワーキングメモリ、エピソディックメモリ、セマンティックメモリの神経科学モデルに基づく