Using Python

Best Practice: Use a Python Virtual Environment

To avoid dependency conflicts and keep your environment clean, create and activate a Python virtual environment before installing any packages:

python3 -m venv venv
source venv/bin/activate

Install Dependencies

pip install llama-cpp-python pymilvus "pymilvus[model]"

For local development use the below installation command

pip install llama-cpp-python pymilvus "pymilvus[milvus_lite]"

Install Alith

python3 -m pip install alith -U

Set Environment Variables

Note: The public address of the private key you expose to the Query server is the LAZAI_IDAO_ADDRESS. Once the query server is running, the URL must be registered using the add_query_node function in Alith. This can only be done by LazAI admins.

For local development ask LazAI admins to add your wallet address & "http://localhost:3000" url to register in Alith funciton.

For OpenAI/ChatGPT API:

export PRIVATE_KEY=<your wallet private key>
export OPENAI_API_KEY=<your openai api key>
export RSA_PRIVATE_KEY_BASE64=<your rsa private key>

For other OpenAI-compatible APIs (DeepSeek, Gemini, etc.):

export PRIVATE_KEY=<your wallet private key>
export LLM_API_KEY=<your api key>
export LLM_BASE_URL=<your api base url>

Step 1: Run the Query Server

Local Development

For OpenAI API or OpenAI-compatible APIs (DeepSeek, Gemini, etc.):

import logging
import sys
import json
import uvicorn
import argparse

from fastapi import FastAPI, Response, status
from fastapi.middleware.cors import CORSMiddleware

from alith.lazai import Client
from alith.lazai.node.middleware import HeaderValidationMiddleware
from alith.lazai.node.validator import decrypt_file_url
from alith import MilvusStore, chunk_text
from alith.query.types import QueryRequest
from alith.query.settlement import QueryBillingMiddleware

import os
from dotenv import load_dotenv

load_dotenv()
# Get OpenAI API key from environment variable
PRIVATE_KEY = os.getenv("PRIVATE_KEY")
RSA_PRIVATE_KEY_BASE64 = os.getenv("RSA_PRIVATE_KEY_BASE64")
# LLM_API_KEY = os.getenv("LLM_API_KEY")
# LLM_BASE_URL = os.getenv("LLM_BASE_URL")
# DSTACK_API_KEY = os.getenv("DSTACK_API_KEY")


# Set the API key for OpenAI
os.environ["PRIVATE_KEY"] = PRIVATE_KEY
os.environ["RSA_PRIVATE_KEY_BASE64"] = RSA_PRIVATE_KEY_BASE64
# os.environ["LLM_API_KEY"] = LLM_API_KEY
# os.environ["LLM_BASE_URL"] = LLM_BASE_URL
# os.environ["DSTACK_API_KEY"] = DSTACK_API_KEY


# Logging configuration
logging.basicConfig(
    stream=sys.stdout,
    level=logging.INFO,
    format="%(asctime)s - %(levelname)s - %(message)s",
)
logger = logging.getLogger(__name__)
client = Client()
app = FastAPI(title="Alith LazAI Privacy Data Query Node", version="1.0.0")

store = MilvusStore()
collection_prefix = "query_"

@app.get("/health")
async def health_check():
    return {"status": "healthy", "message": "Server is running"}

@app.get("/")
async def root():
    return {"message": "Alith LazAI Privacy Data Query Node", "version": "1.0.0"}


@app.post("/query/rag")
async def query_rag(req: QueryRequest):
    try:
        file_id = req.file_id
        if req.file_url:
            file_id = client.get_file_id_by_url(req.file_url)
        if file_id:
            file = client.get_file(file_id)
        else:
            return Response(
                status_code=status.HTTP_400_BAD_REQUEST,
                content=json.dumps(
                    {
                        "error": {
                            "message": "File ID or URL is required",
                            "type": "invalid_request_error",
                        }
                    }
                ),
            )
        owner, file_url, file_hash = file[1], file[2], file[3]
        collection_name = collection_prefix + file_hash
        # Cache data in the vector database
        if not store.has_collection(collection_name):
            encryption_key = client.get_file_permission(
                file_id, client.contract_config.data_registry_address
            )
            data = decrypt_file_url(file_url, encryption_key).decode("utf-8")
            store.create_collection(collection_name=collection_name)
            store.save_docs(chunk_text(data), collection_name=collection_name)
        data = store.search_in(
            req.query, limit=req.limit, collection_name=collection_name
        )
        logger.info(f"Successfully processed request for file: {file}")
        return {
            "data": data,
            "owner": owner,
            "file_id": file_id,
            "file_url": file_url,
            "file_hash": file_hash,
        }
    except Exception as e:
        return Response(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content=json.dumps(
                {
                    "error": {
                        "message": f"Error processing request for req: {req}. Error: {str(e)}",
                        "type": "internal_error",
                    }
                }
            ),
        )


def run(host: str = "0.0.0.0", port: int = 8000, *, settlement: bool = False):

    # FastAPI app and LazAI client initialization

    app.add_middleware(
        CORSMiddleware,
        allow_origins=["*"],
        allow_credentials=True,
        allow_methods=["*"],
        allow_headers=["*"],
    )

    if settlement:
        app.add_middleware(HeaderValidationMiddleware)
        app.add_middleware(QueryBillingMiddleware)

    return uvicorn.run(app, host=host, port=port)


if __name__ == "__main__":
    description = "Alith data query server. Host your own embedding models and support language query!"
    parser = argparse.ArgumentParser(description=description)
    parser.add_argument(
        "--host",
        type=str,
        help="Server host",
        default="0.0.0.0",
    )
    parser.add_argument(
        "--port",
        type=int,
        help="Server port",
        default=8000,
    )
    parser.add_argument(
        "--model",
        type=str,
        help="Model name or path",
        default="/root/models/qwen2.5-1.5b-instruct-q5_k_m.gguf",
    )
    args = parser.parse_args()

    run(host=args.host, port=args.port, settlement=False)

Production Deployment on Phala TEE Cloud

For production-ready applications, deploy your data query server on Phala TEE Cloud  for enhanced security and privacy. Once deployed, you will receive an data query URL that needs to be registered using the add_query_node function by LazAI admins. Use this starter kit to create and push your Docker image https://github.com/0xLazAI/LazAI-DATA-Query-Server-Setup-Kit

You can also use the existing data query nodes.

Step 2: Request Query via LazAI Client

from alith.lazai import Client
import requests

client = Client()
node = "0x2591E4C0e6E771927A45eFAE8Cd5Bf20e585A57A" #change this address with one you registered with admin 

try:
    print("try to get user")
    user =  client.get_user(client.wallet.address)
    print(user)
except Exception as e:
    print("try to get user failed")
    print(e)
    print("try to add user failed")
    client.add_user(1000000)
    print("user added")


print("try to get query account")

url = client.get_query_node(node)[1]
print(url)
headers = client.get_request_headers(node)
print("request headers:", headers)
print(
    "request result:",
    requests.post(
        f"{url}/query/rag",
        headers=headers,
        json={
            "file_id": 10, #change with your file_id 
            "query": "summarise the best character?",
        },
    ).json(),
)

---

Security & Privacy

• Your data never leaves your control. Data query is performed in a privacy-preserving environment, using cryptographic settlement and secure computation.

• Settlement headers ensure only authorized users and nodes can access your data for data query.

• File ID links your data query request to the specific data you contributed, maintaining a verifiable chain of custody.