Backup-bdg/OpenHands
0
1import asyncio2import os3import uuid4from datetime import datetime, timezone5from typing import Callable6 7import openhands8from openhands.core.config.mcp_config import MCPConfig9from openhands.core.logger import openhands_logger as logger10from openhands.events.action.agent import RecallAction11from openhands.events.event import Event, EventSource, RecallType12from openhands.events.observation.agent import (13 MicroagentKnowledge,14 RecallObservation,15)16from openhands.events.observation.empty import NullObservation17from openhands.events.stream import EventStream, EventStreamSubscriber18from openhands.microagent import (19 BaseMicroagent,20 KnowledgeMicroagent,21 RepoMicroagent,22 load_microagents_from_dir,23)24from openhands.runtime.base import Runtime25from openhands.utils.prompt import (26 ConversationInstructions,27 RepositoryInfo,28 RuntimeInfo,29)30 31GLOBAL_MICROAGENTS_DIR = os.path.join(32 os.path.dirname(os.path.dirname(openhands.__file__)),33 'microagents',34)35 36 37class Memory:38 """39 Memory is a component that listens to the EventStream for information retrieval actions40 (a RecallAction) and publishes observations with the content (such as RecallObservation).41 """42 43 sid: str44 event_stream: EventStream45 status_callback: Callable | None46 loop: asyncio.AbstractEventLoop | None47 48 def __init__(49 self,50 event_stream: EventStream,51 sid: str,52 status_callback: Callable | None = None,53 ):54 self.event_stream = event_stream55 self.sid = sid if sid else str(uuid.uuid4())56 self.status_callback = status_callback57 self.loop = None58 59 self.event_stream.subscribe(60 EventStreamSubscriber.MEMORY,61 self.on_event,62 self.sid,63 )64 65 # Additional placeholders to store user workspace microagents66 self.repo_microagents: dict[str, RepoMicroagent] = {}67 self.knowledge_microagents: dict[str, KnowledgeMicroagent] = {}68 69 # Store repository / runtime info to send them to the templating later70 self.repository_info: RepositoryInfo | None = None71 self.runtime_info: RuntimeInfo | None = None72 self.conversation_instructions: ConversationInstructions | None = None73 74 # Load global microagents (Knowledge + Repo)75 # from typically OpenHands/microagents (i.e., the PUBLIC microagents)76 self._load_global_microagents()77 78 def on_event(self, event: Event):79 """Handle an event from the event stream."""80 asyncio.get_event_loop().run_until_complete(self._on_event(event))81 82 async def _on_event(self, event: Event):83 """Handle an event from the event stream asynchronously."""84 try:85 if isinstance(event, RecallAction):86 # if this is a workspace context recall (on first user message)87 # create and add a RecallObservation88 # with info about repo, runtime, instructions, etc. including microagent knowledge if any89 if (90 event.source == EventSource.USER91 and event.recall_type == RecallType.WORKSPACE_CONTEXT92 ):93 logger.debug('Workspace context recall')94 workspace_obs: RecallObservation | NullObservation | None = None95 96 workspace_obs = self._on_workspace_context_recall(event)97 if workspace_obs is None:98 workspace_obs = NullObservation(content='')99 100 # important: this will release the execution flow from waiting for the retrieval to complete101 workspace_obs._cause = event.id # type: ignore[union-attr]102 103 self.event_stream.add_event(workspace_obs, EventSource.ENVIRONMENT)104 return105 106 # Handle knowledge recall (triggered microagents)107 # Allow triggering from both user and agent messages108 elif (109 event.source == EventSource.USER110 or event.source == EventSource.AGENT111 ) and event.recall_type == RecallType.KNOWLEDGE:112 logger.debug(113 f'Microagent knowledge recall from {event.source} message'114 )115 microagent_obs: RecallObservation | NullObservation | None = None116 microagent_obs = self._on_microagent_recall(event)117 if microagent_obs is None:118 microagent_obs = NullObservation(content='')119 120 # important: this will release the execution flow from waiting for the retrieval to complete121 microagent_obs._cause = event.id # type: ignore[union-attr]122 123 self.event_stream.add_event(microagent_obs, EventSource.ENVIRONMENT)124 return125 except Exception as e:126 error_str = f'Error: {str(e.__class__.__name__)}'127 logger.error(error_str)128 self.send_error_message('STATUS$ERROR_MEMORY', error_str)129 return130 131 def _on_workspace_context_recall(132 self, event: RecallAction133 ) -> RecallObservation | None:134 """Add repository and runtime information to the stream as a RecallObservation.135 136 This method collects information from all available repo microagents and concatenates their contents.137 Multiple repo microagents are supported, and their contents will be concatenated with newlines between them.138 """139 140 # Create WORKSPACE_CONTEXT info:141 # - repository_info142 # - runtime_info143 # - repository_instructions144 # - microagent_knowledge145 146 # Collect raw repository instructions147 repo_instructions = ''148 149 # Retrieve the context of repo instructions from all repo microagents150 for microagent in self.repo_microagents.values():151 if repo_instructions:152 repo_instructions += '\n\n'153 repo_instructions += microagent.content154 155 # Find any matched microagents based on the query156 microagent_knowledge = self._find_microagent_knowledge(event.query)157 158 # Create observation if we have anything159 if (160 self.repository_info161 or self.runtime_info162 or repo_instructions163 or microagent_knowledge164 or self.conversation_instructions165 ):166 obs = RecallObservation(167 recall_type=RecallType.WORKSPACE_CONTEXT,168 repo_name=self.repository_info.repo_name169 if self.repository_info and self.repository_info.repo_name is not None170 else '',171 repo_directory=self.repository_info.repo_directory172 if self.repository_info173 and self.repository_info.repo_directory is not None174 else '',175 repo_instructions=repo_instructions if repo_instructions else '',176 runtime_hosts=self.runtime_info.available_hosts177 if self.runtime_info and self.runtime_info.available_hosts is not None178 else {},179 additional_agent_instructions=self.runtime_info.additional_agent_instructions180 if self.runtime_info181 and self.runtime_info.additional_agent_instructions is not None182 else '',183 microagent_knowledge=microagent_knowledge,184 content='Added workspace context',185 date=self.runtime_info.date if self.runtime_info is not None else '',186 custom_secrets_descriptions=self.runtime_info.custom_secrets_descriptions187 if self.runtime_info is not None188 else {},189 conversation_instructions=self.conversation_instructions.content190 if self.conversation_instructions is not None191 else '',192 )193 return obs194 return None195 196 def _on_microagent_recall(197 self,198 event: RecallAction,199 ) -> RecallObservation | None:200 """When a microagent action triggers microagents, create a RecallObservation with structured data."""201 202 # Find any matched microagents based on the query203 microagent_knowledge = self._find_microagent_knowledge(event.query)204 205 # Create observation if we have anything206 if microagent_knowledge:207 obs = RecallObservation(208 recall_type=RecallType.KNOWLEDGE,209 microagent_knowledge=microagent_knowledge,210 content='Retrieved knowledge from microagents',211 )212 return obs213 return None214 215 def _find_microagent_knowledge(self, query: str) -> list[MicroagentKnowledge]:216 """Find microagent knowledge based on a query.217 218 Args:219 query: The query to search for microagent triggers220 221 Returns:222 A list of MicroagentKnowledge objects for matched triggers223 """224 recalled_content: list[MicroagentKnowledge] = []225 226 # skip empty queries227 if not query:228 return recalled_content229 230 # Search for microagent triggers in the query231 for name, microagent in self.knowledge_microagents.items():232 trigger = microagent.match_trigger(query)233 if trigger:234 logger.info("Microagent '%s' triggered by keyword '%s'", name, trigger)235 recalled_content.append(236 MicroagentKnowledge(237 name=microagent.name,238 trigger=trigger,239 content=microagent.content,240 )241 )242 return recalled_content243 244 def load_user_workspace_microagents(245 self, user_microagents: list[BaseMicroagent]246 ) -> None:247 """248 This method loads microagents from a user's cloned repo or workspace directory.249 250 This is typically called from agent_session or setup once the workspace is cloned.251 """252 logger.info(253 'Loading user workspace microagents: %s', [m.name for m in user_microagents]254 )255 for user_microagent in user_microagents:256 if isinstance(user_microagent, KnowledgeMicroagent):257 self.knowledge_microagents[user_microagent.name] = user_microagent258 elif isinstance(user_microagent, RepoMicroagent):259 self.repo_microagents[user_microagent.name] = user_microagent260 261 def _load_global_microagents(self) -> None:262 """263 Loads microagents from the global microagents_dir264 """265 repo_agents, knowledge_agents = load_microagents_from_dir(266 GLOBAL_MICROAGENTS_DIR267 )268 for name, agent in knowledge_agents.items():269 if isinstance(agent, KnowledgeMicroagent):270 self.knowledge_microagents[name] = agent271 for name, agent in repo_agents.items():272 if isinstance(agent, RepoMicroagent):273 self.repo_microagents[name] = agent274 275 def get_microagent_mcp_tools(self) -> list[MCPConfig]:276 """277 Get MCP tools from all repo microagents (always active)278 279 Returns:280 A list of MCP tools configurations from microagents281 """282 mcp_configs: list[MCPConfig] = []283 284 # Check all repo microagents for MCP tools (always active)285 for agent in self.repo_microagents.values():286 if agent.metadata.mcp_tools:287 mcp_configs.append(agent.metadata.mcp_tools)288 logger.debug(289 f'Found MCP tools in repo microagent {agent.name}: {agent.metadata.mcp_tools}'290 )291 292 return mcp_configs293 294 def set_repository_info(self, repo_name: str, repo_directory: str) -> None:295 """Store repository info so we can reference it in an observation."""296 if repo_name or repo_directory:297 self.repository_info = RepositoryInfo(repo_name, repo_directory)298 else:299 self.repository_info = None300 301 def set_runtime_info(302 self,303 runtime: Runtime,304 custom_secrets_descriptions: dict[str, str],305 ) -> None:306 """Store runtime info (web hosts, ports, etc.)."""307 # e.g. { '127.0.0.1': 8080 }308 utc_now = datetime.now(timezone.utc)309 date = str(utc_now.date())310 311 if runtime.web_hosts or runtime.additional_agent_instructions:312 self.runtime_info = RuntimeInfo(313 available_hosts=runtime.web_hosts,314 additional_agent_instructions=runtime.additional_agent_instructions,315 date=date,316 custom_secrets_descriptions=custom_secrets_descriptions,317 )318 else:319 self.runtime_info = RuntimeInfo(320 date=date,321 custom_secrets_descriptions=custom_secrets_descriptions,322 )323 324 def set_conversation_instructions(325 self, conversation_instructions: str | None326 ) -> None:327 """328 Set contextual information for conversation329 This is information the agent may require330 """331 self.conversation_instructions = ConversationInstructions(332 content=conversation_instructions or ''333 )334 335 def send_error_message(self, message_id: str, message: str):336 """Sends an error message if the callback function was provided."""337 if self.status_callback:338 try:339 if self.loop is None:340 self.loop = asyncio.get_running_loop()341 asyncio.run_coroutine_threadsafe(342 self._send_status_message('error', message_id, message), self.loop343 )344 except RuntimeError as e:345 logger.error(346 f'Error sending status message: {e.__class__.__name__}',347 stack_info=False,348 )349 350 async def _send_status_message(self, msg_type: str, id: str, message: str):351 """Sends a status message to the client."""352 if self.status_callback:353 self.status_callback(msg_type, id, message)354