Conceptual Framework Implementation (Python)

The implementation below illustrates the core mechanisms of K2-Agent: the SRLR loop for declarative knowledge evolution, the C-GRPO algorithm for procedural skill acquisition, and the closed-loop co-evolution system. This educational code demonstrates the key architectural concepts described in the paper.

# ─────────────────────────────────────────────────────────────────────────────
# K2-Agent: Co-Evolving Know-What and Know-How for Hierarchical Mobile Control
# Wu, Mo, Lu, Xing et al. · ICLR 2026
# Conceptual implementation of SRLR loop and C-GRPO components
# ─────────────────────────────────────────────────────────────────────────────

import torch
import torch.nn as nn
import torch.nn.functional as F
from typing import List, Dict, Tuple, Optional
from dataclasses import dataclass
from enum import Enum
import numpy as np


# ─── Section 1: Declarative Knowledge Representation ─────────────────────────

class KnowledgeType(Enum):
    """Types of declarative knowledge extracted from demonstrations."""
    STEP_ORDERING = "step_ordering"
    UI_LAYOUT = "ui_layout"
    PARAMETER_CONSTRAINT = "parameter_constraint"
    RECOVERY_STRATEGY = "recovery_strategy"


@dataclass
class KnowledgeItem:
    """Single item in the declarative knowledge base."""
    knowledge_type: KnowledgeType
    description: str
    priority: int = 0  # For highlighting critical constraints
    verified: bool = False


class DeclarativeKnowledgeBase:
    """
    Maintains explicit, language-based task knowledge.
    Evolves through SRLR loop iterations.
    """
    
    def __init__(self):
        self.items: List[KnowledgeItem] = []
        self.revision_history: List[Dict] = []
    
    def summarize_from_demonstration(self, 
                                      trajectory: List[Dict], 
                                      task_goal: str,
                                      vlm_model) -> None:
        """
        Stage 1 of SRLR: Initialize knowledge from single demonstration.
        
        Args:
            trajectory: List of (state, action, instruction) tuples
            task_goal: Natural language task description
            vlm_model: Vision-language model for distillation
        """
        # Use VLM to extract structured knowledge from trajectory
        prompt = self._construct_summarize_prompt(trajectory, task_goal)
        distilled = vlm_model.generate(prompt)
        
        # Parse into structured knowledge items
        self.items = self._parse_knowledge(distilled)
        self.revision_history.append({
            "stage": "summarize",
            "item_count": len(self.items)
        })
    
    def reflect(self, 
                executed_trajectory: List[Dict],
                success: bool,
                vlm_model) -> Optional[str]:
        """
        Stage 2 of SRLR: Analyze execution for deviations.
        
        Returns:
            failure_explanation: Root cause analysis if failed, None if success
        """
        if success:
            # Mark verified items
            for item in self.items:
                item.verified = True
            return None
        
        # Task-level reflection: identify root cause
        prompt = self._construct_reflect_prompt(executed_trajectory)
        failure_explanation = vlm_model.generate(prompt)
        
        # Step-level reflection: verify each action outcome
        for i, step in enumerate(executed_trajectory):
            expected = self._get_expected_outcome(i)
            actual = step['outcome']
            if not self._verify_outcome(expected, actual):
                self.revision_history.append({
                    "stage": "reflect",
                    "step": i,
                    "deviation": True
                })
        
        return failure_explanation
    
    def locate(self,
               executed_trajectory: List[Dict],
               failure_explanation: str,
               vlm_model) -> Optional[int]:
        """
        Stage 3 of SRLR: Pinpoint first failure point.
        
        Returns:
            failure_step: Index of first failed step, or None
        """
        for t, step in enumerate(executed_trajectory):
            if not self._verify_step(step, t, vlm_model):
                self.revision_history.append({
                    "stage": "locate",
                    "failure_step": t
                })
                return t
        return None
    
    def revise(self,
               failure_step: int,
               failure_explanation: str,
               executed_trajectory: List[Dict],
               vlm_model) -> None:
        """
        Stage 4 of SRLR: Apply atomic edits to knowledge base.
        
        Atomic operators: Add, Delete, Update, Highlight
        """
        # Generate revision plan
        prompt = self._construct_revise_prompt(
            failure_step, failure_explanation, self.items
        )
        revision_plan = vlm_model.generate(prompt)
        
        # Apply atomic edits
        edits = self._parse_revision_plan(revision_plan)
        for edit in edits:
            if edit['operation'] == 'ADD':
                self.items.insert(edit['position'], edit['item'])
            elif edit['operation'] == 'DELETE':
                del self.items[edit['index']]
            elif edit['operation'] == 'UPDATE':
                self.items[edit['index']].description = edit['new_description']
            elif edit['operation'] == 'HIGHLIGHT':
                self.items[edit['index']].priority = 1
        
        self.revision_history.append({
            "stage": "revise",
            "edits_applied": len(edits)
        })
    
    def generate_subgoal(self, 
                         current_state,
                         task_goal: str,
                         vlm_model) -> str:
        """Generate immediate sub-goal for low-level executor."""
        context = self._construct_knowledge_context()
        prompt = f"Task: {task_goal}\nKnowledge: {context}\nCurrent State: {current_state}\nNext sub-goal:"
        return vlm_model.generate(prompt)


