"""Aliyun Bailian / DashScope LLM client. Wraps the DashScope native ``services/aigc/text-generation/generation`` endpoint with two thin call shapes: * :meth:`BailianLLMClient.chat_text` — return plain text (used by the rare "summarise this paragraph" callsite). * :meth:`BailianLLMClient.chat_json` — return parsed JSON, with the prompt explicitly instructed to emit a single JSON object. The AI-spending dashboard's monthly summary generator is the primary caller. We pass ``response_format={"type": "json_object"}`` which qwen-max honours since 2024-09; older models silently ignore it, so we still parse defensively. Why DashScope native (not OpenAI-compatible mode): * The OpenAI-compatible endpoint at ``dashscope.aliyuncs.com/compatible-mode/v1/chat/completions`` is newer and not all qwen-* models are wired through it (qwen-max-latest is, qwen-max-2024-09-19 isn't, etc.). Native is the most-stable surface and the one Aliyun's own docs recommend for production. * Native sends the bearer key in ``Authorization: Bearer sk-xxx`` — same envelope shape as OpenAI, so the SDK swap is one line if we ever need to change. * No new dependency: just ``httpx``. Errors ------ The client raises :class:`AppException` with code ``BAILIAN_LLM_ERROR`` for anything non-2xx OR a 200 response carrying ``code != Success`` (DashScope sometimes 200s a soft-failure). The caller — :class:`AiSpendingService.generate_summary` — logs + suppresses these so the rest of the dashboard renders even when the LLM is unavailable. One error is raised BEFORE any request goes out: ``BAILIAN_LLM_INPUT_TOO_LARGE`` (413), when the prompt cannot fit the model's input window (``_MAX_INPUT_TOKENS``). Retrying it is pointless — the caller has to send less. Cost ---- Default model ``qwen-max-latest`` charges ~0.012 RMB per 1k input tokens + 0.04 RMB per 1k output tokens (Aliyun pricelist 2026-04). A ~5 KB input + 2 KB output costs <0.02 RMB; a per-user-per-day cap of 5 regenerations × 30 admins × 30 days ≈ ¥27/month worst case. """ from __future__ import annotations import asyncio import json import logging import time from collections.abc import AsyncIterator, Coroutine, Iterable from typing import Any import httpx from app.core.exceptions import AppException logger = logging.getLogger(__name__) _DASHSCOPE_NATIVE_URL = ( "https://dashscope.aliyuncs.com/api/v1/services/aigc/" "text-generation/generation" ) # Singapore endpoint mirror — DashScope mirrors qwen-* on the # overseas region. The admin selects between ``cn`` (mainland) and # ``intl`` (Singapore) on the AI / LLM connector card; an # Aliyun account provisioned in Singapore can ONLY hit this URL — # its sk- key is rejected at the mainland endpoint. _DASHSCOPE_INTL_URL = ( "https://dashscope-intl.aliyuncs.com/api/v1/services/aigc/" "text-generation/generation" ) # Compatible-mode (OpenAI chat/completions shape) — used for the # tool-calling agent loop. The native text-generation endpoint above is # kept for the single-shot chat_text/chat_json callers; compatible-mode # is the well-documented surface for function calling on qwen-max/plus # (same sk- bearer key, just a different path). _COMPAT_CN_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/chat/completions" _COMPAT_INTL_URL = ( "https://dashscope-intl.aliyuncs.com/compatible-mode/v1/chat/completions" ) # Compatible-mode embeddings (OpenAI /embeddings shape) — semantic retrieval # (KB search). Same sk- bearer + region split; default model text-embedding-v3 # (1024-dim, multilingual). Stored as a portable JSON float array + Python # cosine so it runs on local MySQL tests (no TiDB VECTOR dependency). _EMBED_CN_URL = "https://dashscope.aliyuncs.com/compatible-mode/v1/embeddings" _EMBED_INTL_URL = ( "https://dashscope-intl.aliyuncs.com/compatible-mode/v1/embeddings" ) DEFAULT_EMBED_MODEL = "text-embedding-v3" #: DashScope rejects an embeddings request carrying more than this many inputs #: (HTTP 400, ``batch size is invalid``). Measured against the live endpoint: #: 10 succeeds, 11 does not. ``embed`` chunks to it rather than leaving each #: caller to remember — the one caller that forgot took the assistant's entire #: semantic search offline without a single error reaching a log. _EMBED_MAX_BATCH = 10 #: How many batches may be in flight at once. Measured against the live endpoint #: (2026-08-19): one batch of 10 short texts takes **2.5–2.9s**, and the #: assistant's 60-flow corpus (6 batches) completed in 5.3s at this width. The #: 273-tool corpus is 28 batches, so 8 wide is ~4 rounds ≈ 12s rather than the #: ~80s it would take serially — small enough not to read as a burst, wide enough #: that a cold process does not stall the first employee who asks a question. _EMBED_MAX_CONCURRENCY = 8 #: Ceiling on a whole multi-batch call, however many batches it needs. The #: per-request ``timeout`` cannot express this: serially, 28 batches each got the #: full budget (~15 minutes with a timing-out backend) inside ONE tool dispatch #: of a user-facing turn, where the kernel's turn budget is only checked between #: iterations and cannot interrupt it. Roughly 3x the measured worst case above, #: so a merely-slow backend still succeeds and only a broken one degrades to #: keyword matching — which is a far better answer than a turn that never returns. _EMBED_TOTAL_DEADLINE_SECONDS = 45.0 # ``region`` arg → endpoint URL. REGION_CN = "cn" REGION_INTL = "intl" DEFAULT_REGION = REGION_INTL # Default to Singapore: Helix is deployed # overseas (AWS Singapore) and the org's Bailian tenancy lives there. # Mainland users explicitly opt in by setting region=cn on the Connector. _REGION_ENDPOINTS = { REGION_CN: _DASHSCOPE_NATIVE_URL, REGION_INTL: _DASHSCOPE_INTL_URL, } _COMPAT_REGION_ENDPOINTS = { REGION_CN: _COMPAT_CN_URL, REGION_INTL: _COMPAT_INTL_URL, } _EMBED_REGION_ENDPOINTS = { REGION_CN: _EMBED_CN_URL, REGION_INTL: _EMBED_INTL_URL, } def compat_endpoint_for_region(region: str | None) -> str: """Resolve a region → compatible-mode (OpenAI-shape) endpoint URL.""" code = (region or "").strip().lower() or DEFAULT_REGION return _COMPAT_REGION_ENDPOINTS.get(code, _COMPAT_REGION_ENDPOINTS[DEFAULT_REGION]) def embed_endpoint_for_region(region: str | None) -> str: """Resolve a region → compatible-mode embeddings endpoint URL.""" code = (region or "").strip().lower() or DEFAULT_REGION return _EMBED_REGION_ENDPOINTS.get(code, _EMBED_REGION_ENDPOINTS[DEFAULT_REGION]) def endpoint_for_region(region: str | None) -> str: """Resolve a region string → endpoint URL. Unknown values fall back to the default (Singapore) and are NOT silently ignored — the admin sees a warning at instantiation if they typed a wrong code via the API. """ code = (region or "").strip().lower() or DEFAULT_REGION url = _REGION_ENDPOINTS.get(code) if url is None: logger.warning( "BailianLLMClient: unknown region %r — defaulting to %s", region, DEFAULT_REGION, ) return _REGION_ENDPOINTS[DEFAULT_REGION] return url DEFAULT_MODEL = "qwen-max-latest" DEFAULT_TIMEOUT_SECONDS = 90.0 #: Largest prompt DashScope will accept, in tokens. Measured, not assumed: #: on 2026-09-01 the weekly cross-domain brief sent a 1,049,934-character #: prompt and qwen3-max answered #: ``HTTP 400 {"code":"InvalidParameter","message":"<400> #: InternalError.Algo.InvalidParameter: Range of input length should be #: [1, 258048]"}``. Same shape as ``_EMBED_MAX_BATCH``: a vendor ceiling #: this client knows about, checked here so no caller has to remember it. _MAX_INPUT_TOKENS = 258_048 #: Characters per token assumed when estimating a prompt's token count. #: Deliberately LOW, so the estimate is an over-count and the guard trips #: before the vendor does rather than after. The one ratio we have #: measured is ~3.9, for the ASCII-heavy JSON fact packages (the #: 2026-08-29 brief: 447,124 characters → 114,689 input tokens, as #: recorded in ``intelligence_analyses``). This is deliberately not tuned #: to that case — other callsites send CJK prose, which tokenises far #: denser, and a ratio fitted to JSON would under-count their prompts. At #: 3.0 the guard fires around 774 K characters, i.e. before the ~1.0 M #: characters that actually earned the 400 above. _ESTIMATED_CHARS_PER_TOKEN = 3.0 def _guard_input_size(*parts: str) -> None: """Refuse a prompt that cannot fit the model's input window. Without this the only symptom is the vendor's own 400, which names a token range and nothing about which of our payloads produced it — the 2026-09-01 brief outage took a prod-TiDB query plus a local rebuild of the fact package to attribute. Raising here names the measured size and the ceiling, and it costs nothing: the request is going to be rejected. """ chars = sum(len(part) for part in parts) estimated = int(chars / _ESTIMATED_CHARS_PER_TOKEN) if estimated <= _MAX_INPUT_TOKENS: return raise AppException( code="BAILIAN_LLM_INPUT_TOO_LARGE", message=( f"Bailian LLM prompt is too large: {chars} characters " f"(~{estimated} tokens at {_ESTIMATED_CHARS_PER_TOKEN} chars/token) " f"exceeds the {_MAX_INPUT_TOKENS}-token input window. " "The caller must send fewer facts, not a larger budget." ), status_code=413, ) def _message_text(message: dict[str, Any]) -> str: """The billable text of one chat message, for the input-size guard. A vision turn's ``content`` is a **list of parts** rather than a string, so ``str(content)`` would measure the repr of the whole list — including a base64 data URL, if one is inlined, which over-counts by ~4x and could trip the guard on a request the vendor would have accepted. Only the ``text`` parts are counted here; image parts bill as image tokens, which the guard does not model and the vendor reports separately. """ content = message.get("content") if isinstance(content, str): return content if isinstance(content, list): pieces: list[str] = [] for part in content: if isinstance(part, dict): text = part.get("text") if isinstance(text, str): pieces.append(text) elif isinstance(part, str): pieces.append(part) return "".join(pieces) return "" def _parse_sse_data(line: str) -> dict[str, Any] | None: """One ``data:`` frame of a compatible-mode SSE stream, or ``None``. ``None`` for keep-alives, blank separators, the ``[DONE]`` sentinel, and anything that is not parseable JSON. A malformed frame is skipped rather than raised on: the vendor occasionally emits a comment line, and killing a half-delivered answer over one unreadable frame is strictly worse for the reader than dropping it. """ if not line: return None stripped = line.strip() if not stripped or stripped.startswith(":"): return None if not stripped.startswith("data:"): return None payload = stripped[5:].strip() if not payload or payload == "[DONE]": return None try: parsed = json.loads(payload) except ValueError: logger.debug("Bailian stream: skipping unparseable frame %r", payload[:120]) return None return parsed if isinstance(parsed, dict) else None def _as_int(value: Any) -> int | None: """``int(value)`` when it is a sane non-negative count, else ``None``.""" try: number = int(value) except (TypeError, ValueError): return None return number if number >= 0 else None async def _gather_cancelling( coros: Iterable[Coroutine[Any, Any, list[list[float]]]], ) -> list[list[list[float]]]: """Run them all, in order, cancelling the rest the moment one fails. The first failure is re-raised AS ITSELF. ``TaskGroup`` reports through a ``BaseExceptionGroup``, and letting that out would quietly change this method's documented contract — "raises ``AppException`` on network / non-200 / empty" — for every caller that branches on the exception type. """ try: async with asyncio.TaskGroup() as group: tasks = [group.create_task(coro) for coro in coros] except BaseExceptionGroup as grouped: raise _first_leaf(grouped) from None return [task.result() for task in tasks] def _first_leaf(grouped: BaseExceptionGroup) -> BaseException: """The first non-group exception inside a (possibly nested) group.""" for member in grouped.exceptions: if isinstance(member, BaseExceptionGroup): return _first_leaf(member) return member return grouped class BailianLLMClient: """Stateless Bailian LLM caller; instantiate per-request. Construct with the sk-xxx api_key (NOT a RAM AccessKey — DashScope's chat endpoint is sk-bearer-only). The model id and endpoint can be overridden per-instance for the rare case where admins want to A/B test across qwen-max / qwen-plus. """ def __init__( self, api_key: str, *, model: str | None = None, region: str | None = None, endpoint_url: str | None = None, timeout: float = DEFAULT_TIMEOUT_SECONDS, purpose: str | None = None, ) -> None: api_key = (api_key or "").strip() if not api_key: raise AppException( code="BAILIAN_LLM_NO_KEY", message=( "No Bailian (DashScope) API key configured. " "Open Admin → Connectors → AI / LLM and paste a " "sk-xxx key issued from the Bailian console " "(业务空间 → API-KEY 管理)." ), status_code=503, ) self._api_key = api_key self._model = (model or DEFAULT_MODEL).strip() or DEFAULT_MODEL # Explicit endpoint_url wins (test harnesses, custom proxies); # otherwise pick by region. Caller must pass ONE of the two — # ``region`` is the supported public surface. if endpoint_url: self._endpoint = endpoint_url self._region = region or DEFAULT_REGION else: self._region = (region or DEFAULT_REGION).strip().lower() self._endpoint = endpoint_for_region(self._region) self._timeout = timeout # Which feature this client works for, for the call ledger. A # client built without one falls back to the ambient # ``purpose_scope`` at call time (not here — the scope is # usually entered after construction), and finally to # "unattributed". See :mod:`app.domains.bailian.ledger`. self._purpose = purpose async def _record_call( self, *, operation: str, ok: bool, started: float, input_tokens: object = None, output_tokens: object = None, error_code: str | None = None, ) -> None: """Book this call in the LLM ledger. Never raises. Reads the actor and trace from the ambient request context when there is one; cron work legitimately has neither. """ from app.auth.auth_context import ( # noqa: PLC0415 current_actor_user_id, current_request_trace_id, ) from app.domains.bailian import ledger # noqa: PLC0415 await ledger.record( purpose=self._purpose or ledger.current_purpose(), model=self._model, region=self._region, operation=operation, ok=ok, input_tokens=input_tokens, output_tokens=output_tokens, duration_ms=int((time.monotonic() - started) * 1000), error_code=error_code, trace_id=current_request_trace_id.get(), actor_person_id=current_actor_user_id.get(), ) @property def model(self) -> str: """The model this client will actually call. Resolved, not as-configured: an unset ``model`` setting falls back to ``DEFAULT_MODEL`` in ``__init__``, so a caller recording "which model produced this" from the setting would write NULL for every default-model call. Mirrors ``region`` / ``endpoint``, which exist for the same reason. """ return self._model @property def region(self) -> str: return self._region @property def endpoint(self) -> str: return self._endpoint # ── public api ───────────────────────────────────────────── async def chat_json( self, *, system: str, user: str, max_tokens: int = 3000, temperature: float = 0.3, ) -> dict[str, Any]: """Run one chat completion and parse JSON from the reply. The system message is appended with a "respond ONLY with valid JSON" instruction so older qwen models that don't support ``response_format`` still cooperate. We strip ```json fences if the model adds them. """ forced_system = ( system + "\n\nIMPORTANT: Reply with a single valid JSON object " + "and nothing else. Do NOT wrap it in markdown fences. " + "Do NOT include any explanatory prose before or after." ) result = await self._call( system=forced_system, user=user, max_tokens=max_tokens, temperature=temperature, response_format={"type": "json_object"}, operation="chat_json", ) text = (result.get("output_text") or "").strip() if not text: raise AppException( code="BAILIAN_LLM_EMPTY", message="Bailian LLM returned an empty body.", status_code=502, ) # Tolerant JSON extraction: if the model decided to ignore the # instruction and wrapped in ```json … ``` we still want to # parse out the inner object instead of failing the whole # generate-summary pipeline. cleaned = _strip_code_fences(text) try: payload = json.loads(cleaned) except json.JSONDecodeError as exc: # Separate the two ways JSON parsing fails, because they need # opposite responses. ``finish_reason == "length"`` means the # reply was cut off at ``max_tokens`` mid-object: re-running # with the same cap produces the same truncation, so the # actionable fix is a bigger budget, not a retry. Reporting # that as "the model may have added prose. Re-run generation." # sent the 2026-08-17 daily-brief investigation chasing the # model's wording while the real ceiling sat in code. if result.get("finish_reason") == "length": out = result.get("output_tokens") logger.warning( "Bailian LLM reply truncated at max_tokens=%d " "(output_tokens=%s) — JSON is incomplete", max_tokens, out, ) raise AppException( code="BAILIAN_LLM_OUTPUT_TRUNCATED", message=( f"Bailian LLM reply was cut off at the " f"{max_tokens}-token output budget " f"(output_tokens={out}), so the JSON is " "incomplete. Raise the caller's max_tokens or " "ask for a shorter output." ), status_code=502, ) from exc logger.warning( "Bailian LLM did not emit valid JSON: %s\nbody=%s", exc, text[:600], ) raise AppException( code="BAILIAN_LLM_BAD_JSON", message=( "Bailian LLM did not return valid JSON; the model " "may have added prose. Re-run generation." ), status_code=502, ) from exc if not isinstance(payload, dict): raise AppException( code="BAILIAN_LLM_BAD_JSON", message=( "Bailian LLM JSON top-level is not an object " f"(got {type(payload).__name__})." ), status_code=502, ) # Surface usage so callers can record token counts. payload.setdefault("_meta", {}) payload["_meta"]["input_tokens"] = result.get("input_tokens") payload["_meta"]["output_tokens"] = result.get("output_tokens") payload["_meta"]["model_used"] = result.get("model_used") return payload async def chat_text( self, *, system: str, user: str, max_tokens: int = 1500, temperature: float = 0.7, ) -> str: result = await self._call( system=system, user=user, max_tokens=max_tokens, temperature=temperature, response_format=None, operation="chat_text", ) return (result.get("output_text") or "").strip() async def chat( self, *, messages: list[dict[str, Any]], tools: list[dict[str, Any]] | None = None, max_tokens: int = 1500, temperature: float = 0.3, model: str | None = None, translation_options: dict[str, Any] | None = None, ) -> dict[str, Any]: """Instrumented wrapper — see :meth:`_call` for why the failure path is booked as well as the success path.""" started = time.monotonic() try: result = await self._chat_raw( messages=messages, tools=tools, max_tokens=max_tokens, temperature=temperature, model=model, translation_options=translation_options, ) except AppException as exc: await self._record_call( operation="chat", ok=False, started=started, error_code=getattr(exc, "code", None), ) raise except Exception as exc: await self._record_call( operation="chat", ok=False, started=started, error_code=type(exc).__name__[:64], ) raise await self._record_call( operation="chat", ok=True, started=started, input_tokens=result.get("input_tokens"), output_tokens=result.get("output_tokens"), ) return result async def _chat_raw( self, *, messages: list[dict[str, Any]], tools: list[dict[str, Any]] | None = None, max_tokens: int = 1500, temperature: float = 0.3, model: str | None = None, translation_options: dict[str, Any] | None = None, ) -> dict[str, Any]: """One turn against the compatible-mode (OpenAI chat/completions) endpoint, with optional function-calling ``tools``. ``messages`` is an OpenAI-shape list (system/user/assistant, plus ``{"role": "tool", "tool_call_id", "content"}`` for tool results). Returns ``{"content": str, "tool_calls": [...], "finish_reason", "input_tokens", "output_tokens"}`` — the caller drives the agent loop (execute tool_calls → append a tool message → call again). ``model`` overrides the instance's model for this call only. ``translation_options`` is the ``qwen-mt-*`` family's own parameter (``{"source_lang", "target_lang"}``). It is a first-class field rather than a generic passthrough because it is the difference between translating with a translation model and instructing a chat model to translate — AI Labs needs the former, so that Helix is not authoring an instruction on the employee's behalf. """ endpoint = compat_endpoint_for_region(self._region) # The agent loop appends a tool result per turn, so this shape grows # across a conversation rather than being sized once at the callsite. # ``tools`` is measured too: the schemas travel with every turn and # bill as input, and the assistant's catalogue is large enough that # leaving them out would under-count the request that matters most. _guard_input_size( *(str(message.get("content") or "") for message in messages), json.dumps(tools, ensure_ascii=False) if tools else "", ) body: dict[str, Any] = { "model": (model or self._model), "messages": messages, "max_tokens": max_tokens, "temperature": temperature, } if tools: body["tools"] = tools if translation_options: body["translation_options"] = translation_options headers = { "Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json", } try: async with httpx.AsyncClient( timeout=self._timeout, trust_env=False, ) as client: resp = await client.post(endpoint, headers=headers, json=body) except httpx.HTTPError as exc: raise AppException( code="BAILIAN_LLM_NETWORK", message=f"Bailian LLM network error: {exc}", status_code=502, ) from exc if resp.status_code != 200: raise AppException( code="BAILIAN_LLM_HTTP_ERROR", # Body logged, not relayed: a 4xx from the input filter # quotes the employee's own text back, and this message is # rendered on screen and written to the application log by # ``chat._persist_failure``. Logs go to a company-wide # collector that the employee's hard delete cannot reach. message=( "The model provider rejected this request " f"(HTTP {resp.status_code})." ), status_code=502, ) data = resp.json() choices = data.get("choices") or [] if not choices: raise AppException( code="BAILIAN_LLM_EMPTY", message="Bailian LLM (compat) returned no choices.", status_code=502, ) choice = choices[0] message = choice.get("message") or {} usage = data.get("usage") or {} return { "content": (message.get("content") or "").strip(), "tool_calls": message.get("tool_calls") or [], "finish_reason": choice.get("finish_reason"), "input_tokens": usage.get("prompt_tokens"), "output_tokens": usage.get("completion_tokens"), } async def chat_stream( self, *, messages: list[dict[str, Any]], model: str | None = None, max_tokens: int = 4000, temperature: float = 0.7, enable_thinking: bool | None = None, ) -> AsyncIterator[dict[str, Any]]: """Stream one chat turn, yielding deltas as they arrive. Yields dicts, each exactly one of: * ``{"type": "reasoning", "text": str}`` — a chunk of the model's thinking, when the model emits ``reasoning_content``. * ``{"type": "text", "text": str}`` — a chunk of the answer. * ``{"type": "done", "finish_reason": str|None, "input_tokens": int|None, "output_tokens": int|None}`` — exactly once, last. Why this exists next to :meth:`chat`: the AI Labs chat surface shows output as it is produced, and a 4,000-token answer from a thinking model takes tens of seconds to complete. Buffering it would make every turn feel broken. :meth:`chat` stays the shape for the agent loop, which cannot act on a partial tool call anyway. ``model`` overrides the instance's model for this call only — the AI Labs catalogue lets an employee pick a model per conversation, and building a fresh client per turn would re-resolve config that has not changed. ``enable_thinking`` maps to the vendor's own flag. Measured 2026-09-03 on ``qwen3.8-max``: ``False`` → first text at 2.33s; ``True`` → first reasoning chunk at 2.07s, first text at 3.38s. Left as ``None`` the vendor default applies, which for the ``qwen3.7-*`` family means deep thinking and a 15-18s wait before anything appears — hence the surface exposes it as a toggle. Errors raise :class:`AppException` before the first yield when the connection or HTTP status fails. A stream that dies *mid-flight* raises too, so the caller must be ready for an exception after having already forwarded some text. """ endpoint = compat_endpoint_for_region(self._region) _guard_input_size( *(_message_text(message) for message in messages), ) body: dict[str, Any] = { "model": (model or self._model), "messages": messages, "max_tokens": max_tokens, "temperature": temperature, "stream": True, # Without this the terminal chunk carries no usage block and # every turn would record NULL tokens. "stream_options": {"include_usage": True}, } if enable_thinking is not None: body["enable_thinking"] = enable_thinking headers = { "Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json", } finish_reason: str | None = None _stream_started = time.monotonic() input_tokens: int | None = None output_tokens: int | None = None try: async with httpx.AsyncClient( timeout=self._timeout, trust_env=False, ) as client: async with client.stream( "POST", endpoint, headers=headers, json=body, ) as resp: if resp.status_code != 200: detail = (await resp.aread()).decode( "utf-8", errors="replace", ) logger.warning( "bailian_llm: HTTP %s from %s: %s", resp.status_code, endpoint, detail[:400], ) raise AppException( code="BAILIAN_LLM_HTTP_ERROR", message=( "The model provider rejected this request " f"(HTTP {resp.status_code})." ), status_code=502, ) async for line in resp.aiter_lines(): chunk = _parse_sse_data(line) if chunk is None: continue usage = chunk.get("usage") if isinstance(usage, dict): # The usage-only terminal chunk has an empty # ``choices``, so this cannot live in the loop # below. input_tokens = _as_int(usage.get("prompt_tokens")) output_tokens = _as_int( usage.get("completion_tokens"), ) for choice in chunk.get("choices") or []: if not isinstance(choice, dict): continue if choice.get("finish_reason"): finish_reason = str(choice["finish_reason"]) delta = choice.get("delta") if not isinstance(delta, dict): continue reasoning = delta.get("reasoning_content") if isinstance(reasoning, str) and reasoning: yield {"type": "reasoning", "text": reasoning} text = delta.get("content") if isinstance(text, str) and text: yield {"type": "text", "text": text} except httpx.HTTPError as exc: # Book the failed stream before re-raising: a stream that # died mid-flight consumed input tokens the vendor will # still bill, and an admin chasing "why is chat broken" # needs these rows most. await self._record_call( operation="chat_stream", ok=False, started=_stream_started, input_tokens=input_tokens, output_tokens=output_tokens, error_code="BAILIAN_LLM_NETWORK", ) raise AppException( code="BAILIAN_LLM_NETWORK", message=f"Bailian LLM streaming network error: {exc}", status_code=502, ) from exc await self._record_call( operation="chat_stream", ok=True, started=_stream_started, input_tokens=input_tokens, output_tokens=output_tokens, ) yield { "type": "done", "finish_reason": finish_reason, "input_tokens": input_tokens, "output_tokens": output_tokens, } async def embed( self, texts: list[str], *, model: str = DEFAULT_EMBED_MODEL, ) -> list[list[float]]: """Embed one or more texts via the compatible-mode embeddings endpoint. Returns one float vector per input, in the SAME order as ``texts`` (sorted by the API's per-item ``index``). Raises ``AppException`` on network / non-200 / empty — callers degrade gracefully to keyword search. Batched, because DashScope caps one request at 10 inputs (``InternalError.Algo.InvalidParameter: batch size ... should not be larger than 10``). Measured, not assumed: 10 succeeds, 11 is rejected. Without this the assistant's whole semantic layer was dead on arrival — it warms up by embedding its 273-tool catalogue in one call, every call was a 400, and the failure surfaced only as a silent fall back to keyword matching. So 「软件目录」 matched nothing (English corpus, Chinese query) and the assistant told an employee OpenAI was not in Helix, on a platform whose catalogue has held 「OPEN AI」 with a live seat plan all along. The KB path never noticed because it embeds one article at a time. """ if not texts: return [] if len(texts) > _EMBED_MAX_BATCH: chunks = [ texts[start:start + _EMBED_MAX_BATCH] for start in range(0, len(texts), _EMBED_MAX_BATCH) ] # CONCURRENT, bounded, and with a whole-call deadline. ``timeout`` is # a PER-REQUEST bound, so a 273-item warm-up run serially was 28 # requests each allowed the full budget — 14–42s on a good day, and # ~15 minutes if the backend is timing out. All of it inside one tool # dispatch of a user-facing turn, where the kernel's own turn budget # is only checked between iterations and cannot interrupt it. So the # batches overlap (a small semaphore keeps this from becoming 28 # simultaneous requests at the provider) and the whole set shares one # deadline: past it the caller degrades to keyword matching, which is # a far better answer than a turn that never returns. semaphore = asyncio.Semaphore(_EMBED_MAX_CONCURRENCY) async def _one(chunk: list[str]) -> list[list[float]]: async with semaphore: return await self.embed(chunk, model=model) # ``TaskGroup``, not a bare ``gather``. ``gather`` without # ``return_exceptions`` propagates the first failure and leaves the # SIBLINGS RUNNING: the first bad batch of a 273-item warm-up # returned in a second, the caller degraded to keyword matching, and # 27 orphan tasks went on taking the semaphore and POSTing for # another minute or two with nobody reading a single result — which # made the word "bounded" in this comment false. A task group # cancels its siblings on the first error. # # The deadline is the CONSTANT, not ``max(timeout, constant)``: that # made the documented ceiling a floor, so a caller passing # ``timeout=120`` for its per-request bound silently bought a # 120-second whole-call budget. try: batched = await asyncio.wait_for( _gather_cancelling(_one(chunk) for chunk in chunks), timeout=_EMBED_TOTAL_DEADLINE_SECONDS, ) except TimeoutError as exc: raise AppException( code="BAILIAN_EMBED_TIMEOUT", message=( f"Embedding {len(texts)} texts in {len(chunks)} batches " f"exceeded the overall deadline." ), status_code=504, ) from exc return [vector for chunk_vectors in batched for vector in chunk_vectors] endpoint = embed_endpoint_for_region(self._region) body = {"model": model, "input": texts, "encoding_format": "float"} headers = { "Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json", } try: async with httpx.AsyncClient( timeout=self._timeout, trust_env=False, ) as client: resp = await client.post(endpoint, headers=headers, json=body) except httpx.HTTPError as exc: raise AppException( code="BAILIAN_EMBED_NETWORK", message=f"Bailian embeddings network error: {exc}", status_code=502, ) from exc if resp.status_code != 200: # The embedded text is the caller's content too, and a rejection # body can quote it. Same treatment as the chat paths. logger.warning( "bailian_embed: HTTP %s from %s: %s", resp.status_code, endpoint, resp.text[:400], ) raise AppException( code="BAILIAN_EMBED_HTTP_ERROR", message=( "The embedding provider rejected this request " f"(HTTP {resp.status_code})." ), status_code=502, ) data = resp.json() items = data.get("data") or [] if not items: raise AppException( code="BAILIAN_EMBED_EMPTY", message="Bailian embeddings returned no data.", status_code=502, ) ordered = sorted(items, key=lambda d: d.get("index", 0)) return [[float(x) for x in (d.get("embedding") or [])] for d in ordered] # ── internals ────────────────────────────────────────────── async def _call( self, *, system: str, user: str, max_tokens: int, temperature: float, response_format: dict[str, str] | None, operation: str = "chat", ) -> dict[str, Any]: """Instrumented entry point: books the call, then delegates. Wrapping rather than recording inside ``_call_raw`` is what makes the FAILURE path land in the ledger too. A call that died on a 429 still happened and still tells you something — in fact the failures are the rows an admin most wants, because a key that expired shows up here as a wall of ``BAILIAN_LLM_HTTP_ERROR`` before anyone notices the feature is down. """ started = time.monotonic() try: result = await self._call_raw( system=system, user=user, max_tokens=max_tokens, temperature=temperature, response_format=response_format, ) except AppException as exc: await self._record_call( operation=operation, ok=False, started=started, error_code=getattr(exc, "code", None), ) raise except Exception as exc: await self._record_call( operation=operation, ok=False, started=started, error_code=type(exc).__name__[:64], ) raise await self._record_call( operation=operation, ok=True, started=started, input_tokens=result.get("input_tokens"), output_tokens=result.get("output_tokens"), ) return result async def _call_raw( self, *, system: str, user: str, max_tokens: int, temperature: float, response_format: dict[str, str] | None, ) -> dict[str, Any]: _guard_input_size(system, user) body: dict[str, Any] = { "model": self._model, "input": { "messages": [ {"role": "system", "content": system}, {"role": "user", "content": user}, ], }, "parameters": { "result_format": "message", "max_tokens": max_tokens, "temperature": temperature, }, } if response_format is not None: # qwen-max-2024-09 onwards. Older snapshots ignore. body["parameters"]["response_format"] = response_format headers = { "Authorization": f"Bearer {self._api_key}", "Content-Type": "application/json", } try: # Pre-launch P2 fix (Round 1, SEC-005): trust_env=False so a # rogue HTTPS_PROXY env var can't MitM Bailian / Aliyun # API calls. Matches the platform-wide pattern. async with httpx.AsyncClient( timeout=self._timeout, trust_env=False, ) as client: resp = await client.post( self._endpoint, headers=headers, json=body, ) except httpx.HTTPError as exc: raise AppException( code="BAILIAN_LLM_NETWORK", message=f"Bailian LLM network error: {exc}", status_code=502, ) from exc if resp.status_code != 200: # Body logged, never relayed. A rejection from the input filter # quotes the caller's own text back, and this message is # persisted and rendered by several callers of ``chat_json`` / # ``chat_text``. The status code stays in the message because # it is the part an operator needs and it carries no content. logger.warning( "bailian_llm: HTTP %s from %s: %s", resp.status_code, self._endpoint, resp.text[:400], ) raise AppException( code="BAILIAN_LLM_HTTP_ERROR", message=( "The model provider rejected this request " f"(HTTP {resp.status_code})." ), status_code=502, ) try: data = resp.json() except json.JSONDecodeError as exc: raise AppException( code="BAILIAN_LLM_BAD_JSON", message=( "Bailian LLM returned non-JSON HTTP 200 body." ), status_code=502, ) from exc # DashScope soft-failure: status 200 + ``code`` != ``Success`` # (or absent). Surface the code so the admin can fix the # actual reason (rate limit, bad model id, …). code = data.get("code") if code and code not in ("Success", "200", "OK"): raise AppException( code="BAILIAN_LLM_ERROR", message=( f"Bailian LLM error code={code} " f"message={data.get('message') or '(no message)'}" ), status_code=502, ) # Output extraction. Native ``result_format=message`` returns # output: { choices: [{ message: { content: "..." } }] } output = data.get("output") or {} choices = output.get("choices") or [] if choices: message = choices[0].get("message") or {} text = str(message.get("content") or "").strip() finish_reason = choices[0].get("finish_reason") else: # Older format: output: { text: "..." } text = str(output.get("text") or "").strip() finish_reason = output.get("finish_reason") usage = data.get("usage") or {} return { "output_text": text, "input_tokens": int(usage.get("input_tokens") or 0) or None, "output_tokens": int(usage.get("output_tokens") or 0) or None, # ``"length"`` means the reply was cut off at ``max_tokens``. # Probed against the live DashScope endpoint on 2026-08-17: # a capped reply comes back HTTP 200 with # ``output.choices[0].finish_reason == "length"``, which is # the ONLY way to tell "the model wrote prose" from "the # model was interrupted mid-JSON" — see ``chat_json``. "finish_reason": finish_reason, "model_used": self._model, "raw": data, } def _strip_code_fences(s: str) -> str: """Drop ``` … ``` fences if present. Keeps the inner body. Handles three common shapes: ```json\n{...}\n``` ```\n{...}\n``` {...} """ text = s.strip() if not text.startswith("```"): return text # Find the first newline after ``` and the last ``` block. first_nl = text.find("\n") if first_nl == -1: return text inner = text[first_nl + 1:] if inner.rstrip().endswith("```"): inner = inner.rstrip()[: -3].rstrip() return inner __all__ = [ "BailianLLMClient", "DEFAULT_MODEL", "DEFAULT_REGION", "REGION_CN", "REGION_INTL", "compat_endpoint_for_region", "endpoint_for_region", ]