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

Hotfix main bittensor

Author: surcyf123

.gitignore

@@ -16,5 +16,4 @@ validators/.ipynb_checkpoints/
 **/Cortex.t.egg-info
 **/test.ipynb
 .env
-**/server/**/*.py
 cache.db

cortext/dendrite.py

@@ -26,6 +26,7 @@ async def call_stream(
             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 = (
@@ -61,7 +62,7 @@ async def call_stream(
                 ) as response:
                     # Use synapse subclass' process_streaming_response method to yield the response chunks
                     try:
-                        async for chunk in synapse.process_streaming_response(response):  # type: ignore
+                        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

cortext/protocol.py

@@ -366,10 +366,11 @@ def to_headers(self) -> dict:
 
         return headers
 
-    async def process_streaming_response(self, response: StreamingResponse) -> AsyncIterator[str]:
+    async def process_streaming_response(self, response: StreamingResponse, organic=True) -> AsyncIterator[str]:
         if self.completion is None:
             self.completion = ""
-        async for chunk in response.content.iter_any():
+        chunk_size = 100 if organic else 1024
+        async for chunk in response.content.iter_chunked(chunk_size):
             tokens = chunk.decode("utf-8")
             self.completion += tokens
             yield tokens

server/Dockerfile

@@ -0,0 +1,18 @@
+# Use official Python image
+FROM python:3.10
+
+# Set working directory
+WORKDIR /app
+
+# Copy and install dependencies
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+# Copy the app files to the container
+COPY . .
+
+# Expose the FastAPI port
+EXPOSE 8000
+
+# Start FastAPI app
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

server/app/__init__.py

@@ -0,0 +1,38 @@
+import os
+from typing import List
+from . import models, schemas
+from .database import cur, TABEL_NAME, conn
+from fastapi import HTTPException
+
+
+def create_item(item: schemas.ItemCreate):
+    query = f"INSERT INTO {TABEL_NAME} (p_key, question, answer, provider, model, timestamp) VALUES (%s, %s, %s, %s, %s, %s)"
+    cur.execute(query, item.p_key, item.question, item.answer, item.provider, item.model, item.timestamp)
+    conn.commit()  # Save changes to the database
+    return item
+
+
+def create_items(items: List[schemas.ItemCreate]):
+    query = f"INSERT INTO {TABEL_NAME} (p_key, question, answer, provider, model, timestamp) VALUES (%s, %s, %s, %s, %s, %s)"
+    datas = []
+    for item in items:
+        datas.append((item.p_key, item.question, item.answer, item.provider, item.model, item.timestamp))
+    try:
+        cur.executemany(query, datas)
+        conn.commit()  # Save changes to the database
+    except Exception as err:
+        raise HTTPException(status_code=500, detail=f"Internal Server Error {err}")
+
+
+def get_items(skip: int = 0, limit: int = 10):
+    query = f"SELECT * FROM {TABEL_NAME} LIMIT {limit} OFFSET {skip};"
+    cur.execute(query)
+    items = cur.fetchall()  # Fetch all results
+    return [dict(item) for item in items]
+
+
+def get_item(p_key: int):
+    query = f"SELECT * FROM {TABEL_NAME} WHERE p_key = %s"
+    cur.execute(query, (p_key,))
+    item = cur.fetchone()  # Fetch one result
+    return dict(item)

server/app/curd.py

@@ -0,0 +1,54 @@
+import psycopg2
+import os
+from contextlib import asynccontextmanager
+
+DATABASE_URL = os.getenv("DATABASE_URL")
+TABEL_NAME = 'query_resp_data'
+# PostgreSQL connection parameters
+conn = psycopg2.connect(DATABASE_URL)
+
+# Create a cursor object to interact with the database
+cur = conn.cursor()
+
+
+async def create_table(app):
+    global conn, cur, TABEL_NAME
+    try:
+        # Connect to the PostgreSQL database
+        conn = psycopg2.connect(DATABASE_URL)
+        cur = conn.cursor()
+
+        # SQL command to create a table
+        create_table_query = f"""
+        CREATE TABLE IF NOT EXISTS {TABEL_NAME} (
+            p_key VARCHAR(100) PRIMARY KEY,
+            question JSON,
+            answer TEXT,
+            provider VARCHAR(100),
+            model VARCHAR(100),
+            timestamp FLOAT
+        );
+        """
+
+        # Execute the SQL command
+        cur.execute(create_table_query)
+        conn.commit()  # Save changes
+        create_index_query = f"""
+        CREATE INDEX IF NOT EXISTS question_answer_index ON {TABEL_NAME} (provider, model);
+        """
+        cur.execute(create_index_query)
+        conn.commit()
+        print("Table created successfully!")
+
+    except Exception as e:
+        print(f"Error creating table: {e}")
+
+    finally:
+        # Close the cursor and connection
+        if cur:
+            cur.close()
+        if conn:
+            conn.close()
+
+
+create_table(None)

server/app/database.py

@@ -0,0 +1,37 @@
+from contextlib import asynccontextmanager
+from fastapi import FastAPI, Depends, HTTPException
+from . import curd, models, schemas
+from .database import create_table
+from typing import List
+
+
+@asynccontextmanager
+async def lifespan(app: FastAPI):
+    # Load the ML model
+    await create_table(None)
+    yield
+
+
+app = FastAPI(lifespan=lifespan)
+
+
+# Create an item
+@app.post("/items")
+def create_item(items: List[schemas.ItemCreate]):
+    return curd.create_items(items=items)
+
+
+# Read all items
+@app.get("/items", response_model=list)
+def read_items(skip: int = 0, limit: int = 10):
+    items = curd.get_items(skip=skip, limit=limit)
+    return items
+
+
+# Read a single item by ID
+@app.get("/items/{p_key}", response_model=schemas.Item)
+def read_item(p_key: int):
+    db_item = curd.get_item(p_key=p_key)
+    if db_item is None:
+        raise HTTPException(status_code=404, detail="Item not found")
+    return db_item

server/app/main.py

@@ -0,0 +1,18 @@
+from pydantic import BaseModel, Json
+
+
+class ItemBase(BaseModel):
+    p_key: str
+    question: Json
+    answer: str
+    provider: str
+    model: str
+    timestamp: float
+
+
+class ItemCreate(ItemBase):
+    pass
+
+
+class Item(ItemBase):
+    pass

server/app/models.py

@@ -0,0 +1,30 @@
+version: "3.9"
+
+services:
+  db:
+    image: postgres:13
+    restart: always
+    environment:
+      POSTGRES_USER: ${POSTGRES_USER}
+      POSTGRES_PASSWORD: ${POSTGRES_PASSWORD}
+      POSTGRES_DB: ${POSTGRES_DB}
+    volumes:
+      - postgres_data_score:/var/lib/postgresql/data
+    ports:
+      - "5432:5432"
+
+  web:
+    build: .
+    restart: always
+    ports:
+      - "8000:8000"
+    depends_on:
+      - db
+    environment:
+      DATABASE_URL: postgres://${POSTGRES_USER}:${POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB}
+      POSTGRES_DB: ${POSTGRES_DB}
+    volumes:
+      - .:/app
+
+volumes:
+  postgres_data_score:

server/app/schemas.py

@@ -0,0 +1,5 @@
+fastapi
+uvicorn[standard]
+psycopg2-binary
+sqlalchemy
+pydantic

server/app/utils.py

@@ -1,3 +1,4 @@
+import random
 import sqlite3
 import time
 import hashlib
@@ -54,7 +55,7 @@ def set_cache_in_batch(self, syns: List[StreamPrompting], ttl=3600 * 24, block_n
         datas = []
         last_update_time = time.time()
         for syn in syns:
-            p_key = self.generate_hash(str(time.monotonic_ns()) + str(syn.json()))
+            p_key = self.generate_hash(str(time.monotonic_ns()) + str(syn.json()) + str(random.random()))
             syn.time_taken = syn.dendrite.process_time or 0
             syn.validator_info = {"vali_uid": self.vali_uid, "vali_hotkey": self.vali_hotkey}
             syn.miner_info = {"miner_id": syn.uid, "miner_hotkey": syn.axon.hotkey}

server/docker-compose.yml

@@ -174,7 +174,7 @@ def setup_max_capacity(item):
         if isinstance(value, dict):  # If the value is another dictionary, recurse
             setup_max_capacity(value)
         elif isinstance(value, (int, float)):  # If the value is a number, increment by 5
-            item[key] = min(value, 30)
+            item[key] = min(value, 10)
 
 
 def get_bandwidth(data, uid, provider, model):

server/requirements.txt

@@ -1,20 +1,20 @@
 import asyncio
 import concurrent
 import random
-import traceback
 import threading
 
 import torch
 import time
+import requests
 
 from black.trans import defaultdict
 from substrateinterface import SubstrateInterface
 from functools import partial
-from typing import Tuple, List
+from typing import Tuple
 import bittensor as bt
 from bittensor import StreamingSynapse
 import cortext
-
+import json
 from starlette.types import Send
 
 from cortext.protocol import IsAlive, StreamPrompting, ImageResponse, Embeddings
@@ -93,6 +93,7 @@ def __init__(self, config, cache: QueryResponseCache, loop=None):
         self.loop.create_task(self.process_queries_from_database())
 
         self.saving_datas = []
+        self.url = None
         daemon_thread = threading.Thread(target=self.saving_resp_answers_from_miners)
         daemon_thread.start()
 
@@ -104,11 +105,28 @@ def saving_resp_answers_from_miners(self):
                 time.sleep(1)
             else:
                 bt.logging.info(f"saving responses...")
+                start_time = time.time()
                 self.cache.set_cache_in_batch([item.get('synapse') for item in self.saving_datas],
                                               block_num=self.current_block,
                                               cycle_num=self.current_block // 36, epoch_num=self.current_block // 360)
                 bt.logging.info(f"total saved responses is {len(self.saving_datas)}")
                 self.saving_datas.clear()
+                if not self.url:
+                    return
+                bt.logging.info("sending datas to central server.")
+                json_data = [item.get('synapse').dict() for item in self.saving_datas]
+                headers = {
+                    'Content-Type': 'application/json'  # Specify that we're sending JSON
+                }
+                response = requests.post(self.url, data=json.dumps(json_data), headers=headers)
+                # Check the response
+                if response.status_code == 200:
+                    bt.logging.info(
+                        f"Successfully sent data to central server. {time.time() - start_time} sec total elapsed for sending to central server.")
+                else:
+                    bt.logging.info(
+                        f"Failed to send data. Status code: {response.status_code} {time.time() - start_time} sec total elapsed for sending to central server.")
+                    bt.logging.info(f"Response:{response.text}")
 
     async def run_sync_in_async(self, fn):
         return await self.loop.run_in_executor(None, fn)
@@ -162,7 +180,7 @@ async def update_and_refresh(self):
         await self.initialize_uids_and_capacities()
         bt.logging.info("Metagraph refreshed.")
 
-    async def query_miner(self, uid, query_syn: cortext.ALL_SYNAPSE_TYPE):
+    async def query_miner(self, uid, query_syn: cortext.ALL_SYNAPSE_TYPE, organic=True):
         query_syn.uid = uid
         if query_syn.streaming:
             if uid is None:
@@ -195,6 +213,7 @@ async def handle_response(resp):
                 target_axon=axon,
                 synapse=query_syn,
                 timeout=query_syn.timeout,
+                organic=organic
             )
             await handle_response(response)
         else:
@@ -256,7 +275,7 @@ async def perform_synthetic_queries(self):
                     uid = self.task_mgr.assign_task(query_syn)
                     if uid is None:
                         bt.logging.debug(f"No available uids for synthetic query process.")
-                    synthetic_tasks.append((uid, self.query_miner(uid, query_syn)))
+                    synthetic_tasks.append((uid, self.query_miner(uid, query_syn, organic=False)))
 
             bt.logging.debug(f"{time.time() - start_time} elapsed for creating and submitting synthetic queries.")
 
@@ -283,7 +302,7 @@ async def perform_synthetic_queries(self):
                 f"synthetic queries and answers has been processed in cache successfully. total times {time.time() - start_time}")
 
     def pop_synthetic_tasks_max_100_per_miner(self, synthetic_tasks):
-        batch_size = 10000
+        batch_size = 3000
         max_query_cnt_per_miner = 50
         batch_tasks = []
         remain_tasks = []