# ─── Section 2: Error-Decoupled Replay Buffers ───────────────────────────────

class ReplayBuffer:
    """Experience pool with error-type categorization."""
    
    def __init__(self, buffer_type: str, capacity: int = 10000):
        self.buffer_type = buffer_type  # 'conventional', 'type', 'param'
        self.capacity = capacity
        self.buffer = []
        self.error_rates = []
    
    def add(self, sample: Dict, error_rate: float) -> None:
        """Add sample with associated error rate."""
        self.buffer.append(sample)
        self.error_rates.append(error_rate)
        if len(self.buffer) > self.capacity:
            self.buffer.pop(0)
            self.error_rates.pop(0)
    
    def sample(self, batch_size: int) -> List[Dict]:
        """Sample batch from buffer."""
        if len(self.buffer) <= batch_size:
            return self.buffer
        indices = np.random.choice(len(self.buffer), batch_size, replace=False)
        return [self.buffer[i] for i in indices]


class ErrorDecoupledReplayManager:
    """
    Manages three replay pools for balanced sampling.
    Implements error-decoupled replay balancing from C-GRPO.
    """
    
    def __init__(self, 
                 beta_con: float = 0.5,
                 beta_type: float = 0.25,
                 beta_param: float = 0.25):
        self.buffers = {
            'conventional': ReplayBuffer('conventional'),
            'type': ReplayBuffer('type'),
            'param': ReplayBuffer('param')
        }
        self.ratios = {
            'conventional': beta_con,
            'type': beta_type,
            'param': beta_param
        }
    
    def assign_sample(self, 
                      sample: Dict,
                      eta_type: float,
                      eta_param: float) -> None:
        """
        Assign sample to appropriate buffer based on error rates.
        
        Args:
            sample: Training sample (state, action, reward)
            eta_type: Type error rate
            eta_param: Parameter error rate
        """
        if eta_type > 0.1:  # Threshold for type errors
            self.buffers['type'].add(sample, eta_type)
        elif eta_param > 0.1:  # Threshold for parameter errors
            self.buffers['param'].add(sample, eta_param)
        else:
            self.buffers['conventional'].add(sample, 0.0)
    
    def sample_balanced_batch(self, total_batch_size: int) -> List[Dict]:
        """Construct balanced mini-batch according to preset ratios."""
        batch = []
        for buffer_name, ratio in self.ratios.items():
            sub_batch_size = int(total_batch_size * ratio)
            sub_batch = self.buffers[buffer_name].sample(sub_batch_size)
            batch.extend(sub_batch)
        return batch


# ─── Section 3: Dynamic Demonstration Injection ──────────────────────────────

