Hanzo Dev + DeepSeek + Continuous Learning GRPO Integration¶

def find_maximum(input_list: list) -> int | float:
    """
    Find the maximum element in a list.

    Args:
        input_list (list): List of numbers

    Returns:
        int | float: Maximum element
    """
    if not input_list:  # Edge case: empty list
        raise ValueError("List is empty")

    max_element = input_list[0]
    for element in input_list:
        if element > max_element:
            max_element = element

    return max_element

# In your Hanzo Desktop agent code

from gym.train.grpo.api_model_adapter import DeepSeekAdapter
from gym.train.grpo.experience_manager import ExperienceManager

class HanzoDevAgent:
    def __init__(self, api_key: str):
        # DeepSeek-V3 for code generation
        self.model = DeepSeekAdapter(
            api_key=api_key,
            model="deepseek-chat",
            temperature=0.7
        )

        # Experience library (persistent across sessions)
        self.experiences = ExperienceManager(
            checkpoint_path="~/.hanzo/experiences.json"
        )

        # Load default coding experiences
        self._init_default_experiences()

    def _init_default_experiences(self):
        """Initialize with best practices for coding."""
        if len(self.experiences) == 0:
            self.experiences.add("Write type-safe code with clear annotations.")
            self.experiences.add("Handle edge cases: empty, null, single element.")
            self.experiences.add("Use meaningful variable and function names.")
            self.experiences.add("Add docstrings for functions and classes.")
            self.experiences.add("Consider performance: O(n) vs O(n²) complexity.")

    def generate_code(self, prompt: str) -> str:
        """Generate code with experience injection."""
        return self.model.generate_with_experiences(
            query=prompt,
            experiences=self.experiences.format_for_prompt(),
            temperature=0.7,
            max_tokens=2048
        )

    def learn_from_feedback(self, task: str, attempts: list, correct: str):
        """Learn from successful/failed attempts."""
        # This would integrate with Continuous Learning GRPO
        # to extract new experiences from the attempts
        pass

# Usage
agent = HanzoDevAgent(api_key="sk-82accfbadb484ea7ad986510f88d27f5")
code = agent.generate_code("Write a binary search function")
print(code)

# For continuous learning from coding tasks

from gym.train.grpo.api_model_adapter import DeepSeekAdapter
from gym.train.grpo.experience_manager import ExperienceManager
from gym.train.grpo.semantic_extractor import SemanticExtractor, LLMClient, Trajectory

class ContinuousLearningHanzoAgent:
    def __init__(self, api_key: str):
        # Target model
        self.model = DeepSeekAdapter(api_key=api_key, model="deepseek-chat")

        # Semantic extraction
        self.semantic_llm = LLMClient(
            api_key=api_key,
            base_url="https://api.deepseek.com/v1",
            model="deepseek-chat"
        )

        # Experience manager
        self.experiences = ExperienceManager(
            checkpoint_path="~/.hanzo/experiences.json"
        )

        # Semantic extractor
        self.extractor = SemanticExtractor(
            self.semantic_llm,
            max_operations=5
        )

    def solve_with_learning(self, task: str, correct_solution: str = None):
        """Solve task and learn from attempts."""

        # Generate multiple attempts (group size = 5)
        group_size = 5
        trajectories = []

        for i in range(group_size):
            # Generate solution
            response = self.model.generate_with_experiences(
                query=task,
                experiences=self.experiences.format_for_prompt(),
                temperature=0.7 + (i * 0.1)  # Vary temperature
            )

            # Compute reward (you'd use actual tests here)
            reward = self._evaluate_solution(response, correct_solution)

            # Create trajectory
            traj = Trajectory(
                query=task,
                output=response,
                reward=reward,
                groundtruth=correct_solution
            )
            trajectories.append(traj)

        # Extract experiences from trajectories
        # Stage 1: Summarize
        for traj in trajectories:
            traj.summary = self.extractor.summarize_trajectory(traj)

        # Stage 2: Extract group advantage
        operations = self.extractor.extract_group_advantage(
            trajectories,
            self.experiences.format_for_prompt()
        )

        # Stage 3: Apply operations
        self.experiences.apply_operations(operations)

        # Save updated experiences
        self.experiences.save("~/.hanzo/experiences.json")

        # Return best solution
        best_traj = max(trajectories, key=lambda t: t.reward)
        return best_traj.output

    def _evaluate_solution(self, solution: str, correct: str = None) -> float:
        """Evaluate solution quality (simplified)."""
        # You'd run actual tests here
        score = 0.0
        if "def " in solution: score += 0.3
        if "return" in solution: score += 0.3
        if ":" in solution and "\"\"\"" in solution: score += 0.2  # Has docstring
        if correct and correct in solution: score += 0.2
        return score

# Usage
agent = ContinuousLearningHanzoAgent(api_key="sk-82accfbadb484ea7ad986510f88d27f5")

task = "Write a function to check if a string is a palindrome"
solution = agent.solve_with_learning(task, correct_solution="def is_palindrome")

print("Solution:", solution)
print("Learned experiences:", len(agent.experiences))

# Train on HumanEval dataset

from datasets import load_dataset

agent = ContinuousLearningHanzoAgent(api_key="sk-xxx")

# Load HumanEval
dataset = load_dataset("openai_humaneval")

# Process first 10 problems
for i, example in enumerate(dataset["test"][:10]):
    task = example["prompt"]
    correct = example["canonical_solution"]

    print(f"\nTask {i+1}: {task[:50]}...")
    solution = agent.solve_with_learning(task, correct)
    print(f"Learned {len(agent.experiences)} experiences so far")

# After 10 tasks, the agent has learned coding patterns
# and will perform better on future tasks!