second commit

.gitignore

@@ -109,4 +109,7 @@ dmypy.json
 .idea/
 
 # Mac OS
-.DS_Store
+.DS_Store
+
+TODO.md
+run.py

README.md

@@ -0,0 +1,55 @@
+# ChatXBT Web3 AI Assitant Service
+
+AI-powered unification and execution protocol for web3 applications
+
+## Requirements
+
+The list below contains all the requirements needed to have a complete developer environment to sart working on this project.
+Follow the links to install or update these tools on your machine.
+
+1. [Python 3.12+](https://www.python.org/downloads/release/python-3120/)
+2. [Pyenv](https://github.com/pyenv/pyenv)
+3. [Pip Installer](https://pip.pypa.io/en/stable/installation/)
+4. [Python Venv](https://docs.python.org/3/library/venv.html)
+
+## Installation
+
+1. Clone the repository: `git clone https://github.com/ChatXBT/chatxbt-web3-ai-assistant.git`
+2. Get a copy of the `\`.env`\` file from the project manager`
+3. Create a virtual environment: `python -m venv env`
+4. Activate the virtual environment: `source env/bin/activate`
+4. Install the dependencies: `pip install -r requirements.txt`
+5. Run the application: `chainlit run chatxbt-assistant.py -w`
+6. Open the application in your browser: `http://localhost:8000`
+
+## Development
+
+The project is structured as follows:
+
+1. src/data_sources: `a list of data sources that can be used to retrieve data for the AI assistant.`
+2. src/databases: `a list of databases that can be used to store data for the AI assistant.`
+3. src/libs: `a list of libraries that can be used to extend the classes and functions servicing the AI assistant.`
+4. src/llms: `a list of language models powered libs that can be used to power the AI assistant.`
+5. src/search_services: `a list of search services that can be used to retrieve data for the AI assistant.`
+6. src/tools: `a list of tools that can be used to extend the functionality of the AI assistant.`
+
+## Guides
+
+1. Follow the development, structure and documentation patterns of the `src/tools/coin_data_toolkit.py` to build custom toolkits
+2. Follow the development, structure and documentation patterns of the `src/tools/coin_data_toolkit.py` to build classes 
+3. Pay attention to the `constants` variables and `caching` decorators to optimize perfomance of class functions/methods across the project
+4. Create new classes in the appropriate folders to maintain an organized codebase and project
+5. After installing any new package run `` to update the dependency requirements list
+6. Expose RPC functions for SDKs that do not support python runtime. use [DeepKit](https://deepkit.io/documentation/rpc) and [BunJS](https://bun.sh/) if possible.
+
+## Deplopyment to producton
+
+To be finalised and added
+
+## TODO
+
+TODO: Add deployment instructions
+TODO: Setup CI/CD pipelines for automatic deployment
+TODO: Setup and integrate `LitLLM` instance for LLM service high availability
+TODO: Add RPC class for remmote method calls to other chatxbt services written in other launguages such as `JS`, `TS` & `GO`
+TODO: Write enough unit test to ensure class and function input and output correctness

ai_functions.py

@@ -1,60 +0,0 @@
-from pydantic import BaseModel, Field
-from typing import Dict, Optional, Type
-
-from data_sources.coin_gecko import CoinGecko
-from src.data_sources.cryptocompare import CryptoCompare
-
-from langchain.tools.base import BaseTool
-
-class CryptoCoinPrice(BaseModel):
-    """Represents the prices of a coin in various currencies."""
-    prices: Dict[str, float] = Field(..., description="Prices in various currencies")
-
-class CryptoCoinPriceData(BaseModel):
-    """Encapsulates both CoinGecko and CryptoCompare price data."""
-    coingecko_price: Dict[str, CryptoCoinPrice] = Field(..., description="CoinGecko prices for various coins")
-    crypto_compare_price: Dict[str, CryptoCoinPrice] = Field(..., description="CryptoCompare prices for various coins")
-
-class PriceInput(BaseModel):
-    coin_id: str = Field(..., description="The ID of the cryptocurrency coin to retrieve prices for")
-    vs_currency: str = Field("usd", description="The currency to compare against")
-
-class CryptoCoinPriceOutput(BaseModel):
-    price_data: CryptoCoinPriceData
-
-class CryptoCoinPriceTool(BaseTool):
-    name = "CryptoCoinPriceTool"
-    description = "Fetches price data for a given cryptocurrency coin from CoinGecko and CryptoCompare"
-    args_schema: Type[BaseModel] = PriceInput
-    return_direct: bool = True
-    
-    def __init__(self, id: Optional[str] = None):
-        self.id = id
-        self.coingecko = CoinGecko()
-        self.crypto_compare = CryptoCompare()
-
-    def _run(self, coin_id: str, vs_currency: str = "usd") -> CryptoCoinPriceData:
-        coingecko_price_data = self.coingecko.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
-        crypto_compare_price_data = self.crypto_compare.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
-
-        coingecko_price = {}
-        crypto_compare_price = {}
-
-        if coin_id in coingecko_price_data:
-            coingecko_price[coin_id] = CryptoCoinPrice(prices=coingecko_price_data[coin_id])
-        else:
-            print(f"Warning: CoinGecko data for {coin_id} not found.")
-
-        if coin_id.upper() in crypto_compare_price_data:
-            crypto_compare_price[coin_id] = CryptoCoinPrice(prices=crypto_compare_price_data[coin_id.upper()])
-        else:
-            print(f"Warning: CryptoCompare data for {coin_id} not found.")
-
-        return CryptoCoinPriceData(
-            coingecko_price=coingecko_price,
-            crypto_compare_price=crypto_compare_price
-        )
-
-    def __call__(self, inputs: PriceInput) -> CryptoCoinPriceOutput:
-        price_data = self._run(inputs.coin_id, inputs.vs_currency)
-        return CryptoCoinPriceOutput(price_data=price_data)

chainlit.md

@@ -0,0 +1,14 @@
+# Welcome to Chainlit! 🚀🤖
+
+Hi there, Developer! 👋 We're excited to have you on board. Chainlit is a powerful tool designed to help you prototype, debug and share applications built on top of LLMs.
+
+## Useful Links 🔗
+
+- **Documentation:** Get started with our comprehensive [Chainlit Documentation](https://docs.chainlit.io) 📚
+- **Discord Community:** Join our friendly [Chainlit Discord](https://discord.gg/k73SQ3FyUh) to ask questions, share your projects, and connect with other developers! 💬
+
+We can't wait to see what you create with Chainlit! Happy coding! 💻😊
+
+## Welcome screen
+
+To modify the welcome screen, edit the `chainlit.md` file at the root of your project. If you do not want a welcome screen, just leave this file empty.

chat-app.py

@@ -1,55 +0,0 @@
-from dotenv import load_dotenv
-
-from langchain.chains import LLMMathChain
-from langchain.llms.openai import OpenAI
-from langchain.chat_models import ChatOpenAI
-from langchain.utilities.serpapi import SerpAPIWrapper
-from langchain.agents import initialize_agent, Tool, AgentExecutor
-import chainlit as cl
-
-from src.tools.crypto_coin_price_tool import CryptoCoinPriceTool
-
-load_dotenv()
-
-@cl.on_chat_start
-def start():
-    llm = ChatOpenAI(temperature=0, streaming=True)
-    llm1 = OpenAI(temperature=0, streaming=True)
-    search = SerpAPIWrapper()
-    get_crypto_coin_price = CryptoCoinPriceTool()
-    llm_math_chain = LLMMathChain.from_llm(llm=llm, verbose=True)
-
-    tools = [
-        Tool(
-            name="Search",
-            func=search.run,
-            description="useful for when you need to answer questions about current events. You should ask targeted questions",
-            handle_tool_error=True,
-        ),
-        Tool(
-            name="Calculator",
-            func=llm_math_chain.run,
-            description="useful for when you need to answer questions about math",
-            handle_tool_error=True,
-        ),
-        Tool(
-            name=get_crypto_coin_price.name,
-            func=get_crypto_coin_price.run,
-            description=get_crypto_coin_price.description,
-            handle_tool_error=True,
-        ),
-    ]
-    agent = initialize_agent(
-        tools, llm1, agent="chat-zero-shot-react-description", verbose=True, handle_parsing_errors=True
-    )
-    cl.user_session.set("agent", agent)
-
-
-@cl.on_message
-async def main(message: cl.Message):
-    agent = cl.user_session.get("agent")  # type: AgentExecutor
-    cb = cl.LangchainCallbackHandler(stream_final_answer=True)
-
-    await cl.make_async(agent.run)(message.content, callbacks=[cb])
-
-    

chatxbt-assistant.py

@@ -0,0 +1,56 @@
+import os
+from dotenv import load_dotenv
+load_dotenv()
+
+import chainlit as cl
+from phi.assistant import Assistant
+from phi.tools.duckduckgo import DuckDuckGo
+from phi.llm.openai import OpenAIChat
+from phi.tools.yfinance import YFinanceTools
+from src.tools.coin_data_toolkit import CryptoDataTools
+
+
+# assistant = Assistant(tools=[DuckDuckGo()], show_tool_calls=True)
+# assistant.print_response("Whats happening in France?", markdown=True)
+
+# f_assistant = Assistant(
+#     llm=OpenAIChat(model="gpt-4o"),
+#     tools=[YFinanceTools(stock_price=True, analyst_recommendations=True, company_info=True, company_news=True)],
+#     show_tool_calls=True,
+#     markdown=True,
+# )
+# f_assistant.print_response("What is the stock price of NVDA")
+# f_assistant.print_response("Write a comparison between NVDA and AMD, use all tools available.")
+
+@cl.on_chat_start
+def start():
+    is_dev_mode = True if os.getenv("DEV_MODE") else False
+
+    # Initialize the assistant
+    cxbt_assistant = Assistant(
+        llm=OpenAIChat(model="gpt-4o"),
+        tools=[CryptoDataTools(), DuckDuckGo(), YFinanceTools(stock_price=True)],
+        show_tool_calls= is_dev_mode,
+        markdown=True,
+    )
+
+    # Set the assistant in the user session
+    cl.user_session.set("agent", cxbt_assistant)
+
+@cl.on_message
+async def main(message: cl.Message):
+    # Retrieve the assistant from the user session
+    agent = cl.user_session.get("agent")
+
+    # Process the user message using the assistant
+    # response = agent.run(message.content, stream=True)
+    response = ""
+    for delta in agent.run(message.content, stream=True):
+        response += delta
+    
+    # Send the response back to the user
+    await cl.Message(content=response).send()
+
+# Run the Chainlit application
+if __name__ == "__main__":
+    cl.run()

sample.env

@@ -0,0 +1,28 @@
+OPENAI_API_KEY=""
+
+GROQ_API_KEY=""
+
+LANGCHAIN_TRACING_V2=""
+LANGCHAIN_ENDPOINT=""
+LANGCHAIN_API_KEY=""
+
+JINA_READER_API_KEY=""
+
+EXA_API_KEY=""
+
+SERPAPI_API_KEY=""
+
+CRYPTOCOMPARE_API_KEY=""
+
+COINGECKO_PRO_API_KEY=""
+COINGECKO_DEMO_API_KEY=""
+
+UPSTASH_REDIS_REST_URL=""
+UPSTASH_REDIS_REST_TOKEN=""
+UPSTASH_REDIS_HOST=""
+UPSTASH_REDIS_PORT=""
+UPSTASH_REDIS_PASSWORD=""
+UPSTASH_REDIS_CONNECTION_STRING=""
+
+LOGFIRE_TOKEN=""
+LOG_LEVEL=""

src/data_sources/coin_gecko.py

@@ -1,7 +1,7 @@
 import os
 from dotenv import load_dotenv
 from src.databases.redis import REDIS_CACHED
-from src.libs.constants import ONE_HOUR_IN_SECONDS, ONE_MONTH_IN_SECONDS
+from src.libs.constants import ONE_HOUR_IN_SECONDS, ONE_MINUTE_IN_SECONDS, ONE_MONTH_IN_SECONDS
 from src.libs.logger import logger
 from coingecko import CoinGeckoProClient, CoinGeckoDemoClient
 
@@ -16,9 +16,9 @@ class CoinGecko:
         else:
             self.cgc = CoinGeckoDemoClient(api_key=os.getenv('COINGECKO_DEMO_API_KEY'))
     
-    @redis_cache(ttl=ONE_HOUR_IN_SECONDS)
+    @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
     @logger.instrument()
-    def get_coin_price(self, ids: list, vs_currencies: list, cache_ttl: int = None) -> dict:
+    def get_coin_price(self, ids: list[str], vs_currencies: list[str], cache_ttl: int = None) -> dict:
         # logger.debug(f"ids: {ids}")
         # logger.debug(f"vs_currencies: {vs_currencies}")
 

src/data_sources/cryptocompare.py

@@ -38,7 +38,7 @@ class CryptoCompare:
         
     @redis_cache(ttl=ONE_MINUTE_IN_SECONDS)
     @logger.instrument()
-    def get_coin_price(self, ids: list, vs_currencies: list, cache_ttl: int = None) -> dict:
+    def get_coin_price(self, ids: list[str], vs_currencies: list[str], cache_ttl: int = None) -> dict:
         
         url = f"{self.CRYPTO_COMPARE_BASE_URL}data/pricemulti"
         # logger.debug(url)
@@ -59,5 +59,6 @@ class CryptoCompare:
                 response.raise_for_status()  # Raise an exception if the request was unsuccessful
                 return response.json()
         except httpx.HTTPError as e:
-            logger.debug(f"An error occurred while making the request: {e}")
+            # logger.debug(f"An error occurred while making the request: {e}")
+            print(f"An error occurred while making the request: {e}")
             return None

src/databases/redis.py

@@ -1,5 +1,5 @@
-import json
 import os
+import json
 from dotenv import load_dotenv
 from upstash_redis import Redis
 from src.libs.logger import logger

src/libs/logger.py

@@ -1,17 +1,17 @@
 import os
 from dotenv import load_dotenv
 
-import logfire
+import logfire as lf
 
 load_dotenv()
 
-logfire.configure(
+lf.configure(
     token=os.getenv('LOGFIRE_TOKEN'),
-    pydantic_plugin=logfire.PydanticPlugin(record='all'),
-    console=logfire.ConsoleOptions(min_log_level= os.getenv('LOG_LEVEL'))
+    pydantic_plugin=lf.PydanticPlugin(record='all'),
+    console=lf.ConsoleOptions(min_log_level= os.getenv('LOG_LEVEL'))
 )
-logfire.instrument_redis()
-logfire.instrument_httpx()
-logfire.instrument_requests()
+lf.instrument_redis()
+lf.instrument_httpx()
+lf.instrument_requests()
 
-logger = logfire
+logger = lf

src/libs/redis.py

@@ -1,14 +0,0 @@
-keys = {
-    "cryptocompare": {
-        "top_exchanges": "top_exchanges",
-        "all_exchanges": "all_exchanges",
-        "top_assets": "top_assets",
-        "all_assets": "all_assets",
-    },
-    "coingecko": {
-        "top_exchanges": "top_exchanges",
-        "all_exchanges": "all_exchanges",
-        "top_assets": "top_assets",
-        "all_assets": "all_assets",
-    }
-}

src/requirements.txt

@@ -1,8 +0,0 @@
-auto_mix_prep==0.2.0
-httpx==0.27.0
-logfire==0.42.0
-pyjsonq==1.0.2
-python-dotenv==1.0.1
-scrapegraphai==1.6.0
-ulid==1.1
-upstash_redis==1.1.0

src/tools/coin_data_toolkit.py

@@ -0,0 +1,60 @@
+import json
+
+from phi.tools import Toolkit
+from phi.utils.log import logger
+
+from src.data_sources.coin_gecko import CoinGecko
+from src.data_sources.cryptocompare import CryptoCompare
+
+class CryptoDataTools(Toolkit):
+    def __init__(self):
+        super().__init__(name="crypto_data_tools")
+        self.register(self.get_coin_price)
+        self.coingecko = CoinGecko()
+        self.crypto_compare = CryptoCompare()
+
+    def get_coin_price(self, coin_id: str, vs_currency: str = "usd") -> str:
+        """
+        Fetches the price data for a given cryptocurrency coin from CoinGecko and CryptoCompare.
+
+        Args:
+        coin_id (str): The unique identifier for the cryptocurrency coin.
+        vs_currency (str, optional): The currency to which the coin price will be compared. Defaults to "usd".
+
+        Returns:
+        str: A JSON string containing the price data from CoinGecko and CryptoCompare for the specified coin.
+
+        Raises:
+        Exception: If an error occurs while fetching the price data.
+
+        Example:
+        >>> get_coin_price("bitcoin", "eur")
+        """
+        logger.info(f"Fetching price data for {coin_id} cryptocurrency coin from CoinGecko and CryptoCompare")
+        try:
+            coingecko_price = {}
+            crypto_compare_price = {}
+
+            coingecko_price_data = self.coingecko.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
+            crypto_compare_price_data = self.crypto_compare.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
+
+            logger.debug(f"coingecko_price_data: {coingecko_price_data}")
+            logger.debug(f"crypto_compare_price_data: {crypto_compare_price_data}")
+
+            if coin_id in coingecko_price_data:
+                coingecko_price[coin_id] = coingecko_price_data[coin_id]
+            else:
+                logger.warning(f"Warning: CoinGecko data for {coin_id} not found.")
+
+            if coin_id.upper() in crypto_compare_price_data:
+                crypto_compare_price[coin_id] = crypto_compare_price_data[coin_id.upper()]
+            else:
+                logger.warning(f"Warning: CryptoCompare data for {coin_id} not found.")
+
+            logger.debug(f"respons {coingecko_price}")
+            logger.debug(f"respons {crypto_compare_price}")
+            
+            return f"{(coingecko_price, crypto_compare_price)}"
+        except Exception as e:
+            logger.warning(f"Failed to fetch price data for {coin_id}: {e}")
+            return f"Error: {e}"

src/tools/crypto_coin_price_tool.py

@@ -1,60 +0,0 @@
-from pydantic import BaseModel, Field
-from typing import Dict, Optional, Type
-
-from src.data_sources.coin_gecko import CoinGecko
-from src.data_sources.cryptocompare import CryptoCompare
-
-from langchain.tools.base import BaseTool
-
-class CryptoCoinPrice(BaseModel):
-    """Represents the prices of a coin in various currencies."""
-    prices: Dict[str, float] = Field(..., description="Prices in various currencies")
-
-class CryptoCoinPriceData(BaseModel):
-    """Encapsulates both CoinGecko and CryptoCompare price data."""
-    coingecko_price: Dict[str, CryptoCoinPrice] = Field(..., description="CoinGecko prices for various coins")
-    crypto_compare_price: Dict[str, CryptoCoinPrice] = Field(..., description="CryptoCompare prices for various coins")
-
-class PriceInput(BaseModel):
-    coin_id: str = Field(..., description="The ID of the cryptocurrency coin to retrieve prices for")
-    vs_currency: str = Field("usd", description="The currency to compare against")
-
-class CryptoCoinPriceOutput(BaseModel):
-    price_data: CryptoCoinPriceData
-
-class CryptoCoinPriceTool(BaseTool):
-    name = "CryptoCoinPriceTool"
-    description = "Fetches price data for a given cryptocurrency coin from CoinGecko and CryptoCompare"
-    args_schema: Type[BaseModel] = PriceInput
-    return_direct: bool = True
-    
-    def __init__(self, id: Optional[str] = None):
-        self.id = id
-        self.coingecko = CoinGecko()
-        self.crypto_compare = CryptoCompare()
-
-    def _run(self, coin_id: str, vs_currency: str = "usd") -> CryptoCoinPriceData:
-        coingecko_price_data = self.coingecko.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
-        crypto_compare_price_data = self.crypto_compare.get_coin_price(ids=[coin_id], vs_currencies=[vs_currency])
-
-        coingecko_price = {}
-        crypto_compare_price = {}
-
-        if coin_id in coingecko_price_data:
-            coingecko_price[coin_id] = CryptoCoinPrice(prices=coingecko_price_data[coin_id])
-        else:
-            print(f"Warning: CoinGecko data for {coin_id} not found.")
-
-        if coin_id.upper() in crypto_compare_price_data:
-            crypto_compare_price[coin_id] = CryptoCoinPrice(prices=crypto_compare_price_data[coin_id.upper()])
-        else:
-            print(f"Warning: CryptoCompare data for {coin_id} not found.")
-
-        return CryptoCoinPriceData(
-            coingecko_price=coingecko_price,
-            crypto_compare_price=crypto_compare_price
-        )
-
-    def __call__(self, inputs: PriceInput) -> CryptoCoinPriceOutput:
-        price_data = self._run(inputs.coin_id, inputs.vs_currency)
-        return CryptoCoinPriceOutput(price_data=price_data)

src/xbt-core.py

@@ -1,43 +0,0 @@
-import os
-import httpx
-import urllib.parse
-from typing import Self
-from pyjsonq import JsonQ
-from dotenv import load_dotenv
-
-load_dotenv()
-
-JINA_SEARCH_BASE_ENDPOINT = "r.jina.ai"
-JINA_READER_BASE_ENDPOINT = "s.jina.ai"
-
-class XBTCore:
-    def __init__(self) -> None:
-        self.value = 0
-        self.JINA_SEARCH_BASE_ENDPOINT = "s.jina.ai"
-        self.JINA_READER_BASE_ENDPOINT = "r.jina.ai"
-
-    def search_web_with_jina(self, search_query: str=False) -> Self:
-        url = self.JINA_SEARCH_BASE_ENDPOINT
-        encoded_search_query = urllib.parse.quote(search_query)
-
-        try:
-            with httpx.Client() as client:
-                response = client.get(f"{url}/{encoded_search_query}")
-                response.raise_for_status()
-                return response.json()
-        except httpx.HTTPError as e:
-            print(f"An error occurred: {e}")
-            return None
-        
-    def read_website_with_jina(self, website_url: str=False) -> Self:
-        url = self.JINA_READER_BASE_ENDPOINT
-
-        try:
-            with httpx.Client() as client:
-                response = client.get(f"{url}/{website_url}")
-                response.raise_for_status()
-                return response.json()
-        except httpx.HTTPError as e:
-            print(f"An error occurred: {e}")
-            return None
-