Merge pull request #28 from RSS3-Network/chore/env

chore: refactor expert loading

src/openagent/agent/function_agent.py

@@ -1,5 +1,3 @@
-import asyncio
-
 from langchain.agents import (
     AgentExecutor,
     AgentType,
@@ -23,17 +21,38 @@
 )
 from openagent.conf.env import settings
 from openagent.experts.article_expert import ArticleExpert
+from openagent.experts.dune_expert import DuneExpert
 from openagent.experts.feed_expert import FeedExpert
+from openagent.experts.google_expert import GoogleExpert
 from openagent.experts.nft_expert import NFTExpert
 from openagent.experts.price_expert import PriceExpert
-from openagent.experts.search_expert import SearchExpert
 from openagent.experts.swap_expert import SwapExpert
 from openagent.experts.transfer_expert import TransferExpert
 
 # Initialize cache
 init_cache()
 
 
+# Function to get all experts
+def get_experts():
+    experts = [
+        DuneExpert(),
+        FeedExpert(),
+        PriceExpert(),
+        ArticleExpert(),
+        TransferExpert(),
+        SwapExpert(),
+    ]
+
+    if settings.SERPAPI_API_KEY:
+        experts.append(GoogleExpert())
+
+    if settings.NFTSCAN_API_KEY:
+        experts.append(NFTExpert())
+
+    return experts
+
+
 # Function to create a ReAct agent
 def create_react_agent(session_id: str):
     # Define agent kwargs
@@ -47,15 +66,7 @@ def create_react_agent(session_id: str):
     )
 
     # List of experts to be loaded
-    experts = [
-        SearchExpert(),
-        FeedExpert(),
-        PriceExpert(),
-        ArticleExpert(),
-        NFTExpert(),
-        TransferExpert(),
-        SwapExpert(),
-    ]
+    experts = get_experts()
 
     # Initialize interpreter
     interpreter = create_interpreter(settings.MODEL_NAME)
@@ -128,34 +139,11 @@ def create_tool_call_agent(session_id: str):
     )
 
     # List of experts
-    experts = [
-        SearchExpert(),
-        FeedExpert(),
-        PriceExpert(),
-        ArticleExpert(),
-        NFTExpert(),
-        TransferExpert(),
-        SwapExpert(),
-    ]
+    experts = get_experts()
 
     # Construct the Tools agent
     agent = create_tool_calling_agent(interpreter, experts, prompt)
 
     # Create an agent executor by passing in the agent and tools
     agent_executor = AgentExecutor(agent=agent, tools=experts, verbose=True)
     return agent_executor
-
-
-# Function to get the chat history of a session from Postgres
-
-
-async def main():
-    # Create a tool call agent and use it to handle some inputs
-    agent = get_agent("123")
-    await agent.ainvoke({"input": "Swap 1 eth to usdt, from ethereum to arb"})
-    await agent.ainvoke({"input": "What is the price of ETH?"})
-    await agent.ainvoke({"input": "What did vitalik.eth do recently?"})
-
-
-if __name__ == "__main__":
-    asyncio.run(main())

src/openagent/conf/env.py

@@ -12,7 +12,8 @@ class Settings(BaseSettings):
     LLM_API_BASE: Optional[str] = Field(default=None)
     GOOGLE_CLOUD_PROJECT_ID: Optional[str] = Field(default=None)
     GOOGLE_GEMINI_API_KEY: Optional[str] = Field(default=None)
-    NFTSCAN_API_KEY: str = Field(..., env="NFTSCAN_API_KEY")
+    SERPAPI_API_KEY: Optional[str] = Field(default=None, env="SERPAPI_API_KEY")
+    NFTSCAN_API_KEY: Optional[str] = Field(default=None, env="NFTSCAN_API_KEY")
     DB_CONNECTION: str = Field(..., env="DB_CONNECTION")
     RSS3_DATA_API: str = Field(default="https://gi.rss3.io", env="RSS3_DATA_API")
     RSS3_SEARCH_API: str = Field(default="https://devnet.rss3.io/search", env="RSS3_SEARCH_API")

src/openagent/experts/dune_expert.py

@@ -0,0 +1,43 @@
+from typing import Optional, Type
+
+import requests
+from langchain.callbacks.manager import (
+    AsyncCallbackManagerForToolRun,
+    CallbackManagerForToolRun,
+)
+from langchain.tools import BaseTool
+from pydantic import BaseModel, Field
+
+from openagent.conf.env import settings
+
+
+class DuneSchema(BaseModel):
+    query: str = Field(description="The search query keywords for Dune dashboard search.")
+
+
+async def dune_search(query: str) -> str:
+    url = f"{settings.RSS3_SEARCH_API}/dune/search?keyword={query}"
+    headers = {"Accept": "*/*", "Content-Type": "application/x-www-form-urlencoded"}
+    response = requests.request("GET", url, headers=headers)
+    return response.text
+
+
+class DuneExpert(BaseTool):
+    name = "DuneExecutor"
+    description = """This tool searches for charts, data visualizations, and dashboards on Dune Analytics. \
+Use this tool to find visual representations of blockchain data, trend analysis, and market overviews."""
+    args_schema: Type[DuneSchema] = DuneSchema
+
+    def _run(
+        self,
+        query: str,
+        run_manager: Optional[CallbackManagerForToolRun] = None,
+    ) -> str:
+        raise NotImplementedError
+
+    async def _arun(
+        self,
+        query: str,
+        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
+    ) -> str:
+        return await dune_search(query)

src/openagent/experts/google_expert.py

@@ -0,0 +1,42 @@
+from typing import Optional, Type
+
+from langchain import SerpAPIWrapper
+from langchain.callbacks.manager import (
+    AsyncCallbackManagerForToolRun,
+    CallbackManagerForToolRun,
+)
+from langchain.tools import BaseTool
+from pydantic import BaseModel, Field
+
+
+class GoogleSchema(BaseModel):
+    query: str = Field(description="The search query keywords for Google search.")
+
+
+async def google_search(query: str) -> str:
+    search_wrapper = SerpAPIWrapper(
+        search_engine="google",
+        params={"engine": "google"},
+    )
+    return search_wrapper.run(query)
+
+
+class GoogleExpert(BaseTool):
+    name = "GoogleExecutor"
+    description = """A versatile search tool that retrieves up-to-date information from the web. \
+Use for current events, project details, fact-checking, and general research across various topics."""
+    args_schema: Type[GoogleSchema] = GoogleSchema
+
+    def _run(
+        self,
+        query: str,
+        run_manager: Optional[CallbackManagerForToolRun] = None,
+    ) -> str:
+        raise NotImplementedError
+
+    async def _arun(
+        self,
+        query: str,
+        run_manager: Optional[AsyncCallbackManagerForToolRun] = None,
+    ) -> str:
+        return await google_search(query)