class CurriculumScheduler:
    """
    Implements dynamic demonstration injection for C-GRPO.
    Schedules prefix length based on training progress and sample difficulty.
    """
    
    def __init__(self,
                 max_steps: int = 1000,
                 temperature: float = 0.5,
                 full_demo_length: int = 10):
        self.K_max = max_steps
        self.T = temperature
        self.L = full_demo_length
        self.current_step = 0
    
    def sigma(self, k: int) -> float:
        """Linear annealing scheduler."""
        return max(0.0, 1.0 - k / self.K_max)
    
    def f_gate(self, difficulty: float) -> float:
        """Difficulty-gating function."""
        return np.tanh(difficulty / self.T)
    
    def compute_injection_length(self,
                                  sample_difficulty: float) -> int:
        """
        Calculate number of expert actions to inject.
        
        l = L * sigma(k) * f_gate(d_i)
        """
        l = self.L * self.sigma(self.current_step) * self.f_gate(sample_difficulty)
        return int(l)
    
    def get_demo_prefix(self,
                        full_demonstration: List[Dict],
                        sample_difficulty: float) -> List[Dict]:
        """Extract prefix of expert actions for injection."""
        length = self.compute_injection_length(sample_difficulty)
        return full_demonstration[:length]
    
    def step(self) -> None:
        """Advance training step counter."""
        self.current_step += 1


# ─── Section 4: C-GRPO Training ──────────────────────────────────────────────

class CGRPOTrainer:
    """
    Curriculum-Guided Group Relative Policy Optimization.
    Trains low-level executor with error-decoupled replay and dynamic injection.
    """
    
    def __init__(self,
                 policy_model: nn.Module,
                 replay_manager: ErrorDecoupledReplayManager,
                 curriculum_scheduler: CurriculumScheduler,
                 G: int = 8,  # Number of rollouts per sample
                 epsilon_clip: float = 0.2,
                 kl_weight: float = 0.04):
        self.policy = policy_model
        self.replay_manager = replay_manager
        self.scheduler = curriculum_scheduler
        self.G = G
        self.epsilon_clip = epsilon_clip
        self.kl_weight = kl_weight
        self.old_policy = None
    
    def estimate_error_rates(self,
                             state,
                             expert_action,
                             device) -> Tuple[float, float]:
        """
        Estimate type and parameter error rates via multiple rollouts.
        
        Returns:
            eta_type: Fraction of samples with wrong action type
            eta_param: Fraction with correct type but wrong parameters
        """
        type_errors = 0
        param_errors = 0
        
        with torch.no_grad():
            for _ in range(self.G):
                action = self.policy.sample_action(state)
                
                # Check type match
                if action['type'] != expert_action['type']:
                    type_errors += 1
                else:
                    # Check parameter accuracy
                    coord_diff = torch.norm(
                        torch.tensor(action['coordinate']) - 
                        torch.tensor(expert_action['coordinate'])
                    )
                    if coord_diff >= 10:  # epsilon threshold
                        param_errors += 1
        
        eta_type = type_errors / self.G
        eta_param = param_errors / self.G
        return eta_type, eta_param
    
    def compute_grpo_loss(self,
                          batch: List[Dict],
                          optimizer) -> float:
        """
        Compute C-GRPO objective with group-relative advantages.
        
        J_CGRPO = E[min(r_t * A_t, clip(r_t) * A_t)]
        """
        total_loss = 0.0
        
        for sample in batch:
            state = sample['state']
            action = sample['action']
            reward = sample['reward']
            
            # Generate G rollouts for group-relative baseline
            rollout_rewards = []
            log_probs = []
            
            for _ in range(self.G):
                pred_action, log_prob = self.policy(state)
                r = self._compute_reward(pred_action, action)
                rollout_rewards.append(r)
                log_probs.append(log_prob)
            
            # Compute advantage as deviation from group mean
            mean_reward = np.mean(rollout_rewards)
            std_reward = np.std(rollout_rewards) + 1e-8
            advantage = (reward - mean_reward) / std_reward
            
            # Compute importance ratio and clipped objective
            old_log_prob = self.old_policy.get_log_prob(state, action)
            ratio = torch.exp(log_prob - old_log_prob)
            
            clipped_ratio = torch.clamp(ratio, 
                                        1 - self.epsilon_clip,
                                        1 + self.epsilon_clip)
            
            loss = -torch.min(ratio * advantage, clipped_ratio * advantage)
            total_loss += loss
        
        optimizer.zero_grad()
        total_loss.backward()
        optimizer.step()
        
        return total_loss.item()
    
    def train_step(self, optimizer) -> float:
        """Single C-GRPO training step."""
        # Sample balanced batch
        batch = self.replay_manager.sample_balanced_batch(batch_size=32)
        
        # Augment with dynamic demonstration injection
        augmented_batch = []
        for sample in batch:
            difficulty = sample.get('eta_type', 0) + sample.get('eta_param', 0)
            prefix = self.scheduler.get_demo_prefix(
                sample['demonstration'], difficulty
            )
            sample['injected_prefix'] = prefix
            augmented_batch.append(sample)
        
        # Update policy
        loss = self.compute_grpo_loss(augmented_batch, optimizer)
        
        # Advance curriculum
        self.scheduler.step()
        
        return loss


