81115dc23d
微信自动回复机器人，基于截图+OCR识别消息，支持关键词规则和 AI（OpenAI/DeepSeek/Dify）自动回复。 技术栈：PySide6 + Flask + Vue3 + RapidOCR + SQLite 注：OCR大模型文件（.onnx / .pdiparams）不纳入版本控制，需单独下载。 🤖 Generated with [Qoder][https://qoder.com]
82 lines
4.3 KiB
Python
82 lines
4.3 KiB
Python
import requests
|
||
|
||
from app.infrastructure.service.backend.config import AI_PROVIDER, DEEPSEEK_API_BASE, DEEPSEEK_API_KEY, DEEPSEEK_MODEL, DIFY_API_BASE, DIFY_API_KEY, DIFY_USER, OPENAI_API_BASE, OPENAI_API_KEY, OPENAI_MODEL
|
||
from app.infrastructure.service.logging.log_service import log_event, new_trace_id
|
||
|
||
|
||
def do_openai_like(url, headers, payload):
|
||
try:
|
||
resp = requests.post(url, headers=headers, json=payload, timeout=60)
|
||
data = resp.json() if resp.text else {}
|
||
if resp.status_code >= 400:
|
||
return f"抱歉，AI 服务请求失败（{resp.status_code}）"
|
||
content = (((data or {}).get("choices") or [{}])[0].get("message") or {}).get("content", "")
|
||
return content.strip() if content else "抱歉，AI 暂时没有合理的回复。"
|
||
except Exception:
|
||
return "抱歉，AI 服务暂时不可用，请稍后再试。"
|
||
|
||
|
||
def do_dify(url, headers, payload):
|
||
try:
|
||
resp = requests.post(url, headers=headers, json=payload, timeout=60)
|
||
text = resp.text or ""
|
||
if "data:" in text:
|
||
answer = ""
|
||
for line in text.splitlines():
|
||
line = line.strip()
|
||
if not line.startswith("data:"):
|
||
continue
|
||
chunk = line[5:].strip()
|
||
if not chunk or chunk == "[DONE]":
|
||
continue
|
||
try:
|
||
j = requests.models.complexjson.loads(chunk)
|
||
except Exception:
|
||
continue
|
||
if "answer" in j:
|
||
answer += j["answer"]
|
||
if answer.strip():
|
||
return answer.strip()
|
||
data = resp.json() if resp.text else {}
|
||
if resp.status_code >= 400:
|
||
return f"抱歉，Dify 服务请求失败（{resp.status_code}）"
|
||
return (data.get("answer") or "抱歉，Dify 暂时没有合理的回复。").strip()
|
||
except Exception:
|
||
return "抱歉，Dify 服务暂时不可用，请稍后再试。"
|
||
|
||
|
||
def call_ai(prompt, user_id=""):
|
||
trace_id = new_trace_id("ai")
|
||
provider = AI_PROVIDER
|
||
log_event("INFO", "ai", "ai.request", trace_id, "request", "ok", "发起AI请求", extra={"provider": provider, "user_id": user_id or "", "prompt_len": len(prompt or "")})
|
||
if provider == "mock":
|
||
result = "【自动回复】你刚才说了：" + (prompt or "")[:100]
|
||
log_event("INFO", "ai", "ai.response", trace_id, "response", "ok", "AI回复完成", extra={"provider": provider, "reply_len": len(result)})
|
||
return result
|
||
if provider == "openai":
|
||
result = do_openai_like(
|
||
OPENAI_API_BASE.rstrip("/") + "/chat/completions",
|
||
{"Content-Type": "application/json", "Authorization": f"Bearer {OPENAI_API_KEY}"},
|
||
{"model": OPENAI_MODEL, "messages": [{"role": "system", "content": "你是一个专业的微信私域运营助手，用简洁自然的中文回复用户。"}, {"role": "user", "content": prompt}], "temperature": 0.7, "user": user_id or None},
|
||
)
|
||
elif provider == "deepseek":
|
||
result = do_openai_like(
|
||
DEEPSEEK_API_BASE.rstrip("/") + "/chat/completions",
|
||
{"Content-Type": "application/json", "Authorization": f"Bearer {DEEPSEEK_API_KEY}"},
|
||
{"model": DEEPSEEK_MODEL, "messages": [{"role": "system", "content": "你是一个简洁高效的微信助手。回复要求：一句话，不超过50字。"}, {"role": "user", "content": prompt}], "temperature": 0.7, "max_tokens": 100, "user": user_id or None},
|
||
)
|
||
elif provider == "dify":
|
||
result = do_dify(
|
||
DIFY_API_BASE.rstrip("/") + "/chat-messages",
|
||
{"Content-Type": "application/json", "Authorization": f"Bearer {DIFY_API_KEY}"},
|
||
{"inputs": {}, "query": prompt, "response_mode": "streaming", "user": user_id or DIFY_USER, "conversation_id": ""},
|
||
)
|
||
else:
|
||
result = "AI_PROVIDER 未配置正确，请检查环境变量。"
|
||
|
||
if result.startswith("抱歉") or "未配置正确" in result:
|
||
log_event("WARNING", "ai", "ai.response", trace_id, "response", "failed", "AI回复异常或降级", reason="provider_error", extra={"provider": provider, "reply": result[:120]})
|
||
else:
|
||
log_event("INFO", "ai", "ai.response", trace_id, "response", "ok", "AI回复完成", extra={"provider": provider, "reply_len": len(result)})
|
||
return result
|