import json from datetime import datetime, timezone from typing import Protocol import httpx from app.config import Settings from app.models import LeaveDraftSuggestion, LeaveDraftSuggestionRequest, LeaveProgressAnswerRequest class SuggestionGateway(Protocol): async def suggest(self, request: LeaveDraftSuggestionRequest) -> LeaveDraftSuggestion: ... class ProgressGateway(Protocol): async def answer_progress(self, request: LeaveProgressAnswerRequest) -> str: ... class QwenSuggestionGateway: def __init__(self, settings: Settings) -> None: self.settings = settings async def suggest(self, request: LeaveDraftSuggestionRequest) -> LeaveDraftSuggestion: if not self.settings.qwen_api_key: raise QwenConfigurationError("QWEN_API_KEY is not configured") reference_time = request.now or datetime.now(timezone.utc) payload = { "model": self.settings.qwen_model, "temperature": 0.1, "response_format": {"type": "json_object"}, "messages": [ {"role": "system", "content": SYSTEM_PROMPT}, { "role": "user", "content": json.dumps( { "text": request.text, "timezone": request.timezone, "referenceTime": reference_time.isoformat(), }, ensure_ascii=False, ), }, ], } async with httpx.AsyncClient(timeout=self.settings.request_timeout_seconds) as client: response = await client.post( f"{self.settings.qwen_base_url}/chat/completions", headers={"Authorization": f"Bearer {self.settings.qwen_api_key}"}, json=payload, ) if response.status_code >= 400: raise QwenUpstreamError(f"Qwen returned HTTP {response.status_code}") try: content = response.json()["choices"][0]["message"]["content"] return LeaveDraftSuggestion.model_validate_json(content) except (KeyError, IndexError, TypeError, ValueError) as exc: raise QwenUpstreamError("Qwen returned an invalid structured response") from exc async def answer_progress(self, request: LeaveProgressAnswerRequest) -> str: if not self.settings.qwen_api_key: raise QwenConfigurationError("QWEN_API_KEY is not configured") payload = { "model": self.settings.qwen_model, "temperature": 0.1, "response_format": {"type": "json_object"}, "messages": [ {"role": "system", "content": PROGRESS_SYSTEM_PROMPT}, {"role": "user", "content": request.model_dump_json()}, ], } async with httpx.AsyncClient(timeout=self.settings.request_timeout_seconds) as client: response = await client.post(f"{self.settings.qwen_base_url}/chat/completions", headers={"Authorization": f"Bearer {self.settings.qwen_api_key}"}, json=payload) if response.status_code >= 400: raise QwenUpstreamError(f"Qwen returned HTTP {response.status_code}") try: content = response.json()["choices"][0]["message"]["content"] answer = json.loads(content)["answer"] if not isinstance(answer, str) or not answer.strip() or len(answer) > 2000: raise ValueError("invalid answer") return answer.strip() except (KeyError, IndexError, TypeError, ValueError, json.JSONDecodeError) as exc: raise QwenUpstreamError("Qwen returned an invalid progress answer") from exc SYSTEM_PROMPT = """ 你是企业 OA 请假表单解析器。只把用户自然语言转换为 JSON 建议值,不执行任何业务动作。 输出必须是一个 JSON 对象,只允许字段:type、startsAt、endsAt、reason、assumptions、needsClarification。 type 只能是 PERSONAL、SICK、ANNUAL 或 null。时间必须是带时区偏移的 ISO-8601。 不能确定的值输出 null,并把需要用户补充的问题写入 needsClarification。 不得输出申请人、审批人、租户、权限、流程或隐藏字段。不得使用 Markdown。 """.strip() PROGRESS_SYSTEM_PROMPT = """ 你是企业 OA 只读流程进度助手。数据库记录已经由业务后端鉴权并选定,你只能依据输入 context 回答 question。 输出必须是 JSON 对象且只包含 answer 字段。回答必须明确当前 status;有 activeTaskNames 时说明当前节点;流程结束时说明已结束。 不得猜测审批人、原因、流程变量或预计完成时间,不得给出批准、驳回、撤回、提交等写操作指令,不得使用 Markdown。 """.strip() class QwenConfigurationError(RuntimeError): pass class QwenUpstreamError(RuntimeError): pass