Agentic_Rag/scripts/wait_for_milvus.py

from __future__ import annotations

import time

from pymilvus import MilvusClient

from app.config import get_settings


def main() -> None:
    settings = get_settings()
    kwargs = {"uri": settings.milvus_uri}
    token = settings.milvus_token.get_secret_value()
    if token:
        kwargs["token"] = token

    deadline = time.monotonic() + 120
    last_error: Exception | None = None
    while time.monotonic() < deadline:
        try:
            client = MilvusClient(**kwargs)
            client.list_collections()
            print(f"Milvus 已就绪：{settings.milvus_uri}")
            return
        except Exception as exc:
            last_error = exc
            time.sleep(2)
    raise SystemExit(f"Milvus 在 120 秒内未就绪：{last_error}")


if __name__ == "__main__":
    main()