# ─── Section 5: K2-Agent Integration ─────────────────────────────────────────

class K2Agent:
    """
    Complete K2-Agent integrating high-level planner and low-level executor.
    Implements closed-loop co-evolution of declarative and procedural knowledge.
    """
    
    def __init__(self,
                 high_level_vlm,  # Qwen-2.5-VL-72B for planner
                 low_level_vlm,   # Qwen-2.5-VL-7B for executor
                 n_srlr_iterations: int = 3):
        self.planner = DeclarativeKnowledgeBase()
        self.high_level_vlm = high_level_vlm
        self.low_level_executor = low_level_vlm
        self.n_srlr = n_srlr_iterations
        
        # Initialize C-GRPO components
        self.replay_manager = ErrorDecoupledReplayManager()
        self.scheduler = CurriculumScheduler()
        self.trainer = CGRPOTrainer(
            low_level_vlm, self.replay_manager, self.scheduler
        )
    
    def bootstrap_from_demonstration(self,
                                     demonstration: List[Dict],
                                     task_goal: str) -> None:
        """
        Initialize agent from single demonstration.
        Summarize stage of SRLR loop.
        """
        self.planner.summarize_from_demonstration(
            demonstration, task_goal, self.high_level_vlm
        )
        self.demonstration = demonstration
    
    def execute_with_feedback(self,
                              task_goal: str,
                              environment,
                              max_steps: int = 50) -> Tuple[List[Dict], bool]:
        """
        Execute task with closed-loop feedback.
        
        Returns:
            trajectory: Executed steps
            success: Whether task completed successfully
        """
        trajectory = []
        state = environment.reset()
        
        for step in range(max_steps):
            # High-level planner generates sub-goal
            subgoal = self.planner.generate_subgoal(
                state, task_goal, self.high_level_vlm
            )
            
            # Low-level executor performs action
            action = self.low_level_executor.predict(state, subgoal)
            
            # Execute in environment
            next_state, reward, done, info = environment.step(action)
            
            trajectory.append({
                'state': state,
                'subgoal': subgoal,
                'action': action,
                'reward': reward,
                'next_state': next_state
            })
            
            if done:
                return trajectory, True
            
            state = next_state
        
        return trajectory, False
    
    def srlr_iteration(self,
                       trajectory: List[Dict],
                       success: bool,
                       task_goal: str) -> None:
        """
        Run one SRLR iteration to refine declarative knowledge.
        """
        # Reflect
        failure_exp = self.planner.reflect(
            trajectory, success, self.high_level_vlm
        )
        
        if not success and failure_exp:
            # Locate
            failure_step = self.planner.locate(
                trajectory, failure_exp, self.high_level_vlm
            )
            
            # Revise
            if failure_step is not None:
                self.planner.revise(
                    failure_step, failure_exp, trajectory, self.high_level_vlm
                )
    
    def co_evolve(self,
                  task_goal: str,
                  environment,
                  n_cycles: int = 10) -> None:
        """
        Run closed-loop co-evolution.
        Pattern: (SRLR_H)^n -> C-GRPO_L
        """
        for cycle in range(n_cycles):
            print(f"=== Co-evolution Cycle {cycle + 1}/{n_cycles} ===")
            
            # Phase 1: SRLR iterations on high-level planner
            for i in range(self.n_srlr):
                trajectory, success = self.execute_with_feedback(
                    task_goal, environment
                )
                self.srlr_iteration(trajectory, success, task_goal)
                
                if success:
                    print(f"  Task succeeded at SRLR iteration {i+1}")
                    break
            
            # Phase 2: C-GRPO training on low-level executor
            print("  Training low-level executor with C-GRPO...")
            # Collect experiences and train...
    
    def evaluate(self, task_goal: str, environment) -> bool:
        """Evaluate agent on task."""
        trajectory, success = self.execute_with_feedback(task_goal, environment)
        return success


