Backup-bdg/OpenHands

1import io2import re3from pathlib import Path4from typing import Union5 6import frontmatter7from pydantic import BaseModel8 9from openhands.core.exceptions import (10    MicroagentValidationError,11)12from openhands.core.logger import openhands_logger as logger13from openhands.microagent.types import InputMetadata, MicroagentMetadata, MicroagentType14 15 16class BaseMicroagent(BaseModel):17    """Base class for all microagents."""18 19    name: str20    content: str21    metadata: MicroagentMetadata22    source: str  # path to the file23    type: MicroagentType24 25    @classmethod26    def load(27        cls,28        path: Union[str, Path],29        microagent_dir: Path | None = None,30        file_content: str | None = None,31    ) -> 'BaseMicroagent':32        """Load a microagent from a markdown file with frontmatter.33 34        The agent's name is derived from its path relative to the microagent_dir.35        """36        path = Path(path) if isinstance(path, str) else path37 38        # Calculate derived name from relative path if microagent_dir is provided39        # Otherwise, we will rely on the name from metadata later40        derived_name = None41        if microagent_dir is not None:42            derived_name = str(path.relative_to(microagent_dir).with_suffix(''))43 44        # Only load directly from path if file_content is not provided45        if file_content is None:46            with open(path) as f:47                file_content = f.read()48 49        # Legacy repo instructions are stored in .openhands_instructions50        if path.name == '.openhands_instructions':51            return RepoMicroagent(52                name='repo_legacy',53                content=file_content,54                metadata=MicroagentMetadata(name='repo_legacy'),55                source=str(path),56                type=MicroagentType.REPO_KNOWLEDGE,57            )58 59        file_io = io.StringIO(file_content)60        loaded = frontmatter.load(file_io)61        content = loaded.content62 63        # Handle case where there's no frontmatter or empty frontmatter64        metadata_dict = loaded.metadata or {}65 66        try:67            metadata = MicroagentMetadata(**metadata_dict)68 69            # Validate MCP tools configuration if present70            if metadata.mcp_tools:71                if metadata.mcp_tools.sse_servers:72                    logger.warning(73                        f'Microagent {metadata.name} has SSE servers. Only stdio servers are currently supported.'74                    )75 76                if not metadata.mcp_tools.stdio_servers:77                    raise MicroagentValidationError(78                        f'Microagent {metadata.name} has MCP tools configuration but no stdio servers. '79                        'Only stdio servers are currently supported.'80                    )81        except Exception as e:82            # Provide more detailed error message for validation errors83            error_msg = f'Error validating microagent metadata in {path.name}: {str(e)}'84            if 'type' in metadata_dict and metadata_dict['type'] not in [85                t.value for t in MicroagentType86            ]:87                valid_types = ', '.join([f'"{t.value}"' for t in MicroagentType])88                error_msg += f'. Invalid "type" value: "{metadata_dict["type"]}". Valid types are: {valid_types}'89            raise MicroagentValidationError(error_msg) from e90 91        # Create appropriate subclass based on type92        subclass_map = {93            MicroagentType.KNOWLEDGE: KnowledgeMicroagent,94            MicroagentType.REPO_KNOWLEDGE: RepoMicroagent,95            MicroagentType.TASK: TaskMicroagent,96        }97 98        # Infer the agent type:99        # 1. If inputs exist -> TASK100        # 2. If triggers exist -> KNOWLEDGE101        # 3. Else (no triggers) -> REPO (always active)102        inferred_type: MicroagentType103        if metadata.inputs:104            inferred_type = MicroagentType.TASK105            # Add a trigger for the agent name if not already present106            trigger = f'/{metadata.name}'107            if not metadata.triggers or trigger not in metadata.triggers:108                if not metadata.triggers:109                    metadata.triggers = [trigger]110                else:111                    metadata.triggers.append(trigger)112        elif metadata.triggers:113            inferred_type = MicroagentType.KNOWLEDGE114        else:115            # No triggers, default to REPO116            # This handles cases where 'type' might be missing or defaulted by Pydantic117            inferred_type = MicroagentType.REPO_KNOWLEDGE118 119        if inferred_type not in subclass_map:120            # This should theoretically not happen with the logic above121            raise ValueError(f'Could not determine microagent type for: {path}')122 123        # Use derived_name if available (from relative path), otherwise fallback to metadata.name124        agent_name = derived_name if derived_name is not None else metadata.name125 126        agent_class = subclass_map[inferred_type]127        return agent_class(128            name=agent_name,129            content=content,130            metadata=metadata,131            source=str(path),132            type=inferred_type,133        )134 135 136class KnowledgeMicroagent(BaseMicroagent):137    """Knowledge micro-agents provide specialized expertise that's triggered by keywords in conversations.138 139    They help with:140    - Language best practices141    - Framework guidelines142    - Common patterns143    - Tool usage144    """145 146    def __init__(self, **data):147        super().__init__(**data)148        if self.type not in [MicroagentType.KNOWLEDGE, MicroagentType.TASK]:149            raise ValueError('KnowledgeMicroagent must have type KNOWLEDGE or TASK')150 151    def match_trigger(self, message: str) -> str | None:152        """Match a trigger in the message.153 154        It returns the first trigger that matches the message.155        """156        message = message.lower()157        for trigger in self.triggers:158            if trigger.lower() in message:159                return trigger160 161        return None162 163    @property164    def triggers(self) -> list[str]:165        return self.metadata.triggers166 167 168class RepoMicroagent(BaseMicroagent):169    """Microagent specialized for repository-specific knowledge and guidelines.170 171    RepoMicroagents are loaded from `.openhands/microagents/repo.md` files within repositories172    and contain private, repository-specific instructions that are automatically loaded when173    working with that repository. They are ideal for:174        - Repository-specific guidelines175        - Team practices and conventions176        - Project-specific workflows177        - Custom documentation references178    """179 180    def __init__(self, **data):181        super().__init__(**data)182        if self.type != MicroagentType.REPO_KNOWLEDGE:183            raise ValueError(184                f'RepoMicroagent initialized with incorrect type: {self.type}'185            )186 187 188class TaskMicroagent(KnowledgeMicroagent):189    """TaskMicroagent is a special type of KnowledgeMicroagent that requires user input.190 191    These microagents are triggered by a special format: "/{agent_name}"192    and will prompt the user for any required inputs before proceeding.193    """194 195    def __init__(self, **data):196        super().__init__(**data)197        if self.type != MicroagentType.TASK:198            raise ValueError(199                f'TaskMicroagent initialized with incorrect type: {self.type}'200            )201 202        # Append a prompt to ask for missing variables203        self._append_missing_variables_prompt()204 205    def _append_missing_variables_prompt(self) -> None:206        """Append a prompt to ask for missing variables."""207        # Check if the content contains any variables or has inputs defined208        if not self.requires_user_input() and not self.metadata.inputs:209            return210 211        prompt = "\n\nIf the user didn't provide any of these variables, ask the user to provide them first before the agent can proceed with the task."212        self.content += prompt213 214    def extract_variables(self, content: str) -> list[str]:215        """Extract variables from the content.216 217        Variables are in the format ${variable_name}.218        """219        pattern = r'\$\{([a-zA-Z_][a-zA-Z0-9_]*)\}'220        matches = re.findall(pattern, content)221        return matches222 223    def requires_user_input(self) -> bool:224        """Check if this microagent requires user input.225 226        Returns True if the content contains variables in the format ${variable_name}.227        """228        # Check if the content contains any variables229        variables = self.extract_variables(self.content)230        logger.debug(f'This microagent requires user input: {variables}')231        return len(variables) > 0232 233    @property234    def inputs(self) -> list[InputMetadata]:235        """Get the inputs for this microagent."""236        return self.metadata.inputs237 238 239def load_microagents_from_dir(240    microagent_dir: Union[str, Path],241) -> tuple[dict[str, RepoMicroagent], dict[str, KnowledgeMicroagent]]:242    """Load all microagents from the given directory.243 244    Note, legacy repo instructions will not be loaded here.245 246    Args:247        microagent_dir: Path to the microagents directory (e.g. .openhands/microagents)248 249    Returns:250        Tuple of (repo_agents, knowledge_agents) dictionaries251    """252    if isinstance(microagent_dir, str):253        microagent_dir = Path(microagent_dir)254 255    repo_agents = {}256    knowledge_agents = {}257 258    # Load all agents from microagents directory259    logger.debug(f'Loading agents from {microagent_dir}')260    if microagent_dir.exists():261        for file in microagent_dir.rglob('*.md'):262            # skip README.md263            if file.name == 'README.md':264                continue265            try:266                agent = BaseMicroagent.load(file, microagent_dir)267                if isinstance(agent, RepoMicroagent):268                    repo_agents[agent.name] = agent269                elif isinstance(agent, KnowledgeMicroagent):270                    # Both KnowledgeMicroagent and TaskMicroagent go into knowledge_agents271                    knowledge_agents[agent.name] = agent272            except MicroagentValidationError as e:273                # For validation errors, include the original exception274                error_msg = f'Error loading microagent from {file}: {str(e)}'275                raise MicroagentValidationError(error_msg) from e276            except Exception as e:277                # For other errors, wrap in a ValueError with detailed message278                error_msg = f'Error loading microagent from {file}: {str(e)}'279                raise ValueError(error_msg) from e280 281    logger.debug(282        f'Loaded {len(repo_agents) + len(knowledge_agents)} microagents: '283        f'{[*repo_agents.keys(), *knowledge_agents.keys()]}'284    )285    return repo_agents, knowledge_agents286