Merge pull request #108 from Datura-ai/hotfix-main-bittensor

Hotfix main bittensor

Author: surcyf123

README.md

@@ -243,6 +243,14 @@ pm2 start start_validator.py --interpreter python3 -- --wallet_name "default" --
 ```
 ---
 
+# Cursor App Setup
+add env varialbe CURSOR_API_KEY to .env file 
+go to inside of cursor directory and run this command.
+```bash
+pm2 start start_cursor.sh --name cursor_app
+```
+for settings in cursor. Pls check this [Cursor README](./cursor/readme.md)
+
 ## License
 This repository is licensed under the MIT License.
 ```text

cortext/protocol.py

@@ -314,7 +314,7 @@ class StreamPrompting(bt.StreamingSynapse):
         default={},
         title="miner_info",
     )
-    time_taken: int = pydantic.Field(
+    time_taken: float = pydantic.Field(
         default=0,
         title="time_taken",
     )

cursor/Dockerfile

@@ -0,0 +1,29 @@
+llm_models = [
+    {
+        "id": "gpt-4o",
+        "name": "Meta: gpt-4o Instruct",
+        "created": 8192,
+        "description": "OpenAI gpt-4o model.",
+        "context_length": 8192,
+        "architecture": {"modality": "text->text", "tokenizer": "gpt-4o", "instruct_type": "gpt-4o"},
+        "pricing": {"prompt": "0.000000001", "completion": "0.000000001", "image": "0", "request": "0"},
+    },
+    {
+        "id": "chat-claude-3-5-sonnet-20240620",
+        "name": "Meta: claude-3-5-sonnet-20240620 Instruct",
+        "created": 8192,
+        "description": "claude-3-5-sonnet-20240620.",
+        "context_length": 8192,
+        "architecture": {"modality": "text->text", "tokenizer": "claude-3-5-sonnet-20240620", "instruct_type": "claude-3-5-sonnet-20240620"},
+        "pricing": {"prompt": "0.000000001", "completion": "0.000000001", "image": "0", "request": "0"},
+    },
+    {
+        "id": "chat-llama-3.1-70b-versatile",
+        "name": "Meta: llama-3.1-70b-versatile Instruct",
+        "created": 8192,
+        "description": "llama-3.1-70b-versatile.",
+        "context_length": 8192,
+        "architecture": {"modality": "text->text", "tokenizer": "llama-3.1-70b-versatile", "instruct_type": "llama-3.1-70b-versatile"},
+        "pricing": {"prompt": "0.00000001", "completion": "0.000000001", "image": "0", "request": "0"},
+    }
+]

cursor/app/constants.py

@@ -0,0 +1,28 @@
+import dataclasses
+from dataclasses import dataclass
+from dotenv import load_dotenv
+import os
+
+# Load environment variables from .env file
+load_dotenv()
+
+
+@dataclass
+class Config:
+    wallet_name: str
+    wallet_hotkey: str
+    api_key: str
+
+    @staticmethod
+    def from_env() -> "Config":
+        """Load configuration from environment variables."""
+        return Config(
+            wallet_name=os.getenv("WALLET_NAME", "default"),  # Default to an empty string if not set
+            wallet_hotkey=os.getenv("HOT_KEY", "default"),
+            api_key=os.getenv("CURSOR_API_KEY", "")
+        )
+
+
+# Load config
+config = Config.from_env()
+

cursor/app/core/config.py

@@ -0,0 +1,102 @@
+import asyncio
+from typing import Union, AsyncGenerator, Any
+
+import aiohttp
+import bittensor as bt
+from aiohttp import ServerTimeoutError, ClientConnectorError, ClientConnectionError
+from bittensor import dendrite
+import traceback
+import time
+from typing import Optional, List
+
+from .protocol import StreamPrompting
+
+
+class CortexDendrite(dendrite):
+    task_id = 0
+    miner_to_session = {}
+
+    def __init__(
+            self, wallet: Optional[Union[bt.wallet, bt.Keypair]] = None
+    ):
+        super().__init__(wallet)
+
+    async def call_stream(
+            self,
+            target_axon: Union[bt.AxonInfo, bt.axon],
+            synapse: bt.StreamingSynapse = bt.Synapse(),  # type: ignore
+            timeout: float = 12.0,
+            deserialize: bool = True,
+            organic: bool = True
+    ) -> AsyncGenerator[Any, Any]:
+        start_time = time.time()
+        target_axon = (
+            target_axon.info()
+            if isinstance(target_axon, bt.axon)
+            else target_axon
+        )
+
+        # Build request endpoint from the synapse class
+        request_name = synapse.__class__.__name__
+        endpoint = (
+            f"0.0.0.0:{str(target_axon.port)}"
+            if target_axon.ip == str(self.external_ip)
+            else f"{target_axon.ip}:{str(target_axon.port)}"
+        )
+        url = f"http://{endpoint}/{request_name}"
+
+        # Preprocess synapse for making a request
+        synapse: StreamPrompting = self.preprocess_synapse_for_request(target_axon, synapse, timeout)  # type: ignore
+        max_try = 0
+        timeout = aiohttp.ClientTimeout(total=100, connect=timeout, sock_connect=timeout, sock_read=timeout)
+        connector = aiohttp.TCPConnector(limit=200)
+        session = aiohttp.ClientSession(timeout=timeout, connector=connector)
+        try:
+            while max_try < 2:
+                async with session.post(
+                        url,
+                        headers=synapse.to_headers(),
+                        json=synapse.dict(),
+                ) as response:
+                    # Use synapse subclass' process_streaming_response method to yield the response chunks
+                    try:
+                        async for chunk in synapse.process_streaming_response(response, organic):  # type: ignore
+                            yield chunk  # Yield each chunk as it's processed
+                    except aiohttp.client_exceptions.ClientPayloadError:
+                        pass
+                    except ConnectionRefusedError as err:
+                        bt.logging.error(f"can not connect to miner for now. connection failed")
+                        max_try += 1
+                        continue
+                    except ClientConnectorError as err:
+                        bt.logging.error(f"can not connect to miner for now. retrying")
+                        max_try += 1
+                        continue
+                    except ClientConnectionError as err:
+                        bt.logging.error(f"can not connect to miner for now. retrying")
+                        max_try += 1
+                        continue
+                    except ServerTimeoutError as err:
+                        bt.logging.error(f"timeout error happens. max_try is {max_try}")
+                        max_try += 1
+                        continue
+                    except Exception as err:
+                        bt.logging.error(f"{err} issue from miner {synapse.uid} {synapse.provider} {synapse.model}")
+                    finally:
+                        pass
+                break
+
+        except Exception as e:
+            bt.logging.error(f"{e} {traceback.format_exc()}")
+        finally:
+            synapse.dendrite.process_time = str(time.time() - start_time)
+            await session.close()
+
+    async def call_stream_in_batch(
+            self,
+            target_axons: List[Union[bt.AxonInfo, bt.axon]],
+            synapses: List[bt.StreamingSynapse] = bt.Synapse(),  # type: ignore
+            timeout: float = 12.0,
+            deserialize: bool = True,
+    ) -> AsyncGenerator[Any, Any]:
+        pass

cursor/app/core/dendrite.py

@@ -1,33 +1,29 @@
-import time
-from fastapi import HTTPException
+from fastapi import FastAPI, Depends, HTTPException, Request
+from cursor.app.core.config import config
+from starlette.middleware.base import BaseHTTPMiddleware
+from starlette.responses import JSONResponse
+# Your predefined valid API keys
+VALID_API_KEYS = {config.api_key}
 
 
-async def verify_api_key_rate_limit(config, api_key):
-    # NOTE: abit dangerous but very useful
-    if not config.prod:
-        if api_key == "test":
-            return True
+class APIKeyMiddleware(BaseHTTPMiddleware):
+    async def dispatch(self, request: Request, call_next):
+        # Get the API key from the `Authorization` header
+        if request.method == "OPTIONS":
+            return await call_next(request)
 
-    rate_limit_key = f"rate_limit:{api_key}"
-    rate_limit = await config.redis_db.get(rate_limit_key)
-    if rate_limit is None:
-        async with await config.psql_db.connection() as connection:
-            # rate_limit = await get_api_key_rate_limit(connection, api_key)
-            if rate_limit is None:
-                raise HTTPException(status_code=403, detail="Invalid API key")
-        await config.redis_db.set(rate_limit_key, rate_limit, ex=30)
-    else:
-        rate_limit = int(rate_limit)
+        if not request.headers.get("Authorization"):
+            return JSONResponse(
+                {"detail": "Invalid or missing API Key"}, status_code=401
+            )
 
-    minute = time.time() // 60
-    current_rate_limit_key = f"current_rate_limit:{api_key}:{minute}"
-    current_rate_limit = await config.redis_db.get(current_rate_limit_key)
-    if current_rate_limit is None:
-        current_rate_limit = 0
-        await config.redis_db.expire(current_rate_limit_key, 60)
-    else:
-        current_rate_limit = int(current_rate_limit)
+        api_key = request.headers.get("Authorization").split(" ")[1]
 
-    await config.redis_db.incr(current_rate_limit_key)
-    if current_rate_limit >= rate_limit:
-        raise HTTPException(status_code=429, detail="Too many requests")
+        # Validate the API key
+        if not api_key or api_key not in VALID_API_KEYS:
+            return JSONResponse(
+                {"detail": "Invalid or missing API Key"}, status_code=401
+            )
+
+        # Proceed to the next middleware or route handler
+        return await call_next(request)