# ─── Section 6: Demonstration ────────────────────────────────────────────────

if __name__ == "__main__":
    """
    Demonstration of K2-Agent framework.
    Shows initialization, co-evolution, and evaluation.
    """
    
    print("=" * 70)
    print("K2-Agent: Co-Evolving Know-What and Know-How")
    print("ICLR 2026 - Wu, Mo, Lu, Xing et al.")
    print("=" * 70)
    
    # Configuration
    TASK_GOAL = "Create a new contact with name 'John' and phone '1234567890'"
    
    # Mock VLM models (replace with actual Qwen-2.5-VL implementations)
    class MockVLM:
        def generate(self, prompt: str) -> str:
            return "Mock response"
        
        def sample_action(self, state):
            return {'type': 'click', 'coordinate': [100, 200]}
        
        def predict(self, state, subgoal):
            return {'type': 'click', 'coordinate': [100, 200]}
        
        def get_log_prob(self, state, action):
            return torch.tensor(0.0)
    
    # Initialize agent
    print("\n[1] Initializing K2-Agent...")
    high_level_vlm = MockVLM()  # Qwen-2.5-VL-72B
    low_level_vlm = MockVLM()   # Qwen-2.5-VL-7B
    
    agent = K2Agent(high_level_vlm, low_level_vlm, n_srlr_iterations=3)
    
    # Mock demonstration
    demonstration = [
        {'instruction': 'Swipe up to open app drawer', 
         'action': {'type': 'swipe', 'coordinate': [500, 1500]}},
        {'instruction': 'Tap Contacts app',
         'action': {'type': 'click', 'coordinate': [300, 800]}},
        {'instruction': 'Tap add contact button',
         'action': {'type': 'click', 'coordinate': [900, 400]}},
    ]
    
    # Bootstrap from single demonstration
    print("[2] Bootstrapping from single demonstration...")
    agent.bootstrap_from_demonstration(demonstration, TASK_GOAL)
    print(f"    Knowledge base initialized with {len(agent.planner.items)} items")
    
    # Mock environment
    class MockEnvironment:
        def reset(self):
            return "initial_screen"
        
        def step(self, action):
            return "next_screen", 0.5, False, {}
    
    env = MockEnvironment()
    
    # Run co-evolution
    print("\n[3] Running co-evolution...")
    # agent.co_evolve(TASK_GOAL, env, n_cycles=5)
    
    # Show architecture summary
    print("\n[4] K2-Agent Architecture Summary:")
    print("    High-Level Planner (π_H):")
    print("      - Model: Qwen-2.5-VL-72B (training-free)")
    print("      - Knowledge: Declarative, symbolic, self-evolving via SRLR")
    print("      - Input: Task goal + current state")
    print("      - Output: Sub-goals for executor")
    print("    Low-Level Executor (π_L):")
    print("      - Model: Qwen-2.5-VL-7B (trainable)")
    print("      - Knowledge: Procedural, parametric, learned via C-GRPO")
    print("      - Input: Observation + sub-goal")
    print("      - Output: Atomic UI actions (click, swipe, type)")
    print("    Co-Evolution: Closed-loop feedback between layers")
    
    print("\n" + "=" * 70)
    print("K2-Agent enables human-like cognitive architecture for")
    print("mobile device control through co-evolution of know-what and know-how.")
    print("=" * 70)