Python SDK

pip install -e packages/sdk-python     # внутри монорепозитория

from aacsearch import AACSearchClient, AACSearchError

client = AACSearchClient("aac_live_9f3c2a1b…", base_url="https://search.example.com/api/v1")
result = client.search(index="documentation", query={"q": "тарифы", "per_page": 5})
print(result["found"], result["hits"][0]["document"]["title"])

AACSearchClient(
    api_key: str,
    base_url: str = "http://localhost:3000/api/v1",
    max_retries: int = 2,
)

AACSearchClient("")
# AACSearchError: api_key is required (status=401, code=MISSING_API_KEY)

class AACSearchError(Exception):
    status: int          # HTTP-статус
    code: str            # UNAUTHORIZED, QUOTA_EXCEEDED, VALIDATION_ERROR…
    details: Any         # details из конверта ошибки
    request_id: str | None

try:
    client.upsert_documents(
        index="documentation",
        body={"documents": [{"external_id": "pricing", "content": "Pro стоит 29 USD в месяц."}]},
    )
except AACSearchError as error:
    if error.code == "RATE_LIMITED":
        time.sleep(2)
    elif error.code == "QUOTA_EXCEEDED":
        raise SystemExit(f"Квота исчерпана до {(error.details or {}).get('resetsAt')}")
    else:
        raise RuntimeError(f"{error.code} {error} (request {error.request_id})")

hits = client.search(
    index="documentation",
    query={"q": "тарифы", "query_by": "title,content", "filter_by": "tags:=[billing]", "per_page": 5},
)
for hit in hits["hits"]:
    print(hit["document"]["title"], hit["score"])

client.upsert_documents(
    index="documentation",
    body={
        "documents": [
            {
                "external_id": "pricing",
                "title": "Тарифы",
                "content": "Pro стоит 29 USD в месяц и включает 500 000 поисков.",
                "tags": ["billing"],
            }
        ],
        "action": "upsert",
    },
)

document = client.get_document(index="documentation", externalId="pricing")
client.delete_documents(index="documentation", query={"filter_by": "source_type:=connector"})

answer = client.answer(index="documentation", body={"question": "Сколько стоит pro?"})
print(answer["answer"])
for citation in answer["citations"]:
    print(f"[{citation['index']}] {citation['title']} {citation['url'] or ''}")

for event in client.assistant_stream(index="documentation", body={"question": "А лимиты?"}):
    if event["type"] == "delta":
        print(event["payload"].get("text", ""), end="")