feat: 个人微信-对话历史持久化+平台内查阅页; 修复第二轮不回复(过滤/字段/重试); 多轮上下文

This commit is contained in:
WorkBuddy committed 2026-07-10 07:01:39 +08:00
1 parent 2eadd83304
commit 62f3986e04
5 files changed
+251 -28

No files matched your search

+150 -28
View File
@@ -99,13 +99,44 @@ def _wechat_mask(row) -> dict:
# ============================================================
# 七虎AI 大脑(复用 channels 的真实 LLM 调用)
# ============================================================
def _qihua_reply(text: str) -> str:
def _recent_history(user_id: str, limit: int = 6) -> str:
"""拼接最近若干轮对话,作为多轮上下文。"""
db = None
try:
db = SessionLocal()
rows = _execute_raw(db, text(
"SELECT role, content FROM im_bot_chat_log "
"WHERE user_id=:uid ORDER BY id DESC LIMIT :lim"
), {"uid": user_id, "lim": limit})
pairs = []
for r in reversed(rows):
dd = _row_to_dict(r)
role = "用户" if dd["role"] == "user" else "七虎AI"
pairs.append(f"{role}:{dd['content']}")
return "\n".join(pairs)
except Exception:
return ""
finally:
if db:
db.close()
def _qihua_reply(text: str, user_id: str = None) -> str:
"""调用七虎AI 生成回复。自行创建 DB 会话,避免上层线程无可用 session。"""
prompt = text
if user_id:
hist = _recent_history(user_id)
if hist:
prompt = (
"以下是你与该用户最近的对话历史,请结合上下文作答:\n"
f"{hist}\n\n"
f"用户当前问题:{text}"
)
db = None
try:
from app.api.channels import _call_real_llm
db = SessionLocal()
return _call_real_llm(db, None, text, "个人微信")
return _call_real_llm(db, None, prompt, "个人微信")
except Exception as e:
return f"(AI 回复生成失败:{e})"
finally:
@@ -134,23 +165,29 @@ def _wechat_get_updates(baseurl: str, bot_token: str, cursor: str):
def _wechat_send_message(baseurl: str, bot_token: str, to_user_id: str, context_token: str, text: str) -> bool:
"""发送文本回复(必须带 context_token 关联到正确的对话窗口)"""
try:
resp = requests.post(
baseurl.rstrip("/") + "/ilink/bot/sendmessage",
headers=_wechat_headers(bot_token),
json={"msg": {
"to_user_id": to_user_id,
"message_type": 2,
"message_state": 2,
"context_token": context_token,
"item_list": [{"type": 1, "text_item": {"text": text}}],
}},
timeout=20,
)
return resp.ok
except Exception:
return False
"""发送文本回复(必须带 context_token 关联到正确的对话窗口)。失败重试一次。"""
payload = {"msg": {
"to_user_id": to_user_id,
"message_type": 2,
"message_state": 2,
"context_token": context_token,
"item_list": [{"type": 1, "text_item": {"text": text}}],
}}
last = False
for _ in range(2):
try:
resp = requests.post(
baseurl.rstrip("/") + "/ilink/bot/sendmessage",
headers=_wechat_headers(bot_token),
json=payload,
timeout=20,
)
last = resp.ok
if last:
return True
except Exception:
last = False
return last
def _wechat_extract_text(msg: dict) -> str:
@@ -158,9 +195,55 @@ def _wechat_extract_text(msg: dict) -> str:
for it in (msg.get("item_list") or []):
if it.get("type") == 1 and it.get("text_item", {}).get("text"):
parts.append(it["text_item"]["text"])
if not parts:
for k in ("text", "content", "message", "msg"):
v = msg.get(k)
if isinstance(v, str) and v.strip():
parts.append(v)
return "\n".join(parts)
def _wechat_sender_id(m: dict) -> str:
"""从入站消息中提取发送者 id(兼容多套字段名)。"""
for k in ("from_user_id", "from_user_name", "from_username", "user_id",
"sender_id", "wxid", "from_wxid", "username"):
v = m.get(k)
if v:
return str(v)
return ""
def _wechat_context_token(m: dict) -> str:
"""提取 context_token(回复必须回带,否则不关联到对话窗口)。"""
for k in ("context_token", "ctx_token", "conversation_id", "conv_id",
"msg_id", "context_id", "session_id"):
v = m.get(k)
if v:
return str(v)
return ""
def _log_chat(user_id, platform, from_user, context_token, role, content):
"""持久化 IM Bot 对话历史,便于平台内查阅。"""
db = None
try:
db = SessionLocal()
_execute_raw(db, text(
"INSERT INTO im_bot_chat_log "
"(user_id, platform, from_user, context_token, role, content, created_at) "
"VALUES (:uid, :platform, :fu, :ctx, :role, :content, :created)"
), {
"uid": user_id, "platform": platform, "fu": from_user,
"ctx": context_token, "role": role, "content": content,
"created": datetime.now(),
})
except Exception:
pass
finally:
if db:
db.close()
# ============================================================
# 后台接收线程(每用户独立)
# ============================================================
@@ -183,19 +266,35 @@ def _wechat_receiver_loop(account_id: int):
bot_token = d["bot_token"]
baseurl = d.get("baseurl") or ILINK_BASE
msgs, cursor = _wechat_get_updates(baseurl, bot_token, cursor)
owner_uid = d.get("user_id")
for m in msgs:
if m.get("message_type") not in (1, None):
continue # 只处理用户消息(2=bot 自己发出的)
if m.get("message_type") == 2:
continue # 仅跳过 bot 自己发出的回声,其余(含后续追问)均尝试处理
content = _wechat_extract_text(m)
if not content:
continue
reply = _qihua_reply(content)
_wechat_send_message(
baseurl, bot_token,
m.get("from_user_id", ""),
m.get("context_token", ""),
reply,
)
sender = _wechat_sender_id(m)
ctx = _wechat_context_token(m)
# 诊断:记录原始入站消息,定位字段名 / 第二轮不回复
try:
with open("/tmp/wechat_inbound.log", "a", encoding="utf-8") as _f:
_f.write(datetime.now().isoformat() + " sender=" + repr(sender)
+ " ctx=" + repr(ctx) + " raw=" + json.dumps(m, ensure_ascii=False) + "\n")
except Exception:
pass
# 先生成回复(此时历史尚未包含本条,避免重复)
reply = _qihua_reply(content, owner_uid)
ok = _wechat_send_message(baseurl, bot_token, sender, ctx, reply)
# 持久化对话历史
if owner_uid:
_log_chat(owner_uid, "personal_wechat", sender, ctx, "user", content)
_log_chat(owner_uid, "personal_wechat", sender, ctx, "assistant", reply)
try:
with open("/tmp/wechat_send.log", "a", encoding="utf-8") as _f:
_f.write(datetime.now().isoformat() + " ok=" + str(ok)
+ " sender=" + repr(sender) + " ctx=" + repr(ctx) + "\n")
except Exception:
pass
time.sleep(1)
except Exception:
time.sleep(5)
@@ -354,6 +453,29 @@ async def wechat_disconnect(user: User = Depends(get_current_user)):
# ============================================================
# 启动恢复:为所有已连接用户账号拉起接收线程
# ============================================================
@router.get("/history")
async def wechat_history(
user: User = Depends(get_current_user),
limit: int = Query(200),
):
"""查询当前用户通过个人微信与七虎AI 的对话历史"""
db = SessionLocal()
try:
rows = _execute_raw(
db,
text("SELECT id, role, content, platform, from_user, created_at "
"FROM im_bot_chat_log WHERE user_id=:uid ORDER BY id DESC LIMIT :lim"),
{"uid": user.id, "lim": min(int(limit), 500)},
)
items = [_row_to_dict(r) for r in rows]
items.reverse()
return {"success": True, "data": {"messages": items, "total": len(items)}}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
finally:
db.close()
def start_all_wechat_receivers():
"""在应用启动时调用,恢复此前已连接用户的个人微信接收线程。"""
db = SessionLocal()