921 lines
36 KiB
Python
921 lines
36 KiB
Python
# -*- 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)
|