import os
import textwrap
from typing import List
from dotenv import load_dotenv
from mistralai.client import Mistral
from moorcheh_sdk import MoorchehClient
load_dotenv()
MOORCHEH_API_KEY = os.getenv("MOORCHEH_API_KEY", "").strip()
MISTRAL_API_KEY = os.getenv("MISTRAL_API_KEY", "").strip()
if not MOORCHEH_API_KEY or not MISTRAL_API_KEY:
raise SystemExit("Set MOORCHEH_API_KEY and MISTRAL_API_KEY.")
MODEL = "mistral-embed"
VECTOR_DIMENSION = 1024
NAMESPACE = "mistral-embed-demo"
CHUNK_SIZE = 900
CHUNK_OVERLAP = 180
def mistral_embed(client: Mistral, texts: List[str]) -> List[List[float]]:
resp = client.embeddings.create(model=MODEL, inputs=texts)
data = list(resp.data)
if data and hasattr(data[0], "index"):
data = sorted(data, key=lambda d: d.index)
return [list(d.embedding) for d in data]
def to_float_vector(values: List[float]) -> List[float]:
return [float(x) for x in values]
def chunk_text(text: str, chunk_size: int = CHUNK_SIZE, overlap: int = CHUNK_OVERLAP) -> List[str]:
chunks: List[str] = []
start = 0
while start < len(text):
end = min(start + chunk_size, len(text))
chunks.append(text[start:end].strip())
if end == len(text):
break
start = max(end - overlap, 0)
return [c for c in chunks if c]
def extract_text(result: dict) -> str:
if result.get("text"):
return str(result["text"])
metadata = result.get("metadata") or {}
if isinstance(metadata, dict):
return str(metadata.get("text") or metadata.get("raw_text") or metadata.get("content") or "")
return ""
def clean_text(text: str) -> str:
return " ".join(str(text).split())
def print_result(idx: int, result: dict) -> None:
metadata = result.get("metadata") or {}
text_value = clean_text(extract_text(result))
wrapped = textwrap.fill(text_value, width=100)
print(f"[{idx}] id={result.get('id')}")
print(f"score={result.get('score')} label={result.get('label')}")
print(f"section={metadata.get('section')} source_doc_id={metadata.get('source_doc_id')}")
print("text:")
print(wrapped if wrapped else "(no text returned)")
print("-" * 120)
mistral = Mistral(api_key=MISTRAL_API_KEY)
mc = MoorchehClient(api_key=MOORCHEH_API_KEY)
try:
mc.namespaces.create(
namespace_name=NAMESPACE,
type="vector",
vector_dimension=VECTOR_DIMENSION,
)
except Exception:
pass
source_documents = [
{
"id": "guide-vector-namespaces",
"section": "vector-namespace-best-practices",
"text": (
"Moorcheh vector namespaces support bring-your-own-embedding workflows. "
"Use mistral-embed with vector_dimension 1024 and store chunk text on each vector for display in search results."
),
},
{
"id": "guide-search-tuning",
"section": "semantic-search-tuning",
"text": (
"Tune similarity_search top_k and threshold for your use case. Mistral embeddings are normalized; cosine-style similarity applies."
),
},
]
documents = []
for doc in source_documents:
parts = chunk_text(doc["text"])
for idx, chunk in enumerate(parts):
documents.append(
{
"id": f"{doc['id']}-chunk-{idx}",
"text": chunk,
"source_doc_id": doc["id"],
"section": doc["section"],
"chunk_index": idx,
"total_chunks": len(parts),
}
)
texts = [d["text"] for d in documents]
doc_embeddings = mistral_embed(mistral, texts)
mc.vectors.upload(
namespace_name=NAMESPACE,
vectors=[
{
"id": documents[i]["id"],
"vector": to_float_vector(doc_embeddings[i]),
"text": documents[i]["text"],
"source": "mistral-embed",
"model": MODEL,
"section": documents[i]["section"],
"source_doc_id": documents[i]["source_doc_id"],
"chunk_index": documents[i]["chunk_index"],
"total_chunks": documents[i]["total_chunks"],
}
for i in range(len(documents))
],
)
query = "What vector dimension does mistral-embed use with Moorcheh?"
query_embeddings = mistral_embed(mistral, [query])
query_vec = to_float_vector(query_embeddings[0])
results = mc.similarity_search.query(
namespaces=[NAMESPACE],
query=query_vec,
top_k=5,
kiosk_mode=True,
threshold=0.15,
)
print(f"namespace={NAMESPACE} total_results={len(results.get('results', []))}")
print("=" * 120)
for idx, r in enumerate(results.get("results", []), start=1):
print_result(idx, r)
