company-ai-platform/app/modules/ai_agent/adapters/openclaw_hermes.py

from typing import Any

from fastapi import HTTPException, status

from app.core.config import Settings
from app.modules.ai_agent.adapters.base import AIAdapter
from app.modules.ai_agent.adapters.common import (
    _error_detail,
)
from app.modules.ai_agent.constants import (
    OPENCLAW_HERMES_PIPELINE,
    AIDefault,
    AIContextKey,
    AIErrorKey,
    AIMemoryMode,
    AIProviderName,
    AIResponseKey,
)
from app.modules.ai_agent.skills import AISkillId, get_ai_skill

from app.modules.ai_agent.adapters.hermes import HermesAdapter
from app.modules.ai_agent.adapters.openclaw import OpenClawAdapter

class OpenClawHermesAdapter(AIAdapter):
    """Compose OpenClaw Gateway context with Hermes Agent answers."""

    provider_name = AIProviderName.OPENCLAW_HERMES

    def __init__(self, settings: Settings):
        self.openclaw = OpenClawAdapter(settings)
        self.hermes = HermesAdapter(settings)

    def ask(self, prompt: str, context: dict[str, Any] | None = None) -> dict[str, Any]:
        base_context = context or {}
        recall = self._recall_memory(prompt, base_context)
        openclaw = self._openclaw_context(base_context)
        hermes_context = {
            **base_context,
            AIContextKey.AGENT_PIPELINE: self.provider_name,
            AIContextKey.HERMES_MEMORY: recall[AIResponseKey.ANSWER],
            AIContextKey.OPENCLAW: openclaw,
        }
        hermes_result = self.hermes.ask(prompt, hermes_context)
        remember = self._remember_interaction(
            prompt,
            base_context,
            hermes_result[AIResponseKey.ANSWER],
        )
        return {
            AIResponseKey.ANSWER: hermes_result[AIResponseKey.ANSWER],
            AIResponseKey.RAW: {
                AIResponseKey.PIPELINE: OPENCLAW_HERMES_PIPELINE,
                AIResponseKey.HERMES_RECALL: recall,
                AIResponseKey.OPENCLAW: openclaw,
                AIResponseKey.HERMES_ANSWER: hermes_result.get(AIResponseKey.RAW, {}),
                AIResponseKey.HERMES_REMEMBER: remember,
            },
        }

    def _openclaw_context(self, context: dict[str, Any]) -> dict[str, Any]:
        result: dict[str, Any] = {AIResponseKey.TOOL_INVOKED: False}
        try:
            result[AIResponseKey.HEALTH] = self.openclaw.health()
        except Exception as exc:
            result[AIResponseKey.HEALTH] = {
                AIResponseKey.OK: False,
                AIResponseKey.ERROR: _error_detail(exc),
            }

        tool = context.get(AIContextKey.OPENCLAW_TOOL)
        if not tool:
            return result

        try:
            result[AIResponseKey.TOOL_INVOKED] = True
            result[AIResponseKey.TOOL] = self.openclaw.invoke_tool(
                tool=str(tool),
                action=str(context.get(AIContextKey.OPENCLAW_ACTION) or AIDefault.ACTION_JSON),
                args=context.get(AIContextKey.OPENCLAW_ARGS) or {},
                session_key=str(
                    context.get(AIContextKey.OPENCLAW_SESSION_KEY)
                    or AIDefault.SESSION_KEY_MAIN
                ),
            )
        except HTTPException:
            raise
        except Exception as exc:
            raise HTTPException(
                status_code=status.HTTP_502_BAD_GATEWAY,
                detail={AIErrorKey.OPENCLAW: _error_detail(exc)},
            ) from exc
        return result

    def _recall_memory(self, prompt: str, context: dict[str, Any]) -> dict[str, Any]:
        recall_skill = get_ai_skill(AISkillId.HERMES_MEMORY_RECALL)
        try:
            result = self.hermes.ask(
                recall_skill.render(),
                {
                    AIContextKey.MODE: AIMemoryMode.RECALL,
                    AIContextKey.USER_PROMPT: prompt,
                    AIContextKey.REQUEST_CONTEXT: context,
                },
            )
        except Exception as exc:  # Hermes memory should not block OpenClaw execution.
            return {
                AIResponseKey.ANSWER: "",
                AIResponseKey.RAW: {},
                AIResponseKey.ERROR: _error_detail(exc),
            }
        return {
            AIResponseKey.ANSWER: result[AIResponseKey.ANSWER],
            AIResponseKey.RAW: result.get(AIResponseKey.RAW, {}),
        }

    def _remember_interaction(
        self,
        prompt: str,
        context: dict[str, Any],
        answer: str,
    ) -> dict[str, Any]:
        remember_skill = get_ai_skill(AISkillId.HERMES_MEMORY_WRITE)
        try:
            result = self.hermes.ask(
                remember_skill.render(),
                {
                    AIContextKey.MODE: AIMemoryMode.WRITE,
                    AIContextKey.USER_PROMPT: prompt,
                    AIContextKey.REQUEST_CONTEXT: context,
                    AIContextKey.ASSISTANT_ANSWER: answer,
                },
            )
        except Exception as exc:
            return {
                AIResponseKey.OK: False,
                AIResponseKey.RAW: {},
                AIResponseKey.ERROR: _error_detail(exc),
            }
        return {
            AIResponseKey.OK: True,
            AIResponseKey.RAW: result.get(AIResponseKey.RAW, {}),
            AIResponseKey.ANSWER: result[AIResponseKey.ANSWER],
        }