# -*- coding: utf-8 -*- """Fast, tool-free chat path for the backend-configured Grok Agent model. The Grok Build runtime is intentionally skipped for ordinary language turns. Only requests classified as requiring live data, tools, or side effects should be sent through the Agent path. """ from __future__ import annotations import json import re import threading import urllib.error import urllib.parse import urllib.request from dataclasses import dataclass from typing import Callable, Mapping, Sequence from grok_build_bridge import GrokBuildManager class DirectChatError(RuntimeError): """The configured custom model could not complete a direct chat turn.""" class DirectChatCancelled(DirectChatError): """The caller cancelled an in-flight direct chat stream.""" class DirectChatCancellation: """Thread-safe cancellation handle that also closes the active HTTP stream.""" def __init__(self) -> None: self._event = threading.Event() self._lock = threading.Lock() self._response: object | None = None self._cancel_callback: Callable[[], None] | None = None self._cancel_callback_started = False @property def cancelled(self) -> bool: return self._event.is_set() def cancel(self) -> None: self._event.set() with self._lock: response = self._response callback = self._take_cancel_callback_locked() if response is not None: try: response.close() except (AttributeError, OSError, ValueError): pass if callback is not None: threading.Thread(target=callback, daemon=True).start() def _take_cancel_callback_locked(self) -> Callable[[], None] | None: if self._cancel_callback_started or self._cancel_callback is None: return None self._cancel_callback_started = True return self._cancel_callback def set_cancel_callback(self, callback: Callable[[], None]) -> None: with self._lock: self._cancel_callback = callback pending = self._event.is_set() selected = self._take_cancel_callback_locked() if pending else None if selected is not None: threading.Thread(target=selected, daemon=True).start() def attach(self, response: object) -> None: with self._lock: if self._event.is_set(): try: response.close() except (AttributeError, OSError, ValueError): pass raise DirectChatCancelled("普通对话已停止") self._response = response def detach(self, response: object) -> None: with self._lock: if self._response is response: self._response = None def raise_if_cancelled(self) -> None: if self._event.is_set(): raise DirectChatCancelled("普通对话已停止") @dataclass(frozen=True) class DirectChatResult: text: str model: str protocol: str conversation_id: str = "" _MAX_HISTORY_MESSAGES = 12 _MAX_HISTORY_CHARS = 24_000 _MAX_ANSWER_CHARS = 200_000 _MAX_HTTP_BODY_BYTES = 32 * 1024 * 1024 _MAX_SSE_LINE_BYTES = 1024 * 1024 class _NoRedirectHandler(urllib.request.HTTPRedirectHandler): def redirect_request(self, req, fp, code, msg, headers, newurl): return None _HTTP_OPENER = urllib.request.build_opener(_NoRedirectHandler()) _EXPLICIT_AGENT_RE = re.compile( r"(?:使用|调用|启动|交给|让)\s*(?:Grok\s*Build\s*)?(?:Agent|智能体|代理|MCP|工具)", re.IGNORECASE, ) _LIVE_DATA_RE = re.compile( r"(天气|气温|降雨|空气质量|新闻|热搜|股价|股票行情|汇率|航班|火车票|" r"物流|快递|订单状态|库存|实时数据|最新数据|今天几号|现在几点)" ) _LIVE_INTENT_RE = re.compile( r"(查(?:一下)?|查询|查找|搜索|检索|获取|看看|看一下|告诉我|怎么样|多少|是否)" ) _SIDE_EFFECT_RE = re.compile( r"(打开|运行|执行|测试|安装|卸载|更新|下载|上传|发送|发布|部署|创建|新建|" r"删除|移除|保存|读取|查看|写入|改动|修改|编辑|修复|重命名|复制到|移动到)" ) _TOOL_OBJECT_RE = re.compile( r"(文件|目录|文件夹|项目|代码库|仓库|终端|命令|脚本|程序|浏览器|网页|" r"网站|网址|链接|GitHub|数据库|日志|截图|图片|Excel|表格|文档|PDF)", re.IGNORECASE, ) _WECOM_OBJECT_RE = re.compile( r"(企业微信|企微|通讯录|联系人|群聊|聊天记录|消息|日程|会议|待办)" ) _WECOM_ACTION_RE = re.compile( r"(查|看|获取|发送|创建|安排|预约|取消|更新|删除|添加|移除|回复)" ) _URL_RE = re.compile(r"https?://|www\.", re.IGNORECASE) _URL_ACTION_RE = re.compile(r"(打开|访问|读取|总结|分析|查询|下载|抓取|搜索)") _IDENTITY_RE = re.compile( r"(你是谁|你是(?:什么|哪个|哪一个|哪种)?模型|" r"(?:什么|哪个|哪一个|哪种)模型|模型(?:名称|版本|型号)|底层模型)" ) def classify_chat_route(text: str, *, last_route: str = "") -> str: """Return ``direct`` for language chat or ``agent`` for executable work.""" value = " ".join(str(text or "").strip().split()) if not value: return "direct" if ( last_route == "agent" and len(value) <= 12 and re.fullmatch( r"(?:[?？]|继续|然后呢|还有呢|明天呢|后天呢|再查一下|再看看|详细点)[?？]?", value, ) ): return "agent" if re.match(r"^(?:@|/)\s*(?:agent|智能体)\b", value, re.IGNORECASE): return "agent" if _EXPLICIT_AGENT_RE.search(value): return "agent" if _LIVE_DATA_RE.search(value) and ( _LIVE_INTENT_RE.search(value) or any(token in value for token in ("今天", "现在", "最新", "实时", "帮我")) ): return "agent" if _SIDE_EFFECT_RE.search(value) and _TOOL_OBJECT_RE.search(value): return "agent" if _WECOM_OBJECT_RE.search(value) and _WECOM_ACTION_RE.search(value): return "agent" if _URL_RE.search(value) and _URL_ACTION_RE.search(value): return "agent" return "direct" def configured_identity_reply(text: str, model: str) -> str: """Return a truthful instant model-identity answer when applicable.""" if not _IDENTITY_RE.search(str(text or "")): return "" configured = str(model or "").strip() or "后台配置的自有模型" return ( f"当前使用的是后台配置的自有模型：{configured}。" "普通对话直接由该模型回答；只有需要查询或执行工具时才启动 Grok Build Agent。" ) def _bounded_history(history: Sequence[Mapping[str, object]] | None) -> list[dict[str, str]]: if not history: return [] selected: list[dict[str, str]] = [] total = 0 for item in reversed(list(history)[-_MAX_HISTORY_MESSAGES:]): role = str(item.get("role") or "").strip() content = str(item.get("content") or "").strip() if role not in {"user", "assistant"} or not content: continue remaining = _MAX_HISTORY_CHARS - total if remaining <= 0: break content = content[-remaining:] selected.append({"role": role, "content": content}) total += len(content) selected.reverse() return selected def _headers(api_key: str, auth_scheme: str, *, anthropic: bool = False) -> dict[str, str]: headers = { "Content-Type": "application/json", "Accept": "application/json", "User-Agent": "ZhenYangTang-RPA-Direct-Chat/1.0", } if auth_scheme == "x_api_key": headers["x-api-key"] = api_key else: headers["Authorization"] = f"Bearer {api_key}" if anthropic: headers["anthropic-version"] = "2023-06-01" return headers def _system_prompt(model: str) -> str: return ( "你是当前软件内的 AI 客服对话助手。请直接、自然、简洁地回答用户。" "你的推理由后台配置的自有模型完成，不得声称自己是 xAI、Grok、" "OpenAI、Claude 或其他未配置的厂商模型。" f"后台当前配置的模型名称是 {model}。" "本次是无工具普通对话：不要声称已经查询实时信息、访问网页、读取文件、" "执行命令或完成外部操作。若用户确实要求这些操作，请说明需要切换到 Agent。" ) @dataclass(frozen=True) class _PreparedChat: prompt: str model: str protocol: str url: str headers: Mapping[str, str] payload: Mapping[str, object] timeout: int conversation_id: str identity: str = "" def _prepare_chat( message: str, *, history: Sequence[Mapping[str, object]] | None, user_id: str, conversation_id: str, manager: GrokBuildManager | None, streaming: bool, ) -> _PreparedChat: prompt = str(message or "").strip() if not prompt: raise DirectChatError("消息不能为空") runtime = manager or GrokBuildManager() settings = runtime.load_ai_settings() if not bool(settings.get("GROK_MODEL_ENABLED", False)): raise DirectChatError("后台尚未启用 Agent 自有模型") profile = runtime.model_profile(settings) if not profile.compatible: raise DirectChatError(profile.reason or "后台自有模型配置不可用") api_key = str(settings.get("GROK_API_KEY") or "").strip() if not api_key: raise DirectChatError("后台自有模型缺少 API Key") try: configured_timeout = int(settings.get("GROK_CUSTOMER_SERVICE_TIMEOUT", 180)) except (TypeError, ValueError): configured_timeout = 180 timeout = min(90, max(10, configured_timeout)) bounded = _bounded_history(history) identity = configured_identity_reply(prompt, profile.model) system = _system_prompt(profile.model) protocol = profile.api_backend url = profile.base_url.rstrip("/") headers = _headers(api_key, profile.auth_scheme) if protocol == "dify": query_parts = [system] if bounded and not conversation_id: query_parts.append( "以下是最近对话：\n" + "\n".join( f"{'用户' if item['role'] == 'user' else '助手'}：{item['content']}" for item in bounded ) ) query_parts.append(f"用户当前消息：{prompt}") payload: dict[str, object] = { "inputs": ( settings.get("GROK_DIFY_INPUTS") if isinstance(settings.get("GROK_DIFY_INPUTS"), Mapping) else {} ), "query": "\n\n".join(query_parts), "response_mode": "streaming" if streaming else "blocking", "user": re.sub(r"[^A-Za-z0-9_-]+", "-", user_id)[:64] or "wechat-rpa-chat", } if conversation_id: payload["conversation_id"] = conversation_id url = f"{url}/chat-messages" elif protocol == "chat_completions": payload = { "model": profile.model, "messages": [ {"role": "system", "content": system}, *bounded, {"role": "user", "content": prompt}, ], "temperature": profile.temperature, "max_tokens": profile.max_completion_tokens, "stream": streaming, } url = f"{url}/chat/completions" elif protocol == "responses": payload = { "model": profile.model, "instructions": system, "input": [*bounded, {"role": "user", "content": prompt}], "temperature": profile.temperature, "max_output_tokens": profile.max_completion_tokens, "stream": streaming, } url = f"{url}/responses" elif protocol == "messages": payload = { "model": profile.model, "system": system, "messages": [*bounded, {"role": "user", "content": prompt}], "temperature": profile.temperature, "max_tokens": profile.max_completion_tokens, "stream": streaming, } url = f"{url}/messages" headers = _headers(api_key, profile.auth_scheme, anthropic=True) else: raise DirectChatError(f"普通对话暂不支持接口协议：{protocol}") if streaming: headers = {**headers, "Accept": "text/event-stream"} return _PreparedChat( prompt=prompt, model=profile.model, protocol=protocol, url=url, headers=headers, payload=payload, timeout=timeout, conversation_id=conversation_id, identity=identity, ) def _post_json( url: str, *, headers: Mapping[str, str], payload: Mapping[str, object], timeout: int, ) -> Mapping[str, object]: request = urllib.request.Request( url, data=json.dumps(payload, ensure_ascii=False).encode("utf-8"), headers=dict(headers), method="POST", ) try: with _HTTP_OPENER.open(request, timeout=timeout) as response: status = int(getattr(response, "status", 200) or 200) body = response.read(_MAX_HTTP_BODY_BYTES + 1) except urllib.error.HTTPError as exc: try: detail = exc.read().decode("utf-8", errors="replace")[:500] except OSError: detail = str(exc) raise DirectChatError( f"后台自有模型请求失败（HTTP {exc.code}）：{detail}" ) from exc except (urllib.error.URLError, TimeoutError, OSError) as exc: raise DirectChatError(f"无法连接后台自有模型：{exc}") from exc if not 200 <= status < 300: detail = body.decode("utf-8", errors="replace")[:500] raise DirectChatError( f"后台自有模型请求失败（HTTP {status}）：{detail}" ) if len(body) > _MAX_HTTP_BODY_BYTES: raise DirectChatError("后台自有模型响应过大，已停止读取") try: data = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise DirectChatError("后台自有模型返回了无效 JSON") from exc if not isinstance(data, Mapping): raise DirectChatError("后台自有模型响应结构无效") return data def _answer_from_responses(data: Mapping[str, object]) -> str: output_text = data.get("output_text") if isinstance(output_text, str) and output_text.strip(): return output_text.strip() chunks: list[str] = [] output = data.get("output") if isinstance(output, list): for item in output: if not isinstance(item, Mapping): continue content = item.get("content") if not isinstance(content, list): continue for part in content: if not isinstance(part, Mapping): continue text = part.get("text") if isinstance(text, str): chunks.append(text) return "".join(chunks).strip() def _answer_from_messages(data: Mapping[str, object]) -> str: chunks: list[str] = [] content = data.get("content") if isinstance(content, list): for part in content: if isinstance(part, Mapping) and part.get("type") == "text": text = part.get("text") if isinstance(text, str): chunks.append(text) return "".join(chunks).strip() def _error_message(data: Mapping[str, object], fallback: str) -> str: error = data.get("error") if isinstance(error, Mapping): return str(error.get("message") or error.get("type") or fallback) return str(data.get("message") or error or fallback) def _result_from_json( prepared: _PreparedChat, data: Mapping[str, object], ) -> DirectChatResult: next_conversation_id = prepared.conversation_id if prepared.protocol == "dify": answer = data.get("answer") if not isinstance(answer, str): nested = data.get("data") answer = nested.get("answer") if isinstance(nested, Mapping) else "" next_conversation_id = str( data.get("conversation_id") or prepared.conversation_id ) elif prepared.protocol == "chat_completions": choices = data.get("choices") answer = "" if isinstance(choices, list) and choices and isinstance(choices[0], Mapping): message_data = choices[0].get("message") if isinstance(message_data, Mapping): answer = message_data.get("content") elif prepared.protocol == "responses": answer = _answer_from_responses(data) else: answer = _answer_from_messages(data) text = str(answer or "").strip() if not text: raise DirectChatError("后台自有模型没有返回有效文本") if len(text) > _MAX_ANSWER_CHARS: raise DirectChatError("后台自有模型回复过长，已拒绝显示") return DirectChatResult( text=text, model=prepared.model, protocol=prepared.protocol, conversation_id=next_conversation_id, ) def _open_stream_response(prepared: _PreparedChat): request = urllib.request.Request( prepared.url, data=json.dumps(prepared.payload, ensure_ascii=False).encode("utf-8"), headers=dict(prepared.headers), method="POST", ) try: return _HTTP_OPENER.open(request, timeout=prepared.timeout) except urllib.error.HTTPError as exc: try: detail = exc.read(501).decode("utf-8", errors="replace")[:500] except OSError: detail = str(exc) raise DirectChatError( f"后台自有模型请求失败（HTTP {exc.code}）：{detail}" ) from exc except (urllib.error.URLError, TimeoutError, OSError) as exc: raise DirectChatError(f"无法连接后台自有模型：{exc}") from exc def _stop_dify_task(prepared: _PreparedChat, task_id: str, user_id: str) -> None: safe_task_id = urllib.parse.quote(task_id, safe="") stop_url = f"{prepared.url.rstrip('/')}/{safe_task_id}/stop" request = urllib.request.Request( stop_url, data=json.dumps({"user": user_id}, ensure_ascii=False).encode("utf-8"), headers=dict(prepared.headers), method="POST", ) try: with _HTTP_OPENER.open(request, timeout=5) as response: response.read(1024) except Exception: pass def _read_json_response(response: object) -> Mapping[str, object]: body = response.read(_MAX_HTTP_BODY_BYTES + 1) if len(body) > _MAX_HTTP_BODY_BYTES: raise DirectChatError("后台自有模型响应过大，已停止读取") try: data = json.loads(body.decode("utf-8")) except (UnicodeDecodeError, json.JSONDecodeError) as exc: raise DirectChatError("后台自有模型返回了无效 JSON") from exc if not isinstance(data, Mapping): raise DirectChatError("后台自有模型响应结构无效") return data def _iter_sse_events( response: object, cancellation: DirectChatCancellation, ): event_name = "" data_lines: list[str] = [] total_bytes = 0 while True: cancellation.raise_if_cancelled() raw = response.readline(_MAX_SSE_LINE_BYTES + 1) if not raw: break total_bytes += len(raw) if total_bytes > _MAX_HTTP_BODY_BYTES: raise DirectChatError("后台自有模型流式响应过大，已停止读取") if len(raw) > _MAX_SSE_LINE_BYTES: raise DirectChatError("后台自有模型 SSE 单行数据过大") try: line = raw.decode("utf-8").rstrip("\r\n") except UnicodeDecodeError as exc: raise DirectChatError("后台自有模型 SSE 不是有效 UTF-8") from exc if line.startswith("\ufeff"): line = line.lstrip("\ufeff") if not line: if event_name or data_lines: yield event_name, "\n".join(data_lines) event_name = "" data_lines = [] continue if line.startswith(":"): continue field, separator, value = line.partition(":") if separator and value.startswith(" "): value = value[1:] if field == "event": event_name = value elif field == "data": data_lines.append(value) if event_name or data_lines: yield event_name, "\n".join(data_lines) def stream_direct_chat( message: str, *, on_update: Callable[[str, bool], None], history: Sequence[Mapping[str, object]] | None = None, user_id: str = "wechat-rpa-chat", conversation_id: str = "", manager: GrokBuildManager | None = None, cancellation: DirectChatCancellation | None = None, ) -> DirectChatResult: """Stream a tool-free model turn and return only after a valid terminal event. ``on_update(text, replace)`` receives text deltas. ``replace=True`` is used by Dify's ``message_replace`` snapshot event. """ prepared = _prepare_chat( message, history=history, user_id=user_id, conversation_id=conversation_id, manager=manager, streaming=True, ) cancel = cancellation or DirectChatCancellation() cancel.raise_if_cancelled() if prepared.identity: on_update(prepared.identity, False) return DirectChatResult( text=prepared.identity, model=prepared.model, protocol=prepared.protocol, conversation_id="", ) response = _open_stream_response(prepared) cancel.attach(response) chunks: list[str] = [] current_chars = 0 finished = False next_conversation_id = prepared.conversation_id workflow_started = False workflow_finished = False seen_agent_message = False completion_finish_reason = "" dify_task_id = "" anthropic_started = False anthropic_message_delta = False anthropic_stop_reason = "" anthropic_open_blocks: set[int] = set() def emit(text: object, *, replace: bool = False) -> None: nonlocal chunks, current_chars cancel.raise_if_cancelled() value = str(text or "") if not value: return new_size = len(value) if replace else current_chars + len(value) if new_size > _MAX_ANSWER_CHARS: raise DirectChatError("后台自有模型回复过长，已停止生成") if replace: chunks = [value] current_chars = len(value) else: chunks.append(value) current_chars = new_size try: on_update(value, replace) except Exception as exc: raise DirectChatError("界面无法接收流式内容") from exc try: status = int(getattr(response, "status", 200) or 200) if not 200 <= status < 300: detail = response.read(501).decode("utf-8", errors="replace")[:500] raise DirectChatError( f"后台自有模型请求失败（HTTP {status}）：{detail}" ) headers = getattr(response, "headers", {}) content_type = str(headers.get("Content-Type", "") or "").lower() if "application/json" in content_type: result = _result_from_json(prepared, _read_json_response(response)) emit(result.text) return result try: events = _iter_sse_events(response, cancel) for sse_name, raw_data in events: cancel.raise_if_cancelled() value = raw_data.strip() if not value: continue if value == "[DONE]": if prepared.protocol == "chat_completions": finished = True elif prepared.protocol == "responses": finished = True break try: event = json.loads(value) except json.JSONDecodeError as exc: raise DirectChatError( "后台自有模型 SSE data 不是合法 JSON" ) from exc if not isinstance(event, Mapping): continue event_type = str(event.get("event") or event.get("type") or sse_name) if prepared.protocol == "dify": task_id = str(event.get("task_id") or "").strip() if task_id and not dify_task_id: dify_task_id = task_id dify_user = str( prepared.payload.get("user") or "wechat-rpa-chat" ) cancel.set_cancel_callback( lambda current_task=task_id, current_user=dify_user: _stop_dify_task( prepared, current_task, current_user, ) ) next_conversation_id = str( event.get("conversation_id") or next_conversation_id ) if event_type == "error": raise DirectChatError( _error_message(event, "Dify 流式响应返回错误") ) if event_type == "workflow_started": workflow_started = True elif event_type in {"message", "agent_message"}: answer = str(event.get("answer") or "") if event_type == "agent_message": seen_agent_message = True emit(answer) elif seen_agent_message: emit(answer, replace=True) else: emit(answer) elif event_type == "message_replace": emit(event.get("answer"), replace=True) elif event_type == "text_chunk": data = event.get("data") if isinstance(data, Mapping): emit(data.get("text")) elif event_type == "message_end": finished = True elif event_type in {"workflow_finished", "node_finished"}: data = event.get("data") status_value = ( str(data.get("status") or "").lower() if isinstance(data, Mapping) else "" ) if status_value in {"failed", "error", "stopped"}: raise DirectChatError( str( data.get("error") or data.get("message") or f"Dify {event_type} 失败" ) ) if event_type == "workflow_finished": workflow_finished = True if finished and (not workflow_started or workflow_finished): break elif prepared.protocol == "chat_completions": if event_type == "error" or "error" in event: raise DirectChatError( _error_message(event, "Chat Completions 流式响应返回错误") ) choices = event.get("choices") if not isinstance(choices, list): continue for choice in choices: if not isinstance(choice, Mapping): continue delta = choice.get("delta") if isinstance(delta, Mapping): content = delta.get("content") if isinstance(content, str): emit(content) elif isinstance(content, list): for part in content: if isinstance(part, Mapping): emit(part.get("text")) if delta.get("tool_calls") or delta.get("function_call"): raise DirectChatError( "普通对话模型返回了工具调用，已拒绝执行" ) reason = choice.get("finish_reason") if reason is not None: completion_finish_reason = str(reason) elif prepared.protocol == "responses": if event_type in {"error", "response.failed", "response.incomplete", "response.cancelled"}: raise DirectChatError( _error_message(event, f"Responses 流式响应未完成：{event_type}") ) if event_type in {"response.output_text.delta", "response.refusal.delta"}: emit(event.get("delta")) elif event_type == "response.output_text.done" and not chunks: emit(event.get("text")) elif event_type == "response.refusal.done" and not chunks: emit(event.get("refusal")) elif event_type in {"response.completed", "response.done"}: response_data = event.get("response") if isinstance(response_data, Mapping): response_status = str(response_data.get("status") or "completed") if response_status != "completed": raise DirectChatError( f"Responses 流式响应状态异常：{response_status}" ) canonical = _answer_from_responses(response_data) if canonical and canonical != "".join(chunks).strip(): emit(canonical, replace=True) finished = True break else: declared_type = str(event.get("type") or "") if sse_name and declared_type and sse_name != declared_type: raise DirectChatError("Anthropic SSE 事件名称与数据类型不一致") if event_type == "error": raise DirectChatError( _error_message(event, "Anthropic 流式响应返回错误") ) if event_type == "message_start": anthropic_started = True elif event_type == "content_block_start": if not anthropic_started: raise DirectChatError("Anthropic 内容块早于 message_start") index = int(event.get("index", -1)) if index < 0 or index in anthropic_open_blocks: raise DirectChatError("Anthropic 内容块索引无效") anthropic_open_blocks.add(index) block = event.get("content_block") if isinstance(block, Mapping) and block.get("type") == "text": emit(block.get("text")) elif event_type == "content_block_delta": index = int(event.get("index", -1)) if index not in anthropic_open_blocks: raise DirectChatError("Anthropic 内容增量没有对应的开始事件") delta = event.get("delta") if isinstance(delta, Mapping) and delta.get("type") == "text_delta": emit(delta.get("text")) elif event_type == "content_block_stop": index = int(event.get("index", -1)) if index not in anthropic_open_blocks: raise DirectChatError("Anthropic 内容块结束事件无效") anthropic_open_blocks.remove(index) elif event_type == "message_delta": delta = event.get("delta") stop_reason = ( str(delta.get("stop_reason") or "") if isinstance(delta, Mapping) else "" ) anthropic_message_delta = True anthropic_stop_reason = stop_reason if stop_reason in {"tool_use", "max_tokens"}: raise DirectChatError( f"Anthropic 普通对话未完整结束：{stop_reason}" ) elif event_type == "message_stop": if ( not anthropic_started or anthropic_open_blocks or not anthropic_message_delta or not anthropic_stop_reason ): raise DirectChatError("Anthropic 流式响应结束序列不完整") if anthropic_stop_reason not in { "end_turn", "stop_sequence", "refusal", }: raise DirectChatError( f"Anthropic 普通对话结束原因异常：{anthropic_stop_reason}" ) finished = True break except (OSError, TimeoutError, ValueError) as exc: cancel.raise_if_cancelled() raise DirectChatError(f"读取后台自有模型流失败：{exc}") from exc cancel.raise_if_cancelled() if not finished: raise DirectChatError("后台自有模型流式响应未正常结束") if prepared.protocol == "dify" and workflow_started and not workflow_finished: raise DirectChatError("Dify Chatflow 流式响应缺少 workflow_finished") if prepared.protocol == "chat_completions" and completion_finish_reason != "stop": raise DirectChatError( "Chat Completions 普通对话未完整结束：" f"{completion_finish_reason or '缺少 finish_reason'}" ) text = "".join(chunks).strip() if not text: raise DirectChatError("后台自有模型流式响应没有有效文本") return DirectChatResult( text=text, model=prepared.model, protocol=prepared.protocol, conversation_id=next_conversation_id, ) finally: cancel.detach(response) try: response.close() except (AttributeError, OSError, ValueError): pass def direct_chat( message: str, *, history: Sequence[Mapping[str, object]] | None = None, user_id: str = "wechat-rpa-chat", conversation_id: str = "", manager: GrokBuildManager | None = None, ) -> DirectChatResult: """Call the configured custom model without starting Grok Build.""" prepared = _prepare_chat( message, history=history, user_id=user_id, conversation_id=conversation_id, manager=manager, streaming=False, ) if prepared.identity: return DirectChatResult( text=prepared.identity, model=prepared.model, protocol=prepared.protocol, conversation_id="", ) data = _post_json( prepared.url, headers=prepared.headers, payload=prepared.payload, timeout=prepared.timeout, ) return _result_from_json(prepared, data)