"""AI IT copilot — grounded help-desk assistant (求助域 PR3 + Qwen activation). The unfair advantage here is NOT a generic doc chatbot: it answers grounded in the EMPLOYEE'S OWN platform data via RBAC-scoped read tools (their tickets, their submitted requests) + the published knowledge base, so it can answer "why is my laptop request stuck?" with real state instead of guessing. When it can't resolve a problem the employee escalates to a human ticket (the deterministic `/escalate` path, which works without the LLM). Pluggable LLM backend (so the AI half runs on whatever key the org has): * **bailian** (Qwen via the existing AI/LLM connector — `ai_spending.llm.*` settings + DashScope compatible-mode tool calling). This is the org's actual key, so this is the default when that connector is configured. * **anthropic** (`copilot.anthropic_api_key` / `ANTHROPIC_API_KEY`). Gating: disabled until `copilot.enabled` is true AND a backend resolves. When unavailable the chat endpoint returns a graceful "ask IT directly" fallback. """ from __future__ import annotations import json import logging import os import time from collections.abc import Awaitable, Callable from dataclasses import dataclass, field from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from app.identity.person_models import IdentityPerson from app.integrations.service import get_platform_setting logger = logging.getLogger(__name__) SETTING_ENABLED = "copilot.enabled" SETTING_PROVIDER = "copilot.provider" # "bailian" | "anthropic" | unset(auto) SETTING_ANTHROPIC_KEY = "copilot.anthropic_api_key" SETTING_ANTHROPIC_MODEL = "copilot.model" DEFAULT_ANTHROPIC_MODEL = "claude-opus-4-8" # Mirror the AI/LLM connector keys (app/domains/ai_spending/service.py) so the # copilot reuses the SAME Qwen credential the spend-summary already uses — one # place to configure. Kept as literals to avoid an import cycle. _BAILIAN_KEY = "ai_spending.llm.api_key" _BAILIAN_MODEL = "ai_spending.llm.model" _BAILIAN_REGION = "ai_spending.llm.region" _BAILIAN_EMBED_MODEL = "ai_spending.llm.embed_model" DEFAULT_BAILIAN_MODEL = "qwen-max-latest" #: Raised from 6 (2026-08-19). The assistant's write path spends iterations on #: purpose: search → propose → a server refusal that sends it back to ask the #: employee → propose again. Six left no room for the ticket guard's one forced #: reconsideration, so a legitimate sequence could exhaust the loop and fall out #: at the hardcoded 「这个问题有点复杂,建议「转人工 / 开工单」」 — telling the #: employee to open a ticket by hand, which is the exact outcome the guard exists #: to prevent, and recording it as ``answered``. The turn BUDGET #: (``_COPILOT_TURN_BUDGET_SECONDS``) is what actually bounds cost; this only #: bounds pathological looping. _MAX_TOOL_ITERATIONS = 9 _MAX_TOKENS = 1500 _MAX_HISTORY = 24 # cap turns sent to the model # Pre-file deflection runs synchronously on the request that every ticket-filer # hits — bound its LLM/embeddings HTTP calls tightly so a slow/degraded backend # degrades to "no AI answer / keyword KB" fast instead of blocking the funnel. DEFLECT_LLM_TIMEOUT = 8.0 DEFLECT_EMBED_TIMEOUT = 5.0 # Per-hop timeout inside the agent loop. Bounds ONE model call, not the turn. _COPILOT_HOP_TIMEOUT = 60.0 # Retries the Anthropic SDK may add on top of a hop. # # #2246 — ``AsyncAnthropic(timeout=...)`` bounds ONE ATTEMPT, not the # call: the SDK's ``DEFAULT_MAX_RETRIES`` is 2 and ``APITimeoutError`` is # in its retryable set, so a "60s hop" is really up to 3 x 60s plus # backoff ≈ 185s. Stacked on the turn budget that made the worst case # ≈335s — over nginx's 300s ``proxy_read_timeout``, i.e. right back to # the 504-with-an-HTML-body this whole budget exists to prevent. Zero # retries: the agent loop's own iteration IS the retry, the turn budget # is what we promise the client, and a silent triple-charge on a timing # -out model call is the opposite of a bounded hop. _COPILOT_MAX_RETRIES = 0 # Wall-clock budget for a WHOLE copilot turn. # # The per-hop timeout alone does not bound the turn: the loop runs up to # ``_MAX_TOOL_ITERATIONS`` (9) hops, so 9 x 60s = 540s worst case — past # nginx's ``proxy_read_timeout 300s`` (``web/nginx.conf``) and past the # client's own budget (``web/src/api/timeouts.ts::LLM_SYNC_TIMEOUT_MS``, # 180s). A turn that ran that long could only ever reach the user as a 504 # with an HTML body, having burned nine model calls to get there. # # The budget is checked BETWEEN hops (a hop already in flight is bounded # by ``_COPILOT_HOP_TIMEOUT``), so the real ceiling is # ``budget + hop``. #2246: the previous value made that 150 + 60 = 210s # while the CLIENT gives up at ``LLM_SYNC_TIMEOUT_MS`` = 180s — so the # worst case was still an axios abort, and the user still never saw the # "转人工 / 开工单" degradation this budget was added to guarantee. The # arithmetic has to close: budget + hop < client budget. # # 110s + 60s = 170s < 180s (client) < 300s (nginx) # # Typical turns are 1-3 hops of a few seconds and never see it. _COPILOT_TURN_BUDGET_SECONDS = 110.0 SYSTEM_PROMPT = ( "You are Helix 的 IT 助手 — an IT support copilot for employees of a ~500-person " "company. You help the signed-in employee with IT questions and problems.\n\n" "Ground every factual claim about the employee's own situation in a tool result. " "You have read-only tools that return THIS employee's data (their support tickets, " "their submitted requests/approvals) and a knowledge-base search. Use them before " "answering questions like 'where is my laptop request' or 'how do I connect VPN'. " "Never invent ticket numbers, statuses, dates, or approvers — if a tool doesn't " "return it, say you don't have that information.\n\n" "A question about what the employee is ALLOWED to do — with company data, " "code, devices or accounts — is a policy question, not a how-to. Call the " "security-policy tool for those and answer only from what it returns. A " "security answer is given on the company's behalf: inventing one is worse " "than saying the library does not cover it.\n\n" "If you cannot resolve the problem (hardware is physically broken, you need a human " "to act, the answer isn't in the employee's data), tell them clearly and suggest they " "open a support ticket — there is a '转人工 / 开工单' button in the UI for that.\n\n" "Reply in the employee's language (default 中文). Be concise and direct — lead with the " "answer. Do not include exploratory reasoning, meta-commentary, or 'Here is...' preambles " "in your reply; just answer." ) #: The knowledge-base tool's name, shared rather than spelled out in each place #: that reaches for it — the Feishu prompt names it to the model, and the chat #: turn watches for its result to collect the source links an answer cites. #: Three hardcoded copies of one string is three places to miss a rename. KB_TOOL_NAME = "search_knowledge_base" #: The approval-flow register's tool name. The knowledge base answers #: 「怎么弄」 for things with a written procedure; this answers 「该走哪个 #: 流程、找谁」 for things that go through an approval — which is most of #: what an employee actually needs IT for, and which produced 「知识库中 #: 没有…」 answers until the register existed to consult. FLOW_TOOL_NAME = "find_approval_flow" #: The security answer library's tool name. #: #: The IT knowledge base and the security answer library are two corpora with #: opposite voices, and until now only the first was reachable from a chat. The #: library's 71 entries are written in the register of a CUSTOMER SECURITY #: QUESTIONNAIRE — 「贵公司是否部署数据防泄漏(DLP)方案?」 — while an employee #: asks 「我能用个人电脑写公司代码吗」. Those two sentences share no substring, so #: the employee-facing security page had to grow its own hybrid retrieval to #: bridge them; the assistant simply could not answer the question at all, and #: filed a ticket for something the company had already written down. #: #: This is a SEPARATE tool rather than folding the library into the KB search #: because the two corpora have different owners, different review paths and #: different consequences for being wrong: an IT how-to that is out of date #: wastes ten minutes, a security statement that is out of date is a compliance #: answer given on the company's behalf. SECURITY_TOOL_NAME = "search_security_policy" # Read-only, employee-scoped tools. Every tool resolves data for the SIGNED-IN # employee only — the RBAC boundary is the tool implementation, not the prompt. TOOLS: list[dict[str, Any]] = [ { "name": "get_my_support_tickets", "description": ( "List the signed-in employee's IT help tickets (their own only): " "ticket number, title, status, category, priority, assignee." ), "input_schema": {"type": "object", "properties": {}, "additionalProperties": False}, }, { "name": "get_my_requests", "description": ( "List the signed-in employee's submitted requests across all domains " "(laptop / software seat / cloud access / DNS / VPN / API key / etc.): " "title, type, normalized status, who it's waiting on." ), "input_schema": {"type": "object", "properties": {}, "additionalProperties": False}, }, { "name": FLOW_TOOL_NAME, "description": ( "The company's approval flows (Feishu 审批) and who is responsible " "for each one. Use this whenever the employee asks how to get, " "join, request, open or apply for something — 「我想加入 X 怎么办」" "「这个要怎么申请」「XX 流程谁负责」. Returns flow names with their " "owner, so the answer can be: submit THIS flow, and ask THAT " "person. The register is bilingual and the employee will ask in " "their own words, so match on meaning, not keywords." ), "input_schema": { "type": "object", "properties": { "query": { "type": "string", "description": "What the employee wants to do, in their words.", }, }, "required": ["query"], "additionalProperties": False, }, }, { "name": KB_TOOL_NAME, "description": ( "Search the IT knowledge base (published help articles) for a how-to " "or fix. Use this FIRST for general 'how do I' / 'why does X' questions." ), "input_schema": { "type": "object", "properties": {"query": {"type": "string", "description": "Search keywords"}}, "required": ["query"], "additionalProperties": False, }, }, { "name": SECURITY_TOOL_NAME, "description": ( "The company's own SECURITY AND COMPLIANCE positions, from the " "security team's answer library. Call it for anything about what " "employees are allowed to do with company data, code, devices or " "accounts: 「我能用个人电脑写公司代码吗」「公司允许把代码传到 GitHub " "私有仓库吗」「我的笔记本需要装杀毒软件吗」「能用个人网盘存工作文件吗」" "「密码规则是什么」— and their English and Japanese equivalents.\n" "This is a different corpus from the IT knowledge base and answers a " "different question: the KB says how to do something, this says " "whether the company permits it. When a question is about permission " "or policy rather than procedure, call this one — and call it even " "if the knowledge base already returned something, because a how-to " "article is not a policy statement.\n" "Answer ONLY from what it returns. A security answer is given on the " "company's behalf; inventing one is worse than saying the library " "does not cover it and offering a ticket. When it comes back " "`covered: false`, say exactly that." ), "input_schema": { "type": "object", "properties": { "question": { "type": "string", "description": ( "The employee's question about policy or permission, in " "their own words." ), }, }, "required": ["question"], "additionalProperties": False, }, }, ] _BUILTIN_TOOL_NAMES = frozenset(t["name"] for t in TOOLS) def _openai_tools(extra: list[dict[str, Any]] | None = None) -> list[dict[str, Any]]: """TOOLS → OpenAI/compatible-mode function-tool shape (for Qwen).""" return [ { "type": "function", "function": { "name": t["name"], "description": t["description"], "parameters": t["input_schema"], }, } for t in [*TOOLS, *(extra or [])] ] #: Signature a caller must supply alongside ``extra_tools``. #: Called with (tool_name, raw_json_result) after every tool a turn runs. ToolObserver = Callable[[str, str], None] ExtraDispatch = Callable[ [str, dict[str, Any], AsyncSession, IdentityPerson], Awaitable[str], ] @dataclass class LLMConfig: provider: str # "bailian" | "anthropic" key: str model: str region: str | None = None @dataclass class CopilotResult: #: Is the copilot CONFIGURED AND ON — a capability answer, stable across #: turns. Callers gate their UI on this, so it must not swing on one bad #: turn. available: bool reply: str tool_calls: list[str] = field(default_factory=list) suggest_escalate: bool = True provider: str | None = None #: Did THIS turn fail (provider/transport error)? Orthogonal to #: ``available``: the kernel is configured and on, the call did not work. #: Conflating the two cost the Feishu channel its numbers in one direction #: (an outage counted as ``answered``) and the web panel its composer in #: the other (one hiccup disabled the input until a page reload). turn_failed: bool = False async def _resolve_llm(db: AsyncSession) -> LLMConfig | None: """Resolve the active LLM backend. Prefers an explicit `copilot.provider`; otherwise auto-picks bailian (the org's configured Qwen connector) then anthropic. Returns None when neither is configured.""" explicit = (await get_platform_setting(SETTING_PROVIDER, None, session=db) or "").strip().lower() bailian_key = await get_platform_setting(_BAILIAN_KEY, None, session=db) bailian_key = bailian_key.strip() if isinstance(bailian_key, str) else "" anthropic_key = await _resolve_anthropic_key(db) if explicit == "bailian" and bailian_key: return await _bailian_config(db, bailian_key) if explicit == "anthropic" and anthropic_key: return await _anthropic_config(db, anthropic_key) if not explicit: if bailian_key: return await _bailian_config(db, bailian_key) if anthropic_key: return await _anthropic_config(db, anthropic_key) return None async def _bailian_config(db: AsyncSession, key: str) -> LLMConfig: model = await get_platform_setting(_BAILIAN_MODEL, DEFAULT_BAILIAN_MODEL, session=db) region = await get_platform_setting(_BAILIAN_REGION, None, session=db) return LLMConfig( provider="bailian", key=key, model=str(model or DEFAULT_BAILIAN_MODEL), region=region, ) async def _anthropic_config(db: AsyncSession, key: str) -> LLMConfig: model = await get_platform_setting(SETTING_ANTHROPIC_MODEL, DEFAULT_ANTHROPIC_MODEL, session=db) return LLMConfig(provider="anthropic", key=key, model=str(model or DEFAULT_ANTHROPIC_MODEL)) async def _resolve_anthropic_key(db: AsyncSession) -> str | None: key = await get_platform_setting(SETTING_ANTHROPIC_KEY, None, session=db) if isinstance(key, str) and key.strip(): return key.strip() env = os.environ.get("ANTHROPIC_API_KEY", "").strip() return env or None async def copilot_available(db: AsyncSession) -> bool: """True iff the copilot is enabled AND a LLM backend resolves.""" enabled = await get_platform_setting(SETTING_ENABLED, False, session=db) if not enabled: return False return (await _resolve_llm(db)) is not None async def resolve_active_embed_model(db: AsyncSession) -> str: """The embedding model id currently configured (or the built-in default). Used by the KB backfill to find rows embedded under a now-stale model so a model switch actually re-embeds them. Does not require the copilot to be available — it only reads config, no network call.""" from app.domains.bailian.llm_client import DEFAULT_EMBED_MODEL embed_model = await get_platform_setting(_BAILIAN_EMBED_MODEL, None, session=db) return str(embed_model or DEFAULT_EMBED_MODEL) async def embed_texts( db: AsyncSession, texts: list[str], *, timeout: float = 60.0, ) -> tuple[list[list[float]], str] | None: """Embed texts for semantic retrieval — bailian (DashScope) only, gated on the copilot being available. Returns (vectors, model_id) or None when unavailable / non-bailian / on failure (caller degrades to keyword search). ``timeout`` bounds the HTTP call — pass a short value on synchronous user-facing paths (e.g. the pre-file deflection funnel).""" if not texts: return None if not await copilot_available(db): return None cfg = await _resolve_llm(db) if cfg is None or cfg.provider != "bailian": return None embed_model = await get_platform_setting(_BAILIAN_EMBED_MODEL, None, session=db) try: from app.domains.bailian.llm_client import ( DEFAULT_EMBED_MODEL, BailianLLMClient, ) model = str(embed_model or DEFAULT_EMBED_MODEL) client = BailianLLMClient(api_key=cfg.key, model=cfg.model, region=cfg.region, timeout=timeout, purpose="support.copilot") vectors = await client.embed(texts, model=model) if not vectors or len(vectors) != len(texts): return None return vectors, model except Exception: # noqa: BLE001 logger.exception("embedding call failed") return None async def _dispatch_tool( name: str, tool_input: dict[str, Any], db: AsyncSession, person: IdentityPerson, ) -> str: """Execute a read-only, employee-scoped tool. Returns a JSON string.""" try: if name == KB_TOOL_NAME: from app.domains.support.kb_service import SupportKbService query = str((tool_input or {}).get("query") or "")[:100] arts = await SupportKbService(db).search_published(query, limit=6) return json.dumps([ { "title": a.title, "category": a.category, "body": a.body[:1200], # Feishu-doc mirrored articles carry the source-doc link — # cite it so the employee can open the authoritative doc. "url": a.external_url, } for a in arts ], ensure_ascii=False) if name == SECURITY_TOOL_NAME: from app.core.locale import resolve_lang from app.domains.auditor import security_faq_service as security question = str((tool_input or {}).get("question") or "")[:500] if len(question.strip()) < 2: return json.dumps( {"error": "question is required"}, ensure_ascii=False, ) # Its own name, not ``result``: a second branch of this function # already binds ``result`` to a different type, and one variable # holding two unrelated shapes is what put four errors on this # file's mypy baseline the last time. found = await security.ask( db, query=question, lang=resolve_lang(person), person_id=person.id, limit=4, compose=True, # STRICTLY READ-ONLY, and that is a decision rather than an # oversight. ``record_gaps=True`` writes a row to the security # team's gap ledger — but a model may call this tool two or # three times in one turn while rephrasing, which would file # the same gap repeatedly; and a "read" tool that mutates is # the shape the repo's savepoint guard exists to catch, since a # failed insert here would poison the turn's whole transaction # several statements before anything explains it. # # The signal is not lost: the security page's own ask path # records gaps, and the knowledge loop's lint pass reads that # ledger in a batch, where one row per distinct question is # what is wanted anyway. record_gaps=False, ) # Deliberately narrow: the entries and whether they cover the # question. Everything else ``ask`` returns (confidence tiers, # per-channel provenance, rerank flags) is for the page's own # trace panel, and handing it to the model invites it to quote # retrieval internals at an employee as if they were policy. return json.dumps( { "covered": bool(found.get("covered")), "lead_answer": (found.get("answer") or {}).get("text"), "entries": [ { "id": item.get("id"), "question": item.get("question"), "answer": item.get("answer"), } for item in (found.get("items") or []) ], "note": ( "answer strictly from these entries; if covered is " "false, say the security library does not cover it" ), }, ensure_ascii=False, ) if name == FLOW_TOOL_NAME: from app.assistant.flow_grounding import build_flow_grounding query = str((tool_input or {}).get("query") or "")[:200] grounding = await build_flow_grounding(db, query) if grounding is None: # An empty register is a fact worth returning: the model # should say IT has not published the list yet, not that # the flow does not exist. return json.dumps( {"flows": [], "note": "the approval-flow register is empty"}, ensure_ascii=False, ) return json.dumps({"flows": grounding.flows}, ensure_ascii=False) if name == "get_my_support_tickets": from app.domains.support.service import SupportService rows, _ = await SupportService(db).list_for_requester( person_ids=[person.id], page=1, page_size=20, ) return json.dumps([ { "ticket_number": r.ticket_number, "title": r.title, "status": r.status, "category": r.category, "priority": r.priority, "assignee_id": str(r.assignee_id) if r.assignee_id else None, } for r in rows ], ensure_ascii=False) if name == "get_my_requests": from app.domains.me_universe.my_requests import build_my_requests result = await build_my_requests(db, person_id=person.id) return json.dumps([ { "title": it.title, "type": it.request_type, "status": it.status, "is_open": it.is_open, "waiting_on": it.pending_approver_label or it.pending_approver_kind, } for it in result.items[:30] ], ensure_ascii=False) except Exception: # noqa: BLE001 — surface a tool error to the model, don't 500 logger.exception("copilot tool %s failed for person=%s", name, person.id) return json.dumps({"error": "tool failed"}) return json.dumps({"error": f"unknown tool {name}"}) async def _dispatch_any( name: str, tool_input: dict[str, Any], db: AsyncSession, person: IdentityPerson, extra_dispatch: ExtraDispatch | None, observer: ToolObserver | None = None, ) -> str: """Built-in tools first; a caller-supplied tool cannot shadow one of ours.""" if name in _BUILTIN_TOOL_NAMES or extra_dispatch is None: result = await _dispatch_tool(name, tool_input, db, person) else: result = await extra_dispatch(name, tool_input, db, person) if observer is not None: # One choke point for every provider loop, so a caller that wants to # know what a turn actually read cannot miss half of it. observer(name, result) return result def _disabled_result() -> CopilotResult: return CopilotResult( available=False, reply=( "AI 助手当前未启用。你可以直接「转人工 / 开工单」向 IT 提交求助," "我们会尽快处理。" ), suggest_escalate=True, ) async def run_copilot_turn( db: AsyncSession, person: IdentityPerson, history: list[dict[str, str]], *, extra_tools: list[dict[str, Any]] | None = None, extra_dispatch: ExtraDispatch | None = None, system_prompt: str | None = None, observer: ToolObserver | None = None, ) -> CopilotResult: """Run one copilot turn over the supplied conversation history. Gated; returns a graceful disabled result when no backend is configured. ``system_prompt`` lets a channel replace the web copilot's instructions. The default prompt points at page affordances ("转人工 / 开工单" button) that only exist in the web UI, so a chat channel that inherited it would send employees to a button they cannot see. ``extra_tools`` / ``extra_dispatch`` let a caller add tools this module does not own — the Feishu channel uses them for its whitelisted *propose* tools. They are strictly additive: the built-in read-only set is always present, and an extra tool is only reachable when its owner supplied the dispatcher for it, so nothing here becomes callable by default. """ if not await get_platform_setting(SETTING_ENABLED, False, session=db): return _disabled_result() cfg = await _resolve_llm(db) if cfg is None: return _disabled_result() messages: list[dict[str, Any]] = [] for turn in history[-_MAX_HISTORY:]: role = turn.get("role") content = (turn.get("content") or "").strip() if role in ("user", "assistant") and content: messages.append({"role": role, "content": content}) if not messages or messages[-1]["role"] != "user": return CopilotResult(available=True, reply="请描述你的问题。", provider=cfg.provider) try: if cfg.provider == "bailian": return await _run_bailian_loop( cfg, db, person, messages, extra_tools=extra_tools, extra_dispatch=extra_dispatch, system_prompt=system_prompt, observer=observer, ) return await _run_anthropic_loop( cfg, db, person, messages, extra_tools=extra_tools, extra_dispatch=extra_dispatch, system_prompt=system_prompt, observer=observer, ) except Exception: # noqa: BLE001 — LLM/transport failure → graceful fallback logger.exception("copilot turn failed (provider=%s) person=%s", cfg.provider, person.id) return CopilotResult( # ``turn_failed``, NOT ``available=False``. A transport failure is # not an answer — saying it was cost the Feishu channel its numbers # (every provider outage recorded as ``outcome=answered``, in the # helpful-rate denominator). But it is not a switched-off feature # either: the web panel gates its composer on ``available``, so # reporting one hiccup as unavailable left the employee unable to # retry until they reloaded the page. The kernel IS configured and # on; this call did not work. Callers that know their own channel # substitute their own wording off ``turn_failed``. available=True, turn_failed=True, provider=cfg.provider, reply="AI 助手暂时不可用，你可以「转人工 / 开工单」直接联系 IT。", ) async def _run_bailian_loop( cfg: LLMConfig, db: AsyncSession, person: IdentityPerson, history: list[dict[str, Any]], *, extra_tools: list[dict[str, Any]] | None = None, extra_dispatch: ExtraDispatch | None = None, system_prompt: str | None = None, observer: ToolObserver | None = None, ) -> CopilotResult: """Qwen agent loop via DashScope compatible-mode (OpenAI tool shape).""" from app.domains.bailian.llm_client import BailianLLMClient client = BailianLLMClient( api_key=cfg.key, model=cfg.model, region=cfg.region, timeout=_COPILOT_HOP_TIMEOUT, purpose="support.copilot") tools = _openai_tools(extra_tools) messages: list[dict[str, Any]] = [ {"role": "system", "content": system_prompt or SYSTEM_PROMPT}, *history, ] tool_calls: list[str] = [] deadline = time.monotonic() + _COPILOT_TURN_BUDGET_SECONDS for _ in range(_MAX_TOOL_ITERATIONS): if time.monotonic() >= deadline: logger.warning( "copilot bailian turn exceeded %.0fs budget after %d tool calls", _COPILOT_TURN_BUDGET_SECONDS, len(tool_calls), ) break resp = await client.chat(messages=messages, tools=tools, max_tokens=_MAX_TOKENS) calls = resp.get("tool_calls") or [] if calls: messages.append({ "role": "assistant", "content": resp.get("content") or "", "tool_calls": calls, }) for tc in calls: fn = (tc.get("function") or {}) name = fn.get("name") or "" try: args = json.loads(fn.get("arguments") or "{}") except (ValueError, TypeError): args = {} tool_calls.append(name) out = await _dispatch_any( name, args, db, person, extra_dispatch, observer, ) messages.append({ "role": "tool", "tool_call_id": tc.get("id") or name, "content": out, }) continue text = (resp.get("content") or "").strip() return CopilotResult( available=True, provider="bailian", tool_calls=tool_calls, reply=text or "抱歉,我没能给出回答。你可以「转人工 / 开工单」。", ) return CopilotResult( available=True, provider="bailian", tool_calls=tool_calls, reply="这个问题有点复杂,建议「转人工 / 开工单」让 IT 同事帮你处理。", ) async def _run_anthropic_loop( cfg: LLMConfig, db: AsyncSession, person: IdentityPerson, history: list[dict[str, Any]], *, extra_tools: list[dict[str, Any]] | None = None, extra_dispatch: ExtraDispatch | None = None, system_prompt: str | None = None, observer: ToolObserver | None = None, ) -> CopilotResult: """Claude agent loop (Anthropic tool-use blocks).""" from anthropic import AsyncAnthropic # Explicit timeout: the SDK default is 600s, which would let a single # hop outlive nginx's 300s proxy_read_timeout AND the client budget. # ``max_retries=0`` (#2246) — without it the SDK silently turns this # 60s bound into ~3x60s + backoff. client = AsyncAnthropic( api_key=cfg.key, timeout=_COPILOT_HOP_TIMEOUT, max_retries=_COPILOT_MAX_RETRIES, ) messages = list(history) tool_calls: list[str] = [] deadline = time.monotonic() + _COPILOT_TURN_BUDGET_SECONDS for _ in range(_MAX_TOOL_ITERATIONS): if time.monotonic() >= deadline: logger.warning( "copilot anthropic turn exceeded %.0fs budget after %d tool calls", _COPILOT_TURN_BUDGET_SECONDS, len(tool_calls), ) break # Annotated loosely for the same reason the module has always passed # ``TOOLS`` here: these are plain dicts, not the SDK's TypedDict union. anthropic_tools: list[Any] = [*TOOLS, *(extra_tools or [])] resp = await client.messages.create( model=cfg.model, max_tokens=_MAX_TOKENS, system=system_prompt or SYSTEM_PROMPT, tools=anthropic_tools, messages=messages, ) if resp.stop_reason == "tool_use": messages.append({"role": "assistant", "content": resp.content}) results = [] for block in resp.content: if getattr(block, "type", None) == "tool_use": tool_calls.append(block.name) out = await _dispatch_any( block.name, block.input, db, person, extra_dispatch, observer, ) results.append({ "type": "tool_result", "tool_use_id": block.id, "content": out, }) messages.append({"role": "user", "content": results}) continue text = "".join( b.text for b in resp.content if getattr(b, "type", None) == "text" ).strip() return CopilotResult( available=True, provider="anthropic", tool_calls=tool_calls, reply=text or "抱歉,我没能给出回答。你可以「转人工 / 开工单」。", ) return CopilotResult( available=True, provider="anthropic", tool_calls=tool_calls, reply="这个问题有点复杂,建议「转人工 / 开工单」让 IT 同事帮你处理。", ) async def _one_shot_llm( db: AsyncSession, system: str, prompt: str, *, timeout: float = 60.0, ) -> str | None: """A single gated LLM completion (no tools), provider-agnostic. Returns None when unavailable or on failure — callers fall back deterministically. ``timeout`` bounds the HTTP call (short on synchronous user-facing paths).""" if not await get_platform_setting(SETTING_ENABLED, False, session=db): return None cfg = await _resolve_llm(db) if cfg is None: return None try: if cfg.provider == "bailian": from app.domains.bailian.llm_client import BailianLLMClient bailian = BailianLLMClient(api_key=cfg.key, model=cfg.model, region=cfg.region, timeout=timeout, purpose="support.copilot") return await bailian.chat_text(system=system, user=prompt, max_tokens=1200) from anthropic import AsyncAnthropic # Same retry cap as the agent loop (#2246): ``timeout`` bounds one # ATTEMPT, and the default 2 retries would triple the budget the # caller thinks it set. # Distinct names per branch on purpose: one ``client`` bound to two # unrelated SDK types is what put four [assignment]/[attr-defined] # errors on this file's mypy baseline. anthropic_client = AsyncAnthropic( api_key=cfg.key, timeout=timeout, max_retries=_COPILOT_MAX_RETRIES, ) resp = await anthropic_client.messages.create( model=cfg.model, max_tokens=1200, system=system, messages=[{"role": "user", "content": prompt}], ) return _anthropic_text(resp) or None except Exception: # noqa: BLE001 logger.exception("one-shot LLM failed (provider=%s)", cfg.provider) return None def _anthropic_text(resp: Any) -> str: """Concatenate the text blocks of an Anthropic response. ``resp.content`` is a union of a dozen block types and only ``TextBlock`` carries ``.text``. A ``getattr`` guard reads correctly but narrows nothing for the type checker, so writing the comprehension inline against a properly-typed client produces one [union-attr] error per sibling block type. Taking ``Any`` here says what is actually true: the SDK union is wide, this duck-types on ``type == "text"`` on purpose, and the narrowing lives in one place instead of at every call site. """ parts: list[str] = [] for block in resp.content: if getattr(block, "type", None) == "text": parts.append(str(getattr(block, "text", ""))) return "".join(parts).strip() def _strip_json_fences(text: str) -> str: """Unwrap a ```json … ``` block a model added despite being told not to. Local rather than imported from the Bailian client: that one is private to its module, and the Anthropic branch is the only reason this is needed at all (the Bailian branch strips fences on its own inside ``chat_json``). """ body = text.strip() if not body.startswith("```"): return body body = body[3:] if body[:4].lower() == "json": body = body[4:] closing = body.rfind("```") if closing != -1: body = body[:closing] return body.strip() async def one_shot_json( db: AsyncSession, system: str, prompt: str, *, timeout: float = 60.0, max_tokens: int = 1200, ) -> dict[str, Any] | None: """A single gated LLM completion that must come back as a JSON object. The provider-agnostic sibling of :func:`_one_shot_llm`. Bailian has a real ``response_format={"type":"json_object"}`` path (``chat_json``) and its own fence-stripping; Anthropic has neither, so the instruction is appended to the system prompt and the reply is parsed here. Either way the caller gets a ``dict`` or ``None`` — never prose it has to guess at. ``None`` covers every degradation: not configured, provider error, unparseable reply, a reply whose top level is not an object. Callers must treat it as "no answer available" and degrade deterministically, exactly as they do for :func:`_one_shot_llm`. Public because the evaluation harness's judge needs a *schema-bound* answer rather than a paragraph, and re-deriving provider plumbing in a second place is how the two copies drift. """ if not await get_platform_setting(SETTING_ENABLED, False, session=db): return None cfg = await _resolve_llm(db) if cfg is None: return None try: if cfg.provider == "bailian": from app.domains.bailian.llm_client import BailianLLMClient bailian = BailianLLMClient( api_key=cfg.key, model=cfg.model, region=cfg.region, timeout=timeout, purpose="support.copilot") payload = await bailian.chat_json( system=system, user=prompt, max_tokens=max_tokens, ) return payload if isinstance(payload, dict) else None from anthropic import AsyncAnthropic anthropic_client = AsyncAnthropic( api_key=cfg.key, timeout=timeout, max_retries=_COPILOT_MAX_RETRIES, ) resp = await anthropic_client.messages.create( model=cfg.model, max_tokens=max_tokens, 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." ), messages=[{"role": "user", "content": prompt}], ) text = _anthropic_text(resp) if not text: return None parsed = json.loads(_strip_json_fences(text)) return parsed if isinstance(parsed, dict) else None except Exception: # noqa: BLE001 - degradation, not failure logger.exception("one-shot JSON LLM failed (provider=%s)", cfg.provider) return None async def active_llm_identity(db: AsyncSession) -> tuple[str | None, str | None]: """``(provider, model)`` currently in force, or ``(None, None)``. Recorded on an evaluation run so a step change in the trend line can be attributed to a provider or model switch rather than investigated as a quality regression. Config read only — no network call. """ cfg = await _resolve_llm(db) if cfg is None: return None, None return cfg.provider, cfg.model _AGENT_ASSIST_SYSTEM = ( "You are assisting an IT support agent. Given a ticket and context (the " "conversation so far, similar past resolutions, and relevant knowledge-base " "articles), draft a concise, actionable reply the agent can send to the " "requester, or concrete next steps to resolve the issue. Ground your " "suggestion in the provided context; don't invent specifics. Reply in the " "requester's language (default 中文)." ) async def suggest_resolution_for_ticket( db: AsyncSession, ticket, *, agent_id=None, ) -> dict[str, Any]: """IT-side AI assist: suggest a resolution/reply for a ticket, grounded in its timeline + similar resolved tickets + relevant KB. Gated. Similar-ticket recall is shingle-based (``find_similar_resolved``) — the old first-whitespace-token ILIKE was an effective no-op for Chinese titles. Every non-empty draft is persisted to the ai-suggest ledger so the adoption rate (agent reply ≈ suggestion) is measurable.""" if not await copilot_available(db): return {"available": False, "suggestion": ""} from app.domains.support.kb_service import SupportKbService from app.domains.support.models import SupportAiSuggestion from app.domains.support.service import SupportService svc = SupportService(db) messages = await svc.list_messages(ticket.id) similar = await svc.find_similar_resolved( title=ticket.title, body=ticket.body, exclude_id=ticket.id, limit=3, ) # KB search grounds on title+body (semantic first) — mirrored Feishu # docs are part of this surface since PR-1. kb = await SupportKbService(db).search_published( f"{ticket.title} {ticket.body}"[:100], limit=3, ) parts = [ f"工单: {ticket.title}\n分类: {ticket.category}\n描述: {ticket.body}", "对话记录:\n" + "\n".join(f"- {m.body}" for m in messages[-8:]) if messages else "", "相似的已解决工单:\n" + "\n".join( f"- {s.title} → {(s.resolution or '')[:300]}" for s in similar ) if similar else "", "相关知识库:\n" + "\n".join(f"- {a.title}: {a.body[:300]}" for a in kb) if kb else "", ] prompt = "\n\n".join(p for p in parts if p) suggestion = await _one_shot_llm(db, _AGENT_ASSIST_SYSTEM, prompt) if suggestion: cfg = await _resolve_llm(db) db.add(SupportAiSuggestion( ticket_id=ticket.id, agent_id=agent_id, suggestion=suggestion, model=cfg.model if cfg else None, )) return {"available": True, "suggestion": suggestion or ""} _DEFLECT_SYSTEM = ( "You are Helix 的 IT 助手 helping an employee BEFORE they file a support " "ticket. Given their problem and relevant knowledge-base articles, write a " "short, concrete answer (中文) that tries to solve it from the articles. " "Ground every step in the provided articles — if they don't cover it, say so " "briefly and suggest filing a ticket. Be concise (a few sentences / steps), " "no preamble." ) async def deflect_answer( db: AsyncSession, query: str, articles: list, *, timeout: float = DEFLECT_LLM_TIMEOUT, ) -> str | None: """A single grounded answer for the pre-file deflection funnel, over the KB articles already found. Gated + best-effort; None when unavailable or empty. No tools (cheap, fast) — the article search happened in the caller. Bounded by a SHORT timeout: this is a synchronous, every-ticket-filer path, so a slow backend must degrade to "no answer" quickly rather than block the funnel.""" if not await copilot_available(db): return None if not articles: return None kb_text = "\n\n".join( f"【{a.title}】\n{(a.body or '')[:1200]}" for a in articles[:4] ) prompt = f"员工的问题:\n{query}\n\n相关知识库文章:\n{kb_text}" return await _one_shot_llm(db, _DEFLECT_SYSTEM, prompt, timeout=timeout) async def draft_kb_from_ticket(db: AsyncSession, ticket) -> dict[str, str]: """Draft a KB article (title/body/category) from a resolved ticket. Uses the LLM when available; otherwise a deterministic draft from the resolution.""" deterministic = { "title": (ticket.title or "")[:200] or "IT 知识库条目", "body": (ticket.resolution or ticket.body or "").strip()[:8000], "category": ticket.category or "other", } system = ( "Turn a resolved IT support ticket into a reusable knowledge-base article " "for future self-service. Output STRICT JSON: " '{"title": "...", "body": "...", "category": "..."}. The body should be a ' "general how-to/fix (markdown ok), not specific to this one person. " "category ∈ [hardware, account_access, software, network, facility, other]. " "Write in 中文." ) prompt = ( f"标题: {ticket.title}\n分类: {ticket.category}\n问题: {ticket.body}\n" f"解决方案: {ticket.resolution or '(无)'}" ) out = await _one_shot_llm(db, system, prompt) if out: try: cleaned = out.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip() data = json.loads(cleaned) title = str(data.get("title") or "").strip() body = str(data.get("body") or "").strip() cat = str(data.get("category") or "").strip() if title and body: from app.domains.support.models import CATEGORIES return { "title": title[:200], "body": body[:8000], "category": cat if cat in CATEGORIES else "other", } except (ValueError, TypeError): pass return deterministic _TRIAGE_SYSTEM = ( "Classify an IT support ticket. Output STRICT JSON: " '{"category": "...", "priority": "..."}. ' "category ∈ [hardware, account_access, software, network, facility, other]. " "priority ∈ [low, normal, high, urgent] — base it on urgency/impact: someone " "fully blocked from working or a security/account-compromise issue = high/urgent; " "a convenience or future request = low/normal. No prose, JSON only." ) async def triage_ticket(db: AsyncSession, ticket) -> dict[str, Any] | None: """AI auto-triage: classify a fresh ticket's category/priority and fill DEFAULTS via SupportService.apply_ai_triage (never overrides explicit choices). Gated; returns the applied change list, or None when the copilot is unavailable / the model output was unusable.""" if not await copilot_available(db): return None out = await _one_shot_llm( db, _TRIAGE_SYSTEM, f"标题: {ticket.title}\n描述: {ticket.body}", ) if not out: return None try: cleaned = out.strip().removeprefix("```json").removeprefix("```").removesuffix("```").strip() data = json.loads(cleaned) except (ValueError, TypeError): return None if not isinstance(data, dict): return None from app.domains.support.service import SupportService changed = await SupportService(db).apply_ai_triage( ticket=ticket, category=(str(data.get("category") or "").strip() or None), priority=(str(data.get("priority") or "").strip() or None), ) return {"changed": changed} async def run_ticket_triage(ticket_id) -> None: """Background entrypoint — fire-and-forget after ticket create (so create isn't blocked on LLM latency). Opens a fresh session, triages the ticket if it is still fresh, commits. Exceptions are caught/logged by spawn().""" from app.core.database import async_session_factory from app.domains.support.models import ( STATUS_OPEN, STATUS_TRIAGING, SupportTicket, ) async with async_session_factory() as db: ticket = await db.get(SupportTicket, ticket_id) # Only touch a still-fresh ticket — never fight an agent who already # claimed / recategorised it between create and this background run. if ticket is None or ticket.status not in (STATUS_OPEN, STATUS_TRIAGING): return result = await triage_ticket(db, ticket) if result is not None: await db.commit() def draft_ticket_from_history(history: list[dict[str, str]]) -> tuple[str, str]: """Deterministically draft a ticket (title, body) from a copilot conversation — used by the escalate path so it works WITHOUT the LLM.""" first_user = next( (t.get("content", "").strip() for t in history if t.get("role") == "user"), "", ) title = (first_user.splitlines()[0] if first_user else "AI 求助升级")[:200] or "AI 求助升级" lines = ["（由 AI 助手对话升级）", ""] for t in history: role = t.get("role") if role not in ("user", "assistant"): continue who = "我" if role == "user" else "AI 助手" lines.append(f"{who}: {(t.get('content') or '').strip()}") return title, "\n".join(lines)[:8000] __all__ = [ "CopilotResult", "SECURITY_TOOL_NAME", "active_llm_identity", "copilot_available", "deflect_answer", "embed_texts", "draft_kb_from_ticket", "draft_ticket_from_history", "one_shot_json", "resolve_active_embed_model", "run_copilot_turn", "run_ticket_triage", "suggest_resolution_for_ticket", "triage_ticket", ]