Merge pull request #173 from crestalnetwork/feat/acolyt-integration

Feat: Acolyt gpt chat integration

app/core/engine.py

@@ -40,6 +40,7 @@
 from models.db import get_pool, get_session
 from models.skill import AgentSkillData, ThreadSkillData
 from skill_sets import get_skill_set
+from skills.acolyt import get_Acolyt_skill
 from skills.common import get_common_skill
 from skills.crestal import get_crestal_skill
 from skills.enso import get_enso_skill
@@ -205,6 +206,20 @@ async def initialize_agent(aid):
                 tools.append(s)
             except Exception as e:
                 logger.warning(e)
+    # Acolyt skills
+    if agent.acolyt_skills and len(agent.acolyt_skills) > 0 and agent.acolyt_config:
+        for skill in agent.acolyt_skills:
+            try:
+                s = get_Acolyt_skill(
+                    skill,
+                    agent.acolyt_config.get("api_key"),
+                    skill_store,
+                    agent_store,
+                    aid,
+                )
+                tools.append(s)
+            except Exception as e:
+                logger.warning(e)
     # Twitter skills
     twitter_prompt = ""
     if agent.twitter_skills and len(agent.twitter_skills) > 0:

docs/create_agent.sh

@@ -85,6 +85,12 @@ ENSO_CONFIG='{
 }'
 ENSO_SKILLS='["get_tokens"]'
 
+# Acolyt settings (optional)
+ACOLYT_CONFIG='{
+  "api_key": ""
+}'
+ACOLYT_SKILLS='["ask_gpt"]'
+
 # Twitter settings (optional)
 # If you don't need to use the twitter skills, you can remove it in TWITTER_SKILLS
 TWITTER_ENTRYPOINT_ENABLED=false
@@ -135,6 +141,8 @@ JSON_DATA=$(cat << EOF
   "enso_enabled": $ENSO_ENABLED,
   "enso_config": $ENSO_CONFIG,
   "enso_skills": $ENSO_SKILLS,
+  "acolyt_config": $ACOLYT_CONFIG,
+  "acolyt_skills": $ACOLYT_SKILLS,
   "twitter_enabled": $TWITTER_ENTRYPOINT_ENABLED,
   "twitter_entrypoint_enabled": $TWITTER_ENTRYPOINT_ENABLED,
   "twitter_config": $TWITTER_CONFIG,

models/agent.py

@@ -149,6 +149,17 @@ class Agent(SQLModel, table=True):
         sa_column=Column(JSONB, nullable=True),
         description="Enso integration configuration settings",
     )
