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 venvsource venv/bin/activateInstall Dependencies
pip install llama-cpp-python pymilvus "pymilvus[model]"Install Alith
python3 -m pip install alith -USet Environment Variables
For OpenAI/ChatGPT API:
export PRIVATE_KEY=<your wallet private key>
export OPENAI_API_KEY=<your openai api 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
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 theadd_query_nodefunction in Alith. This can only be done by LazAI admins.
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"
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,
"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.
Last updated