397 lines
16 KiB
Python
397 lines
16 KiB
Python
"""
|
||
AI 聊天模块
|
||
===========
|
||
负责与 AI API 通信，支持文本模式、视觉模式，以及 MCP 工具增强回复。
|
||
使用标准 requests 库，无需安装 openai SDK。
|
||
"""
|
||
|
||
import requests
|
||
import base64
|
||
import json
|
||
import re
|
||
from urllib.parse import urlparse
|
||
import ai_config
|
||
|
||
# ⚠ 本模块所有配置一律通过 ai_config.XXX 动态读取（而非 from-import 快照），
|
||
# 这样在 GUI「AI 高级配置」中修改保存后，下一次请求立即生效，无需重启。
|
||
|
||
|
||
def _completions_url() -> str:
|
||
"""
|
||
组装实际请求地址：
|
||
- AI_API_BASE 形如 .../v1/chat-messages（/v1/ 后已有路径）→ 原样使用，不再拼 /chat/completions
|
||
- AI_API_BASE 形如 https://api.deepseek.com 或 .../v1 → 追加 /chat/completions
|
||
"""
|
||
base = (ai_config.AI_API_BASE or "").rstrip("/")
|
||
path = urlparse(base).path or ""
|
||
if re.match(r"^/v1/.+", path):
|
||
return base
|
||
return f"{base}/chat/completions"
|
||
|
||
|
||
def _is_dify_endpoint() -> bool:
|
||
"""AI_API_BASE 指向 Dify 的 chat-messages / completion-messages 时走 Dify 协议。"""
|
||
path = (urlparse((ai_config.AI_API_BASE or "").rstrip("/")).path or "").lower()
|
||
return "chat-messages" in path or "completion-messages" in path
|
||
|
||
|
||
def _system_prompt() -> str:
|
||
"""
|
||
动态构建系统提示词：基础人设（用当前昵称/医院名实时渲染）
|
||
+ 按开关追加的可选模块（对骂模式 / MCP 工具说明）。
|
||
"""
|
||
prompt = ai_config.build_system_prompt()
|
||
if getattr(ai_config, 'AI_COUNTER_INSULT_ENABLED', False):
|
||
prompt += "\n" + getattr(ai_config, 'AI_COUNTER_INSULT_PROMPT', '')
|
||
if getattr(ai_config, 'AI_MCP_ENABLED', False):
|
||
prompt += "\n" + getattr(ai_config, 'AI_MCP_PROMPT', '')
|
||
return prompt
|
||
|
||
|
||
def _history_messages(history: list) -> list:
|
||
"""
|
||
将会话历史裁剪为最近 N 轮，作为多轮上下文消息插入到请求中。
|
||
history 为会话档案中的消息列表（可能带 ts 等额外字段，发 API 前只保留 role/content）。
|
||
运行时动态读取 ai_config（GUI 可在启动时修改开关）。
|
||
"""
|
||
if not getattr(ai_config, 'AI_CONTEXT_ENABLED', False) or not history:
|
||
return []
|
||
max_rounds = getattr(ai_config, 'AI_CONTEXT_MAX_ROUNDS', 5)
|
||
return [
|
||
{"role": m["role"], "content": m["content"]}
|
||
for m in history[-max_rounds * 2:]
|
||
]
|
||
|
||
|
||
def _headers():
|
||
key = (ai_config.AI_API_KEY or "").strip()
|
||
return {
|
||
"Authorization": f"Bearer {key}",
|
||
"Content-Type": "application/json",
|
||
"Accept": "application/json, text/plain, */*",
|
||
"Accept-Language": "zh-CN,zh;q=0.9,en;q=0.8",
|
||
"User-Agent": (
|
||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
|
||
"AppleWebKit/537.36 (KHTML, like Gecko) "
|
||
"Chrome/124.0.0.0 Safari/537.36"
|
||
),
|
||
"Connection": "keep-alive",
|
||
}
|
||
|
||
|
||
def _call_dify(query: str, user: str = "wechat-rpa", conversation_id: str = "") -> str:
|
||
"""
|
||
调用 Dify /v1/chat-messages（blocking）。
|
||
鉴权用应用 API Key（通常以 app- 开头），与 OpenAI 兼容接口不同。
|
||
"""
|
||
payload = {
|
||
"inputs": {},
|
||
"query": query,
|
||
"response_mode": "blocking",
|
||
"user": user or "wechat-rpa",
|
||
}
|
||
if conversation_id:
|
||
payload["conversation_id"] = conversation_id
|
||
|
||
url = _completions_url()
|
||
resp = requests.post(
|
||
url, headers=_headers(), json=payload, timeout=ai_config.AI_TIMEOUT
|
||
)
|
||
if resp.status_code == 401:
|
||
detail = ""
|
||
try:
|
||
detail = resp.json().get("message") or resp.text[:200]
|
||
except Exception:
|
||
detail = resp.text[:200]
|
||
raise RuntimeError(
|
||
f"Dify 鉴权失败（401）: {detail}。"
|
||
f"请到 Dify 应用 →「访问 API」复制 API Key（一般以 app- 开头）填到 AI_API_KEY，"
|
||
f"当前 Key 看起来不像有效应用密钥。"
|
||
)
|
||
if not resp.ok:
|
||
detail = (resp.text or "")[:400]
|
||
raise RuntimeError(f"Dify 请求失败 {resp.status_code}: {detail}")
|
||
data = resp.json()
|
||
answer = data.get("answer")
|
||
if answer is None:
|
||
# 少数部署可能包一层 data
|
||
answer = (data.get("data") or {}).get("answer")
|
||
if not answer:
|
||
raise RuntimeError(f"Dify 未返回 answer: {json.dumps(data, ensure_ascii=False)[:300]}")
|
||
return str(answer)
|
||
|
||
|
||
def _dify_query_from_chat(chat_text: str, history: list = None) -> str:
|
||
"""
|
||
把本地会话档案拼进 Dify query，并注入事实铁律 + 挂号医院铁律。
|
||
"""
|
||
try:
|
||
hosp = getattr(ai_config, "AI_HOSPITAL_NAME", None) or "甄养堂互联网医院"
|
||
except Exception:
|
||
hosp = "甄养堂互联网医院"
|
||
rules = (
|
||
"【事实铁律｜必须遵守】\n"
|
||
"1. 严禁编造快递单号、物流状态、签收时间、订单号、库存等业务数据。\n"
|
||
"2. 查快递/订单时：没有真实系统查询结果，只能追问单号/姓名/手机，或说去后台核实后回；"
|
||
"不准说「查到了」「已签收」并随口编一个单号。\n"
|
||
"3. 历史里出现的可疑示例单号（如 SF1234567890）一律视为无效，不要继续沿用。\n"
|
||
f"4. 凡建议就医/挂号/面诊，或回复里提到医院，【只能】写「{hosp}」，"
|
||
"严禁「正规医院」「当地医院」「三甲医院」「内分泌科」等说法（急救拨120除外）。"
|
||
f"本院是糖尿病中医专科医院，没有内分泌科。\n"
|
||
"5. 客户只是咨询（如血糖不稳定怎么办）时：先给专业建议，"
|
||
"可轻提一句需要可来本院挂号；【禁止】直接说已帮您预约。\n"
|
||
f"6. 仅当客户明确说要挂号/预约/面诊/帮我约 时，才说「已帮您预约了，稍后预约上了再联系您」，"
|
||
f"医院是{hosp}。客户说不需要/挂啥号时绝不能预约。\n"
|
||
"7. 只回复客户最后一条需要处理的问题。语气像干了十几年的老客服：口语、沉稳、"
|
||
"不急不躁，一次只说一件事，最多顺带问一个问题，不要复述对方原话，不要一次抛一大段方案。\n"
|
||
)
|
||
hist = _history_messages(history)
|
||
if not hist:
|
||
body = f"以下是最近的聊天记录：\n\n{chat_text}\n\n请生成回复："
|
||
else:
|
||
lines = []
|
||
for m in hist:
|
||
role = "客户" if m.get("role") == "user" else "客服"
|
||
lines.append(f"{role}: {m.get('content', '')}")
|
||
body = (
|
||
"【近期对话｜仅供参考，其中客服所述业务数据可能不实】\n"
|
||
+ "\n".join(lines)
|
||
+ f"\n\n【最新消息】\n{chat_text}\n\n请生成回复："
|
||
)
|
||
return rules + "\n" + body
|
||
|
||
|
||
def _strip_thinking(text: str) -> str:
|
||
"""
|
||
去除 Qwen3 / DeepSeek-R1 等模型输出的 <think>...</think> 推理块，
|
||
只保留最终回复内容。
|
||
"""
|
||
if not text:
|
||
return ""
|
||
cleaned = re.sub(r'<think>.*?</think>', '', text, flags=re.DOTALL)
|
||
return cleaned.strip()
|
||
|
||
|
||
# 上游模型（尤其是 Dify 应用侧自带的固定人设/挡话话术）有时不听系统提示词，
|
||
# 这里做最后一道兜底净化：去掉暴露机器身份的自称，以及被禁止的轻浮语气词。
|
||
_SELF_ID_RE = re.compile(
|
||
r"[，,]?\s*(我是|作为)(您|你)的?(专属)?"
|
||
r"(健康(顾问|助理|管家|专家)|智能(客服|助手)|AI(客服|助手)?|人工智能|"
|
||
r"虚拟(客服|助手)|机器人客服|机器人|自动回复(助手)?|系统助手|助手)"
|
||
r"[，,。！]?"
|
||
)
|
||
_LAUGH_RE = re.compile(r"哈{2,}[，,]?|哈哈[，,]?")
|
||
|
||
|
||
def _humanize(text: str) -> str:
|
||
"""去除自称机器人/助手的句子，以及禁止的"哈哈"语气词，收拾多余标点。"""
|
||
if not text:
|
||
return text
|
||
# 用逗号占位替换，避免前后半句直接粘连；随后统一收拢多余标点
|
||
cleaned = _SELF_ID_RE.sub("，", text)
|
||
cleaned = _LAUGH_RE.sub("", cleaned)
|
||
cleaned = re.sub(r"[，,]{2,}", "，", cleaned)
|
||
cleaned = re.sub(r"[，,]\s*([。！？])", r"\1", cleaned)
|
||
cleaned = re.sub(r"([。！？])\s*[，,]+", r"\1", cleaned)
|
||
cleaned = re.sub(r"[。！？]{2,}", "。", cleaned)
|
||
cleaned = re.sub(r"^[，,、\s]+", "", cleaned)
|
||
cleaned = re.sub(r"\s{2,}", " ", cleaned)
|
||
return cleaned.strip()
|
||
|
||
|
||
def _chat_completion(messages: list, tools: list = None) -> dict:
|
||
"""
|
||
调用 chat/completions，返回 message 对象（含 content / tool_calls）。
|
||
"""
|
||
if _is_dify_endpoint():
|
||
# 兜底：误入 OpenAI 路径时改走 Dify（避免 400 Arg user must be provided）
|
||
last_user = ""
|
||
for m in reversed(messages):
|
||
if m.get("role") == "user":
|
||
c = m.get("content")
|
||
last_user = c if isinstance(c, str) else str(c)
|
||
break
|
||
answer = _call_dify(last_user or "请回复")
|
||
return {"role": "assistant", "content": answer}
|
||
|
||
payload = {
|
||
"model": ai_config.AI_MODEL,
|
||
"messages": messages,
|
||
"max_tokens": ai_config.AI_MAX_TOKENS,
|
||
"temperature": ai_config.AI_TEMPERATURE,
|
||
}
|
||
if tools:
|
||
payload["tools"] = tools
|
||
payload["tool_choice"] = "auto"
|
||
url = _completions_url()
|
||
resp = requests.post(
|
||
url, headers=_headers(), json=payload, timeout=ai_config.AI_TIMEOUT
|
||
)
|
||
if not resp.ok:
|
||
detail = (resp.text or "")[:400]
|
||
raise RuntimeError(f"AI 请求失败 {resp.status_code}: {detail}")
|
||
return resp.json()["choices"][0]["message"]
|
||
|
||
|
||
def _user_turn(chat_text: str) -> dict:
|
||
return {
|
||
"role": "user",
|
||
"content": f"以下是最近的聊天记录：\n\n{chat_text}\n\n请生成回复：",
|
||
}
|
||
|
||
|
||
def call_ai_text(chat_text: str, history: list = None) -> str:
|
||
"""
|
||
文本模式：将聊天记录文字发给文本 AI，返回回复。
|
||
若开启 AI_MCP_ENABLED，会连接外部 MCP Server，让模型按需调用工具后再回复。
|
||
若 AI_API_BASE 为 Dify chat-messages，走 Dify 协议（不支持 OpenAI tools）。
|
||
"""
|
||
if _is_dify_endpoint():
|
||
print(" [AI] 检测到 Dify 接口，使用 chat-messages 协议")
|
||
return _humanize(_strip_thinking(_call_dify(_dify_query_from_chat(chat_text, history))))
|
||
|
||
if getattr(ai_config, "AI_MCP_ENABLED", False):
|
||
try:
|
||
from mcp_bridge import run_coro
|
||
return _humanize(run_coro(_call_ai_text_with_mcp(chat_text, history)))
|
||
except Exception as e:
|
||
print(f" [MCP] ⚠ 工具增强失败，回退普通回复: {e}")
|
||
|
||
messages = [{"role": "system", "content": _system_prompt()}]
|
||
messages += _history_messages(history)
|
||
messages.append(_user_turn(chat_text))
|
||
msg = _chat_completion(messages)
|
||
return _humanize(_strip_thinking(msg.get("content") or ""))
|
||
|
||
|
||
async def _call_ai_text_with_mcp(chat_text: str, history: list = None) -> str:
|
||
"""带 MCP 工具调用的多轮回复。"""
|
||
from mcp_bridge import McpHub
|
||
|
||
async with McpHub() as hub:
|
||
tools = hub.openai_tools()
|
||
if not tools:
|
||
print(" [MCP] 未加载到任何工具，使用普通文本回复")
|
||
messages = [{"role": "system", "content": _system_prompt()}]
|
||
messages += _history_messages(history)
|
||
messages.append(_user_turn(chat_text))
|
||
msg = _chat_completion(messages)
|
||
return _strip_thinking(msg.get("content") or "")
|
||
|
||
print(f" [MCP] 已加载 {hub.tool_count} 个工具"
|
||
f"（服务器: {', '.join(hub.server_names) or '-'}）")
|
||
|
||
messages = [{"role": "system", "content": _system_prompt()}]
|
||
messages += _history_messages(history)
|
||
messages.append(_user_turn(chat_text))
|
||
|
||
max_rounds = int(getattr(ai_config, "AI_MCP_MAX_ROUNDS", 5) or 5)
|
||
for i in range(max_rounds):
|
||
msg = _chat_completion(messages, tools=tools)
|
||
tool_calls = msg.get("tool_calls") or []
|
||
content = msg.get("content")
|
||
|
||
if not tool_calls:
|
||
return _strip_thinking(content or "")
|
||
|
||
# 保留 assistant 消息（含 tool_calls）
|
||
messages.append({
|
||
"role": "assistant",
|
||
"content": content,
|
||
"tool_calls": tool_calls,
|
||
})
|
||
|
||
for tc in tool_calls:
|
||
fn = tc.get("function") or {}
|
||
name = fn.get("name") or ""
|
||
raw_args = fn.get("arguments") or "{}"
|
||
try:
|
||
args = json.loads(raw_args) if isinstance(raw_args, str) else (raw_args or {})
|
||
except json.JSONDecodeError:
|
||
args = {}
|
||
print(f" [MCP] 调用工具 {name} args={json.dumps(args, ensure_ascii=False)[:120]}")
|
||
result = await hub.call_tool(name, args)
|
||
preview = (result or "").replace("\n", " ")[:120]
|
||
print(f" [MCP] 工具结果: {preview}{'…' if len(result or '') > 120 else ''}")
|
||
messages.append({
|
||
"role": "tool",
|
||
"tool_call_id": tc.get("id") or name,
|
||
"content": result or "",
|
||
})
|
||
|
||
# 超过轮数：强制再要一次纯文本回复
|
||
print(" [MCP] 已达最大工具轮数，请求最终回复")
|
||
msg = _chat_completion(messages)
|
||
return _strip_thinking(msg.get("content") or "")
|
||
|
||
|
||
def call_ai_vision(image_bytes: bytes, history: list = None) -> str:
|
||
"""
|
||
视觉模式：将聊天区域截图发给多模态 AI，让 AI 直接阅读并回复。
|
||
image_bytes 为 PNG 图片的 bytes。
|
||
history 为该会话的历史上下文（多轮记忆），可为 None。
|
||
"""
|
||
b64 = base64.b64encode(image_bytes).decode("utf-8")
|
||
simple_headers = {
|
||
"Authorization": f"Bearer {ai_config.AI_API_KEY}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
messages = [{"role": "system", "content": _system_prompt()}]
|
||
messages += _history_messages(history)
|
||
payload = {
|
||
"model": ai_config.AI_MODEL,
|
||
"messages": messages + [
|
||
{"role": "user", "content": [
|
||
{
|
||
"type": "text",
|
||
"text": (
|
||
"这是一个聊天对话窗口的截图。"
|
||
"左边的灰色气泡是对方（客户）发的消息，右边的蓝色气泡是我方之前的回复。"
|
||
"请只关注对方（客户）发的最后一条消息，针对那条消息直接回复。"
|
||
"只输出回复内容，不要描述图片，不要解释，不要加引号。"
|
||
),
|
||
},
|
||
{
|
||
"type": "image_url",
|
||
"image_url": {"url": f"data:image/png;base64,{b64}"},
|
||
},
|
||
]},
|
||
],
|
||
"max_tokens": ai_config.AI_MAX_TOKENS,
|
||
"temperature": ai_config.AI_TEMPERATURE,
|
||
}
|
||
url = _completions_url()
|
||
resp = requests.post(url, headers=simple_headers, json=payload, timeout=ai_config.AI_TIMEOUT)
|
||
resp.raise_for_status()
|
||
content = resp.json()["choices"][0]["message"]["content"]
|
||
return _humanize(_strip_thinking(content))
|
||
|
||
|
||
def get_ai_reply(chat_text: str = None, image_bytes: bytes = None, history: list = None) -> str:
|
||
"""
|
||
统一入口：根据 AI_USE_VISION 配置自动选择模式。
|
||
history 为该会话的历史上下文（多轮记忆），可为 None。
|
||
返回 AI 生成的回复文本。
|
||
"""
|
||
try:
|
||
if ai_config.AI_USE_VISION and image_bytes:
|
||
print(" [AI] 使用视觉模式分析聊天截图...")
|
||
return call_ai_vision(image_bytes, history=history)
|
||
elif chat_text:
|
||
if getattr(ai_config, "AI_MCP_ENABLED", False):
|
||
print(" [AI] 文本模式 + MCP 工具增强...")
|
||
else:
|
||
print(" [AI] 使用文本模式分析聊天记录...")
|
||
return call_ai_text(chat_text, history=history)
|
||
else:
|
||
return ""
|
||
except requests.exceptions.Timeout:
|
||
print(" [AI] ⚠ API 请求超时")
|
||
return ""
|
||
except requests.exceptions.RequestException as e:
|
||
print(f" [AI] ⚠ API 请求失败: {e}")
|
||
return ""
|
||
except (KeyError, IndexError, json.JSONDecodeError) as e:
|
||
print(f" [AI] ⚠ 解析响应失败: {e}")
|
||
return ""
|