+    # Acolyt skills
+    acolyt_skills: Optional[List[str]] = Field(
+        default=None,
+        sa_column=Column(ARRAY(String)),
+        description="List of Acolyt-specific skills available to this agent",
+    )
+    acolyt_config: Optional[dict] = Field(
+        default=None,
+        sa_column=Column(JSONB, nullable=True),
+        description="Acolyt integration configuration settings",
+    )
     # skill set
     skill_sets: Optional[Dict[str, Dict[str, Any]]] = Field(
         default=None,

skills/acolyt/__init__.py

@@ -0,0 +1,27 @@
+"""Acolyt skills."""
+
+from abstracts.skill import SkillStoreABC
+from skills.acolyt.ask import AcolytAskGpt
+from skills.acolyt.base import AcolytBaseTool
+
+
+def get_Acolyt_skill(
+    name: str,
+    api_key: str,
+    skill_store: SkillStoreABC,
+    agent_store: SkillStoreABC,
+    agent_id: str,
+) -> AcolytBaseTool:
+    if not api_key:
+        raise ValueError("Acolyt API token is empty")
+
+    if name == "ask_gpt":
+        return AcolytAskGpt(
+            api_key=api_key,
+            agent_id=agent_id,
+            skill_store=skill_store,
+            agent_store=agent_store,
+        )
+
+    else:
+        raise ValueError(f"Unknown Acolyt skill: {name}")

skills/acolyt/ask.py

@@ -0,0 +1,130 @@
+from typing import Type
+
+import httpx
+from langchain.tools.base import ToolException
+from pydantic import BaseModel, Field
+from typing_extensions import Literal
+
+from skills.acolyt.base import AcolytBaseTool
+
+from .base import base_url
+
+
+class AcolytAskGptInput(BaseModel):
+    question: str
+
+
+class InputMessage(BaseModel):
+    role: Literal["system", "user", "assistant", "tool", "function"] = Field(
+        "user", description="The role of the message sender."
+    )
+    content: str
+
+
+class AcolytAskGptRequest(BaseModel):
+    messages: list[InputMessage]
+    model: str | None = Field("gpt-4o", description="The AI model to be used.")
+    stream: bool | None = Field(
+        False, description="To request for response of type stream."
+    )
+    temperature: float | None = Field(
+        0.7,
+        le=2,
+        ge=0,
+        description="Controls the degree of randomness in the generated text.",
+    )
+
+
+class OutputMessage(BaseModel):
+    content: str | None = Field(
+        None,
+        description="The output content of the question response from acolyt API call.",
+    )
+
+
+class OutputChoices(BaseModel):
+    finish_reason: Literal[
+        "stop", "length", "tool_calls", "content_filter", "function_call"
+    ] = Field(description="The reason of GPT method stop.")
+    message: OutputMessage
+
+
+class AcolytAskGptOutput(BaseModel):
+    choices: list[OutputChoices]
+
+
+class AcolytAskGpt(AcolytBaseTool):
+    """
+    This tool allows users to ask questions which are then sent to the Acolyt API. this should be run if the user requests to
+    ask Acolyt explicitly.
+    The API response is processed and summarized before being returned to the user.
+
+
+    Attributes:
+        name (str): Name of the tool, specifically "acolyt_ask_gpt".
+        description (str): Comprehensive description of the tool's purpose and functionality.
+        args_schema (Type[BaseModel]): Schema for input arguments, specifying expected parameters.
+    """
+
+    name: str = "acolyt_ask_gpt"
+    description: str = (
+        """
+        This tool allows users to ask questions which are then sent to the Acolyt API. this should be run if the user requests to
+        ask Acolyt explicitly.
+        The API response is processed and summarized before being returned to the user.
+        """
+    )
+    args_schema: Type[BaseModel] = AcolytAskGptInput
+
+    def _run(self, question: str) -> AcolytAskGptOutput:
+        """Run the tool to get the tokens and APYs from the API.
+
+        Returns:
+             AcolytAskOutput: A structured output containing output of Acolyt chat completion API.
+
+        Raises:
+            Exception: If there's an error accessing the Acolyt API.
+        """
+        raise NotImplementedError("Use _arun instead")
+
+    async def _arun(self, question: str) -> AcolytAskGptOutput:
+        """Run the tool to get answer from Acolyt GPT.
+        Args:
+            question (str): The question body from user.
+        Returns:
+            AcolytAskOutput: A structured output containing output of Acolyt chat completion API.
+
+        Raises:
+            Exception: If there's an error accessing the Acolyt API.
+        """
+        url = f"{base_url}/api/chat/completions"
+        headers = {
+            "accept": "application/json",
+            "Authorization": f"Bearer {self.api_key}",
+        }
+
+        body = AcolytAskGptRequest(
+            messages=[InputMessage(content=question)],
+        ).model_dump(exclude_none=True)
+
+        async with httpx.AsyncClient() as client:
+            try:
+                response = await client.post(
+                    url, headers=headers, timeout=30, json=body
+                )
+                response.raise_for_status()
+                json_dict = response.json()
+
+                res = AcolytAskGptOutput(**json_dict)
+
+                return res
+            except httpx.RequestError as req_err:
+                raise ToolException(
+                    f"request error from Acolyt API: {req_err}"
+                ) from req_err
+            except httpx.HTTPStatusError as http_err:
+                raise ToolException(
+                    f"http error from Acolyt API: {http_err}"
+                ) from http_err
+            except Exception as e:
+                raise ToolException(f"error from Acolyt API: {e}") from e

skills/acolyt/base.py

@@ -0,0 +1,25 @@
+from typing import Type
+
+from pydantic import BaseModel, Field
+
+from abstracts.agent import AgentStoreABC
+from abstracts.skill import IntentKitSkill, SkillStoreABC
+
+base_url = "https://acolyt-oracle-poc.vercel.app"
+
+
+class AcolytBaseTool(IntentKitSkill):
+    """Base class for Acolyt tools."""
+
+    api_key: str = Field(description="API Key")
+
+    name: str = Field(description="The name of the tool")
+    description: str = Field(description="A description of what the tool does")
+    args_schema: Type[BaseModel]
+    agent_id: str = Field(description="The ID of the agent")
+    agent_store: AgentStoreABC = Field(
+        description="The agent store for persisting data"
+    )
+    skill_store: SkillStoreABC = Field(
+        description="The skill store for persisting data"
+    )