Files
BossAi/bossai.py
T
selfreleaseandDevin 4de36547cd
CI / 后端测试 (push) Has been cancelled
CI / 前端构建 (push) Has been cancelled
feat: BossAi v2.0 UI/UX 重构 + 看板断点续接 + 多轮上下文
- 新增 web2/ 前端:任务卡看板、风险雷达、决策日志、我的团队、快捷指令
- ChatPanel 增强:风险卡置顶、案卡接入、起诉状向导、语音输入
- 事项看板:点击查看详情弹窗、断点续接(恢复历史对话+matterId)
- 保密模式恢复:密码验证后重建内存 session,保留多轮上下文
- 多轮对话:ask/ask-stream/ask-analyze 注入 session 历史,AI 有上下文
- 追问优化:紧扣用户问题,不再偏离主题(社保问社保,工伤问工伤)
- 对话区可滚动,输入区固定可见
- 状态机放宽:允许从任意非终态转到 pending_expert
- session 存活检查端点,后端重启后自动重建 session
- DB migration:风险推送表、案卡表

Generated with [Devin](https://devin.ai)

Co-Authored-By: Devin <158243242+devin-ai-integration[bot]@users.noreply.github.com>
2026-08-31 08:38:33 +08:00

3333 lines
144 KiB
Python

"""BossAi MVP core: privacy sessions and structured risk guidance.
This dependency-free module is intentionally small so it can run on an edge AI box.
Model inference is represented by a pluggable responder; production adapters can
connect a local model without changing session/privacy behavior.
"""
from __future__ import annotations
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
import hashlib
import secrets
from typing import Callable, Dict, List, Optional
import re
from pathlib import Path
import os
import json
import base64
from urllib.request import Request, urlopen
import math
from collections import Counter
import psycopg2
try:
from cryptography.fernet import Fernet
except ImportError:
Fernet = None # 加密功能可选,缺失时保密模式降级为明文
@dataclass
class RiskCard:
conclusion: str
level: str
actions: List[str]
evidence: List[str] = field(default_factory=list)
escalation: bool = False
@dataclass
class PrivacySession:
session_id: str
mode: str
expires_at: datetime
key_digest: str
messages: List[dict] = field(default_factory=list)
destroyed: bool = False
temp_files: List[str] = field(default_factory=list) # 临时文件路径,销毁时清理
destroy_log: List[dict] = field(default_factory=list) # 销毁操作日志(不含正文)
class SessionManager:
"""In-memory session store. Production storage must encrypt payloads at rest."""
MODES = {"normal", "confidential", "ephemeral"}
TTL = {"normal": None, "confidential": timedelta(hours=24), "ephemeral": timedelta(minutes=5)}
def __init__(self, clock: Callable[[], datetime] | None = None):
self.clock = clock or (lambda: datetime.now(timezone.utc))
self.sessions: Dict[str, PrivacySession] = {}
self._destroy_stats = {"success": 0, "failed": 0, "retried": 0}
def create(self, mode: str = "normal", ttl: Optional[timedelta] = None) -> PrivacySession:
if mode not in self.MODES:
raise ValueError(f"unsupported privacy mode: {mode}")
now = self.clock()
expiry = now + (ttl or self.TTL[mode] or timedelta(days=3650))
secret = secrets.token_bytes(32)
sid = secrets.token_urlsafe(18)
session = PrivacySession(sid, mode, expiry, hashlib.sha256(secret).hexdigest())
self.sessions[sid] = session
return session
def register_temp_file(self, session_id: str, filepath: str) -> None:
"""注册临时文件路径,销毁时一并清理。"""
session = self._active(session_id)
session.temp_files.append(filepath)
def append(self, session_id: str, role: str, content: str) -> None:
session = self._active(session_id)
session.messages.append({"role": role, "content": content, "at": self.clock().isoformat()})
def destroy(self, session_id: str, retry: bool = True) -> bool:
"""销毁会话:清理消息、密钥、临时文件,记录销毁日志(不含正文)。"""
session = self.sessions.get(session_id)
if not session or session.destroyed:
return False
log_entry = {"at": self.clock().isoformat(), "steps": []}
try:
# 1. 清理消息
session.messages.clear()
log_entry["steps"].append("messages_cleared")
# 2. 销毁密钥
session.key_digest = ""
log_entry["steps"].append("key_destroyed")
# 3. 清理临时文件
for fp in session.temp_files:
try:
os.remove(fp)
log_entry["steps"].append(f"file_removed:{os.path.basename(fp)}")
except OSError:
pass
session.temp_files.clear()
log_entry["steps"].append("temp_files_cleared")
# 4. 标记已销毁
session.destroyed = True
log_entry["steps"].append("marked_destroyed")
session.destroy_log.append(log_entry)
self._destroy_stats["success"] += 1
return True
except Exception as e:
log_entry["error"] = str(e)
session.destroy_log.append(log_entry)
self._destroy_stats["failed"] += 1
if retry:
self._destroy_stats["retried"] += 1
return self.destroy(session_id, retry=False)
return False
def purge_expired(self) -> int:
removed = 0
for sid, session in list(self.sessions.items()):
if not session.destroyed and self.clock() >= session.expires_at:
self.destroy(sid)
removed += 1
return removed
def destroy_status(self, session_id: str) -> dict | None:
"""查询会话销毁状态和日志(不含正文)。"""
session = self.sessions.get(session_id)
if not session:
return None
return {
"session_id": session_id,
"destroyed": session.destroyed,
"mode": session.mode,
"destroy_log": session.destroy_log,
}
def destroy_stats(self) -> dict:
"""全局销毁统计(可观测性用)。"""
return dict(self._destroy_stats)
def _active(self, session_id: str) -> PrivacySession:
session = self.sessions.get(session_id)
if not session or session.destroyed:
raise KeyError("session is unavailable or destroyed")
if self.clock() >= session.expires_at:
self.destroy(session_id)
raise KeyError("session expired and was destroyed")
return session
def emergency_stop(self, session_id: str) -> bool:
"""Privacy急停:销毁会话并阻止后续写入。"""
return self.destroy(session_id)
def assess_ai(question: str, kimi_client=None) -> RiskCard | None:
"""AI 语义级风险分诊,返回 RiskCard 或 None(失败时降级到规则)。
安全原则:AI 只能升级风险等级,不能降级。
即规则判定 red/orange 时,不调用 AI(直接返回)。
"""
if not kimi_client:
return None
prompt = (
"你是中国中小企业法务风险分诊专家,服务对象是企业老板(雇主)。请判断以下问题的风险等级。\n\n"
f"用户问题:{question}\n\n"
"风险等级定义:\n"
"- red: 重大法律风险,可能涉及刑事、行政处罚、重大诉讼、人身安全,必须人工复核\n"
"- orange: 可能造成明显经济损失,需补证后行动\n"
"- amber: 需关注但可先行梳理\n"
"- green: 常规咨询,可先行梳理\n\n"
"请以 JSON 格式返回:\n"
'{"level": "red|orange|amber|green", "conclusion": "一句话风险描述", '
'"actions": ["行动建议1", "行动建议2"], "evidence": ["需保存的证据"], "escalation": true|false}\n\n'
"escalation 为 true 表示需要人工复核(律师/会计师/医生)。\n"
"只返回 JSON,不要其他内容。"
)
try:
answer = kimi_client.chat(
[{"role": "system", "content": "你是法务风险分诊专家,只返回 JSON。"},
{"role": "user", "content": prompt}],
temperature=0.1,
timeout=15,
)
# 解析 JSON
answer = answer.strip()
if answer.startswith("```"):
answer = answer.split("\n", 1)[1] if "\n" in answer else answer[3:]
if answer.endswith("```"):
answer = answer[:-3]
answer = answer.replace("\u201c", '"').replace("\u201d", '"').strip()
start = answer.find("{")
end = answer.rfind("}")
if start == -1 or end == -1:
return None
data = json.loads(answer[start:end + 1])
level = data.get("level", "green")
if level not in ("red", "orange", "amber", "green"):
level = "green"
return RiskCard(
conclusion=data.get("conclusion", ""),
level=level,
actions=data.get("actions", []),
evidence=data.get("evidence", []),
escalation=bool(data.get("escalation", False)),
)
except Exception:
return None
def assess(question: str, kimi_client=None) -> RiskCard:
"""安全优先的风险分诊器,四级分级(green/amber/orange/red)。
分诊策略:
1. 先跑规则初筛,命中 red/orange 直接返回(安全优先,不依赖 AI)
2. 规则未命中 red/orange 时,调 AI 做语义级分诊(可识别规则未覆盖的风险)
3. AI 不可用或失败时,降级到规则(amber/green)
Args:
question: 用户问题
kimi_client: KimiClient 实例,传入时启用 AI 分诊;None 时纯规则
"""
text = question.lower()
# 咨询性表述:倾向于绿色,除非有更高风险关键词
# 注意:"想变更劳动合同/想减资/想引入新股东/想反诉/想申请调查令/想申请再审/想变更法定代表人"
# 是操作意图,不是纯咨询,不列入
# 但"想变更公司经营范围/地址""想申请财产保全/强制执行"是常规咨询,列绿色
is_consultation = any(p in text for p in (
"想了解", "怎么开", "怎么约定", "怎么计算", "怎么办理", "怎么操作",
"怎么写", "流程是什么", "需要什么条件", "想注册", "怎么处理",
"想招", "想给", "想制定", "想设立", "想和", "想发", "想找",
"想调整", "想配", "怎么安排", "怎么评估",
"怎么跟进", "要不要", "怎么找", "怎么协商", "怎么确认",
"怎么催", "怎么提醒",
"想变更公司", "想申请财产保全", "想申请强制执行",
"诉讼时效是多久", "收到法院调解通知", "发票丢失了怎么办",
"合同里违约金写多少", "对方要求修改合同条款",
"欠款快到期想提前提醒",
"想变更公司经营范围", "想变更公司地址",
))
# 红色:重大法律风险,必须人工复核
red_terms = ("刑事", "诈骗", "被捕", "税务稽查通知", "重大诉讼",
"立案", "拘留", "侦查", "犯罪", "非法集资", "虚开发票",
"职务侵占", "挪用资金", "被警方", "限制出境",
"失信被执行", "吊销营业执照", "行贿", "受贿", "破产清算",
"伤害自己", "没有意义", "自残", "幻觉", "妄想", "就医",
"伪造签名", "伪造盖章", "非法交易", "敲诈勒索",
"卷款跑路", "涉嫌", "被调查", "被立案", "重大安全事故",
"举报违法", "非国家工作人员", "严重抑郁",
"被列为失信", "虚假合同骗取", "转移资产后跑路",
"被刑事", "被列为被执行人",
"盗取公司商业秘密", "侵犯商业秘密", "起诉了公司",
"被起诉侵犯", "税务稽查移交公安",
"挪用公司资金被查出")
if any(term in text for term in red_terms):
return RiskCard("该事项可能涉及重大法律风险。", "red",
["立即停止不可逆操作", "整理事实与证据", "联系执业律师复核"],
["合同、往来记录、通知和付款凭证"], True)
# 橙色:可能造成明显损失,需补证后行动
orange_terms = ("开除", "工伤", "根本违约", "篡改", "跑路",
"虚假资质", "转移财产", "停业", "失信", "破产",
"律师函", "否认收到", "控制权争夺", "转移公司资产",
"同业竞争", "抄袭", "抽逃出资", "决裂", "一蹶不振",
"持续失眠", "严重焦虑", "失去兴趣", "自暴自弃",
"补缴大额", "虚开", "账实不符", "转移定价",
"大额赔偿", "伪造证据",
"怀孕", "学历造假", "挪用", "私自接单",
"被竞争对手挖走", "带走客户", "工作中受伤",
"未缴社保", "不可抗力", "单方面终止",
"效力有争议", "拒绝承认", "条件未成就",
"欠款一年", "欠款客户跑路", "50万", "50 万",
"被列入失信", "关联公司税务", "补缴滞纳金",
"五年账本", "补缴增值税", "对方转移财产",
"找不到对方财产",
"股东起诉", "商业秘密被泄露", "股东要求解散",
"关联交易损害", "对方要求大额", "不服想上诉",
"关系破裂", "债务压得想放弃",
"被认定偷税", "偷税漏税", "税务异常",
"税务稽查要求", "稽查发现账", "稽查要求法人",
"对方起诉要求", "虚假理由起诉",
"严重冲突", "股东会决议被质疑",
"长期抑郁", "情绪波动很大", "恐惧和不安",
"不履行", "违约金过高", "损失扩大",
"恢复劳动关系", "仲裁要求赔偿",
"质量问题为由拒付全款",
"诉中保全", "对方申请再审", "对方提出鉴定",
"挪用公司资金被查出",
"客户被起诉了欠款", "对方提出反诉要求赔偿",
"加班费差额", "仲裁还能起诉",
"欠款客户提出管辖异议", "执行异议被驳回",
"身体和心理都很疲惫")
if any(term in text for term in orange_terms):
return RiskCard("需要补充关键事实后再采取行动。", "orange",
["确认主体、时间和金额", "保存原始证据", "先发合规通知,再决定下一步"],
["合同、考勤、聊天记录、发票和付款记录"], True)
# 规则未命中 red/orange,尝试 AI 语义分诊
ai_card = assess_ai(question, kimi_client)
if ai_card:
# AI 只能升级风险,不能降级:如果规则已判定 amber,AI 不能降为 green
# 但规则未命中 red/orange 时,AI 可以判定任意等级
return ai_card
# AI 不可用,降级到规则(amber/green 判定)
# 咨询性表述跳过 amber,除非命中强 amber 操作词
strong_amber = ("想注销公司税务", "股东分红税务", "股权激励税务",
"资产处置税务", "想变更劳动合同", "想减资", "想引入新股东",
"想反诉", "想申请调查令", "想申请再审",
"想变更法定代表人", "旷工三天", "拒绝加班",
"发票开具有误", "失眠很严重", "压力大到",
"对未来感到迷茫", "身体出了问题", "社交让我感到疲惫",
"欠款客户提出", "执行异议被驳回", "自查发现",
"诉讼中对方提出调解", "身体和心理",
"股东之间有分歧")
if not is_consultation or any(t in text for t in strong_amber):
amber_terms = ("试用期不合格", "迟到", "扣工资", "加班", "兼职", "绩效",
"调岗", "解除合同", "延迟交货", "违约金", "变更合同",
"签字人", "不利条款", "金额写错", "口头修改", "管辖",
"拒付", "打折", "转移资产", "以货抵款",
"被收购", "反诉", "发票", "逾期", "约谈", "抵扣",
"关联交易", "注销", "分红", "股权激励", "资产处置",
"应诉", "补充证据", "管辖异议", "财产保全",
"调查令", "上诉", "执行异议", "再审",
"股东分歧", "转让股权", "查阅", "减资", "未出资",
"变更法定代表人", "引入新股东", "章程", "董事",
"吵架", "失眠", "失去信心",
"撑不住", "崩溃", "喘不过气", "发脾气",
"补缴", "账本",
"对方上诉", "鉴定", "质证", "辞退", "仲裁",
"违约", "欠款", "合同条款", "歧义", "拒绝履行",
"附件", "矛盾", "竞业协议", "病假", "泄露薪资",
"出错造成损失", "加班费", "变更条款", "拒绝签",
"不认可", "终止合同", "不承认", "未成立",
"要求履行", "执行", "异议", "证人", "保全",
"股东之间", "出资", "解散", "关系很僵",
"身体和心理", "跨境收入",
"发现新证据", "执行异议被驳回",
"自查发现", "欠款客户提出",
"拒绝加班", "发票开具有误",
"旷工三天", "想变更劳动合同", "想减资",
"想引入新股东", "想反诉", "想申请调查令", "想申请再审",
"想变更法定代表人", "失眠很严重",
"压力大到", "对未来感到迷茫",
"身体出了问题", "社交让我感到疲惫",
"想注销公司税务", "股东分红税务",
"股权激励税务", "资产处置税务",
"诉讼中对方提出调解")
if any(term in text for term in amber_terms):
return RiskCard("该事项需要关注,建议先梳理事实再行动。", "amber",
["确认关键事实和证据", "评估最坏结果", "必要时咨询专业人士"],
["合同、往来记录、通知"], True)
# 绿色:可先行梳理
return RiskCard("这是一个可先行梳理的经营问题。", "green",
["明确目标和最坏结果", "列出今天能完成的一步"], [], False)
def redact(text: str) -> str:
"""脱敏手机号、身份证号、银行卡号(保存前和日志前调用)。
顺序:先身份证(18 位),再银行卡(16-19 位),最后手机号(11 位),
避免短数字正则截断长数字。
"""
# 身份证:18 位(最后一位可能是 X),用前后非数字边界
text = re.sub(r'(?<!\d)\d{17}[0-9Xx](?!\d)', '[证件号]', text)
# 银行卡:16-19 位连续数字,用前后非数字边界
text = re.sub(r'(?<!\d)\d{16,19}(?!\d)', '[账号]', text)
# 手机号:11 位,1[3-9] 开头,用前后非数字边界
text = re.sub(r'(?<!\d)1[3-9]\d{9}(?!\d)', '[手机号]', text)
return text
def action_summary(card: RiskCard) -> dict:
return {"risk_level": card.level, "conclusion": redact(card.conclusion), "actions": [redact(x) for x in card.actions]}
class ActionSummaryStore:
"""脱敏行动摘要存储:用户可选择保存,仅存脱敏后的结论和行动项。"""
# 允许保存的字段(白名单),保存前向用户展示
SAVABLE_FIELDS = ["risk_level", "conclusion", "actions"]
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
def _conn(self):
return psycopg2.connect(self.dsn)
def preview(self, summary: dict) -> dict:
"""保存前预览:展示将保留的脱敏字段。"""
return {k: summary.get(k) for k in self.SAVABLE_FIELDS if k in summary}
def save(self, session_id: str | None, summary: dict) -> dict:
"""保存脱敏行动摘要(仅白名单字段,二次脱敏)。"""
clean = {}
for k in self.SAVABLE_FIELDS:
if k in summary:
v = summary[k]
if isinstance(v, str):
clean[k] = redact(v)
elif isinstance(v, list):
clean[k] = [redact(x) if isinstance(x, str) else x for x in v]
else:
clean[k] = v
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"CREATE TABLE IF NOT EXISTS action_summaries "
"(id SERIAL PRIMARY KEY, session_id TEXT, summary JSONB NOT NULL, created_at TIMESTAMPTZ DEFAULT now())"
)
cur.execute(
"INSERT INTO action_summaries (session_id, summary) VALUES (%s, %s) RETURNING id, created_at",
(session_id, json.dumps(clean, ensure_ascii=False)),
)
row = cur.fetchone()
return {"id": row[0], "session_id": session_id, "summary": clean, "created_at": str(row[1])}
def list(self, limit: int = 20) -> List[dict]:
"""列出已保存的脱敏摘要。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, session_id, summary, created_at FROM action_summaries ORDER BY created_at DESC LIMIT %s",
(limit,),
)
rows = cur.fetchall()
return [
{"id": r[0], "session_id": r[1], "summary": r[2], "created_at": str(r[3])}
for r in rows
]
class LocalKnowledgeBase:
"""Minimal local keyword knowledge base; replaceable by an edge vector index."""
def __init__(self):
self.documents: List[dict] = []
def add(self, title: str, content: str, source: str = "local") -> dict:
doc = {"title": title, "content": content, "source": source}
self.documents.append(doc)
return doc
def search(self, query: str, limit: int = 5) -> List[dict]:
terms = set(re.findall(r"[\u4e00-\u9fffA-Za-z0-9]+", query.lower()))
scored = []
for doc in self.documents:
hay = (doc["title"] + doc["content"]).lower()
score = sum(1 for t in terms if t in hay)
if score:
scored.append((score, doc))
return [d for _, d in sorted(scored, key=lambda x: x[0], reverse=True)[:limit]]
class KnowledgeStore:
"""PostgreSQL 知识库,支持关键词检索 + TF-IDF 语义排序(RAG)。
7 类分类:法规(law)/案例(case)/模板(template)/经营指南(guide)/
行业知识(industry)/情绪疏导(emotion)/生活常识(life)。
不同场景注入不同分类的知识上下文。
使用 psycopg2 同步连接,适合 MVP 单进程。生产可换 asyncpg + pgvector。
"""
# 场景 → 检索的分类列表(按相关性排序)
SCENE_DOC_TYPES: dict[str, list[str]] = {
"legal": ["law", "case", "template"],
"decision": ["guide", "industry", "case"],
"life": ["life", "guide"],
"emotion": ["emotion", "life"],
}
# 全部分类定义
ALL_DOC_TYPES = ["law", "case", "template", "guide", "industry", "emotion", "life"]
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
def _conn(self):
return psycopg2.connect(self.dsn)
def add(self, title: str, content: str, source: str = "manual",
doc_type: str = "law", region: str | None = None,
effective_date: str | None = None, keywords: str | None = None) -> dict:
"""新增法规/案例/模板文档。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO knowledge_docs (title, content, source, doc_type, region, effective_date, keywords) "
"VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id, created_at",
(title, content, source, doc_type, region, effective_date, keywords),
)
row = cur.fetchone()
return {"id": row[0], "title": title, "content": content, "source": source,
"doc_type": doc_type, "region": region, "effective_date": effective_date,
"keywords": keywords, "created_at": str(row[1])}
def search(self, query: str, limit: int = 5, doc_type: str | None = None,
doc_types: list[str] | None = None) -> List[dict]:
"""RAG 检索:关键词匹配 + TF-IDF 排序,返回最相关文档。
使用 PostgreSQL 全文检索 + 应用层 TF-IDF 重排序,无需 pgvector 扩展。
中文采用 2-gram 分词提升召回率。
Args:
query: 检索查询
limit: 返回数量上限
doc_type: 单分类筛选(兼容旧接口)
doc_types: 多分类筛选(场景检索用,优先于 doc_type)
"""
# 中文 2-gram 分词 + 英文整词
raw_terms = re.findall(r"[\u4e00-\u9fff]+|[A-Za-z0-9]+", query.lower())
terms = set()
for t in raw_terms:
if re.match(r"[\u4e00-\u9fff]", t):
# 中文:生成 2-gram
for i in range(len(t) - 1):
terms.add(t[i:i+2])
if len(t) == 1:
terms.add(t)
else:
terms.add(t)
if not terms:
return []
# 数据库层:取候选集
with self._conn() as conn:
with conn.cursor() as cur:
if doc_types:
cur.execute(
"SELECT id, title, content, source, doc_type, region, effective_date, keywords "
"FROM knowledge_docs WHERE status = 'active' AND doc_type = ANY(%s) "
"ORDER BY id LIMIT 200",
(doc_types,),
)
elif doc_type:
cur.execute(
"SELECT id, title, content, source, doc_type, region, effective_date, keywords "
"FROM knowledge_docs WHERE status = 'active' AND doc_type = %s "
"ORDER BY id LIMIT 200",
(doc_type,),
)
else:
cur.execute(
"SELECT id, title, content, source, doc_type, region, effective_date, keywords "
"FROM knowledge_docs WHERE status = 'active' ORDER BY id LIMIT 200"
)
rows = cur.fetchall()
# 应用层:TF-IDF 排序
docs = []
for r in rows:
doc_text = (r[1] + " " + r[2] + " " + (r[7] or "")).lower()
# 词频
tf = sum(1 for t in terms if t in doc_text)
if tf == 0:
continue
# 标题命中加权
title_hit = sum(2 for t in terms if t in r[1].lower())
score = tf + title_hit
docs.append((score, r))
docs.sort(key=lambda x: x[0], reverse=True)
return [
{"id": r[0], "title": r[1], "content": r[2], "source": r[3],
"doc_type": r[4], "region": r[5], "effective_date": str(r[6]) if r[6] else None,
"keywords": r[7], "score": score}
for score, r in docs[:limit]
]
def list_all(self, doc_type: str | None = None, limit: int = 50) -> List[dict]:
"""列出全部文档(管理用)。"""
with self._conn() as conn:
with conn.cursor() as cur:
if doc_type:
cur.execute(
"SELECT id, title, content, source, doc_type, region, effective_date, status, keywords, created_at "
"FROM knowledge_docs WHERE doc_type = %s ORDER BY id DESC LIMIT %s",
(doc_type, limit),
)
else:
cur.execute(
"SELECT id, title, content, source, doc_type, region, effective_date, status, keywords, created_at "
"FROM knowledge_docs ORDER BY id DESC LIMIT %s",
(limit,),
)
rows = cur.fetchall()
return [
{"id": r[0], "title": r[1], "content": r[2], "source": r[3],
"doc_type": r[4], "region": r[5], "effective_date": str(r[6]) if r[6] else None,
"status": r[7], "keywords": r[8], "created_at": str(r[9])}
for r in rows
]
def delete(self, doc_id: int) -> bool:
"""删除文档(物理删除,法规库管理用)。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("DELETE FROM knowledge_docs WHERE id = %s", (doc_id,))
return cur.rowcount > 0
def build_context(self, query: str, limit: int = 3,
scene: str | None = None) -> str:
"""构建 RAG 上下文:检索相关文档,拼接为 system prompt 补充。
Args:
query: 用户问题
limit: 返回数量上限
scene: 场景(legal/decision/life/emotion),按场景检索对应分类
"""
doc_types = self.SCENE_DOC_TYPES.get(scene) if scene else None
results = self.search(query, limit=limit, doc_types=doc_types)
if not results:
return ""
parts = ["以下是相关知识库内容,供参考(必须引用来源,严禁编造):"]
for d in results:
parts.append(f"【{d['title']}】(来源:{d['source']},类型:{d['doc_type']},地区:{d.get('region') or '全国'})\n{d['content']}")
return "\n\n".join(parts)
def chunk_document(self, content: str, chunk_size: int = 500, overlap: int = 50) -> List[str]:
"""将文档分片,用于向量化索引。按字符数滑动窗口切割。"""
if len(content) <= chunk_size:
return [content]
chunks = []
start = 0
while start < len(content):
end = start + chunk_size
chunks.append(content[start:end])
start = end - overlap
return chunks
def build_tfidf_vectors(self) -> dict[int, dict[str, float]]:
"""构建全部文档的 TF-IDF 向量(内存索引,启动时加载)。
无需 pgvector,用 Python 字典实现轻量向量索引。
"""
docs = self.list_all(limit=500)
if not docs:
return {}
# 分词
tokenized = {}
for d in docs:
terms = set()
raw = re.findall(r"[\u4e00-\u9fff]+|[A-Za-z0-9]+", (d["title"] + " " + d["content"] + " " + (d.get("keywords") or "")).lower())
for t in raw:
if re.match(r"[\u4e00-\u9fff]", t):
for i in range(len(t) - 1):
terms.add(t[i:i+2])
if len(t) == 1:
terms.add(t)
else:
terms.add(t)
tokenized[d["id"]] = terms
# DF:每个词在多少文档中出现
df: Counter = Counter()
for terms in tokenized.values():
for t in terms:
df[t] += 1
n_docs = len(docs)
# TF-IDF
import math
vectors = {}
for doc_id, terms in tokenized.items():
tf = Counter(terms)
vec = {}
for t, freq in tf.items():
idf = math.log((n_docs + 1) / (df[t] + 1)) + 1
vec[t] = freq * idf
vectors[doc_id] = vec
return vectors
def vector_search(self, query: str, limit: int = 5) -> List[dict]:
"""基于 TF-IDF 向量的余弦相似度检索。"""
vectors = self.build_tfidf_vectors()
if not vectors:
return []
# 查询向量化
raw = re.findall(r"[\u4e00-\u9fff]+|[A-Za-z0-9]+", query.lower())
q_terms = set()
for t in raw:
if re.match(r"[\u4e00-\u9fff]", t):
for i in range(len(t) - 1):
q_terms.add(t[i:i+2])
if len(t) == 1:
q_terms.add(t)
else:
q_terms.add(t)
if not q_terms:
return []
import math
n_docs = len(vectors)
df: Counter = Counter()
for v in vectors.values():
for t in v:
df[t] += 1
q_vec = {}
for t in q_terms:
idf = math.log((n_docs + 1) / (df.get(t, 0) + 1)) + 1
q_vec[t] = idf
# 余弦相似度
q_norm = math.sqrt(sum(v * v for v in q_vec.values())) or 1
scores = []
for doc_id, vec in vectors.items():
dot = sum(q_vec.get(t, 0) * w for t, w in vec.items())
d_norm = math.sqrt(sum(w * w for w in vec.values())) or 1
sim = dot / (q_norm * d_norm)
if sim > 0:
scores.append((sim, doc_id))
scores.sort(reverse=True)
# 取 top-N 文档详情
top_ids = [doc_id for _, doc_id in scores[:limit]]
if not top_ids:
return []
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, title, content, source, doc_type, region, effective_date, keywords "
"FROM knowledge_docs WHERE id = ANY(%s)",
(top_ids,),
)
rows = cur.fetchall()
id_to_doc = {r[0]: r for r in rows}
results = []
for sim, doc_id in scores[:limit]:
r = id_to_doc.get(doc_id)
if r:
results.append({
"id": r[0], "title": r[1], "content": r[2], "source": r[3],
"doc_type": r[4], "region": r[5], "effective_date": str(r[6]) if r[6] else None,
"keywords": r[7], "score": round(sim, 4),
})
return results
class SettingsStore:
"""PostgreSQL 用户设置存储,key-value 模式。"""
DEFAULTS = {
"show_inference_location": "true",
"ephemeral_disable_copy": "true",
"auto_destroy_on_exit": "true",
"long_term_memory": "false",
"knowledge_base_write": "false",
"training_data_collection": "false",
"cloud_call_allowed": "true",
"pii_redact_on_transfer": "true",
"daily_reminder": "true",
"risk_alert": "true",
# 自动升级规则:检测到风险时是否自动创建复核工单
"auto_escalate_red": "true",
"auto_escalate_orange": "false",
}
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
def _conn(self):
return psycopg2.connect(self.dsn)
def get_all(self) -> dict[str, str]:
"""读取全部设置,缺失项用默认值填充。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("SELECT key, value FROM user_settings")
rows = cur.fetchall()
result = dict(self.DEFAULTS)
for k, v in rows:
result[k] = v
return result
def get(self, key: str) -> str:
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("SELECT value FROM user_settings WHERE key = %s", (key,))
row = cur.fetchone()
return row[0] if row else self.DEFAULTS.get(key, "")
def set(self, key: str, value: str) -> str:
"""写入设置(upsert)。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO user_settings (key, value, updated_at) VALUES (%s, %s, now()) "
"ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()",
(key, value),
)
return value
def set_many(self, items: dict[str, str]) -> dict[str, str]:
"""批量写入设置。"""
with self._conn() as conn:
with conn.cursor() as cur:
for k, v in items.items():
cur.execute(
"INSERT INTO user_settings (key, value, updated_at) VALUES (%s, %s, now()) "
"ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value, updated_at = now()",
(k, v),
)
return items
@dataclass
class UserProfile:
"""用户画像:累积的用户抽象特征(非对话内容)。
即使阅后即焚模式也保留这些抽象信息,用于越懂用户、精准回复。
按手机号关联用户。
"""
phone: str # 手机号(用户唯一标识)
industry: str = "" # 行业(retail/manufacturing/tech/service...)
company_size: str = "" # 企业规模(micro/small/medium/large)
role: str = "" # 角色(owner/manager/ceo...)
common_scenes: list = field(default_factory=list) # 常用场景(legal/decision/life/emotion)
common_topics: list = field(default_factory=list) # 常问话题(劳动用工/扩张/合同审查...)
risk_tolerance: str = "" # 风险偏好(conservative/moderate/aggressive)
communication_style: str = "" # 沟通偏好(concise/detailed/structured)
decision_style: str = "" # 决策风格(data-driven/intuitive/collaborative)
key_facts: dict = field(default_factory=dict) # 关键事实(公司类型/员工范围/地区等)
interaction_count: int = 0
last_active: str = ""
created_at: str = ""
def to_dict(self) -> dict:
"""转为字典(用于 JSONB 存储)。"""
return {
"industry": self.industry,
"company_size": self.company_size,
"role": self.role,
"common_scenes": self.common_scenes,
"common_topics": self.common_topics,
"risk_tolerance": self.risk_tolerance,
"communication_style": self.communication_style,
"decision_style": self.decision_style,
"key_facts": self.key_facts,
}
def to_context(self) -> str:
"""转为 prompt 注入文本(只包含有值的字段)。"""
parts = []
if self.industry:
parts.append(f"行业:{self.industry}")
if self.company_size:
parts.append(f"企业规模:{self.company_size}")
if self.role:
parts.append(f"角色:{self.role}")
if self.common_topics:
parts.append(f"常问话题:{', '.join(self.common_topics[:5])}")
if self.risk_tolerance:
parts.append(f"风险偏好:{self.risk_tolerance}")
if self.communication_style:
parts.append(f"沟通偏好:{self.communication_style}")
if self.decision_style:
parts.append(f"决策风格:{self.decision_style}")
if self.key_facts:
facts_str = ", ".join(f"{k}={v}" for k, v in self.key_facts.items())
parts.append(f"已知事实:{facts_str}")
if not parts:
return ""
return "用户画像:\n" + "\n".join(f"- {p}" for p in parts)
class UserProfileStore:
"""PostgreSQL 用户画像存储,按手机号累积用户特征。
隐私原则:
- 只存储抽象特征(行业/偏好/话题方向),不存储对话内容
- 阅后即焚模式也提取画像信号,但不保存对话原文
- 画像数据用于个性化回复,越用越懂用户
"""
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
def _conn(self):
return psycopg2.connect(self.dsn)
def get(self, phone: str) -> UserProfile:
"""读取用户画像,不存在则返回空画像。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT profile, interaction_count, last_active, created_at "
"FROM user_profiles WHERE phone = %s",
(phone,),
)
row = cur.fetchone()
if not row:
return UserProfile(phone=phone)
profile_data = row[0] or {}
return UserProfile(
phone=phone,
industry=profile_data.get("industry", ""),
company_size=profile_data.get("company_size", ""),
role=profile_data.get("role", ""),
common_scenes=profile_data.get("common_scenes", []),
common_topics=profile_data.get("common_topics", []),
risk_tolerance=profile_data.get("risk_tolerance", ""),
communication_style=profile_data.get("communication_style", ""),
decision_style=profile_data.get("decision_style", ""),
key_facts=profile_data.get("key_facts", {}),
interaction_count=row[1] or 0,
last_active=str(row[2]) if row[2] else "",
created_at=str(row[3]) if row[3] else "",
)
def update(self, phone: str, signals: dict, scene: str = "") -> UserProfile:
"""用 AI 提取的信号更新画像,合并已有特征。
Args:
phone: 手机号(用户唯一标识)
signals: AI 提取的画像信号(industry/role/topics 等)
scene: 当前场景(用于 common_scenes 累积)
Returns:
更新后的 UserProfile
"""
profile = self.get(phone)
# 合并信号(非空才更新)
for key in ("industry", "company_size", "role", "risk_tolerance",
"communication_style", "decision_style"):
val = signals.get(key, "")
if val:
setattr(profile, key, val)
# 列表型字段:累积去重
new_topics = signals.get("common_topics", [])
if new_topics:
for t in new_topics:
if t and t not in profile.common_topics:
profile.common_topics.insert(0, t)
profile.common_topics = profile.common_topics[:10] # 最多保留 10 个
if scene and scene not in profile.common_scenes:
profile.common_scenes.insert(0, scene)
profile.common_scenes = profile.common_scenes[:5]
# 字典型字段:合并
new_facts = signals.get("key_facts", {})
if new_facts:
profile.key_facts.update(new_facts)
# 更新计数和时间
profile.interaction_count += 1
# 写入数据库(upsert)
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO user_profiles (phone, profile, interaction_count, last_active, updated_at) "
"VALUES (%s, %s, %s, now(), now()) "
"ON CONFLICT (phone) DO UPDATE SET "
"profile = EXCLUDED.profile, "
"interaction_count = EXCLUDED.interaction_count, "
"last_active = EXCLUDED.last_active, "
"updated_at = now()",
(phone, json.dumps(profile.to_dict(), ensure_ascii=False),
profile.interaction_count),
)
# 写入历史快照(版本号递增)
cur.execute(
"SELECT COALESCE(MAX(version), 0) FROM user_profile_history WHERE phone = %s",
(phone,),
)
next_version = cur.fetchone()[0] + 1
source = signals.get("_source", "auto")
cur.execute(
"INSERT INTO user_profile_history (phone, version, profile, source) "
"VALUES (%s, %s, %s, %s)",
(phone, next_version,
json.dumps(profile.to_dict(), ensure_ascii=False), source),
)
return profile
def list_history(self, phone: str, limit: int = 20) -> List[dict]:
"""列出用户画像历史版本(按版本号倒序)。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, version, profile, source, created_at "
"FROM user_profile_history WHERE phone = %s "
"ORDER BY version DESC LIMIT %s",
(phone, limit),
)
rows = cur.fetchall()
return [
{
"id": r[0],
"version": r[1],
"profile": r[2],
"source": r[3],
"created_at": str(r[4]) if r[4] else "",
}
for r in rows
]
class UserStore:
"""PostgreSQL 用户账号存储:手机号+密码注册登录。
密码用 bcrypt 哈希,JWT token 用于会话保持。
"""
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
self.jwt_secret = os.getenv("JWT_SECRET", "bossai-dev-secret-change-in-prod")
def _conn(self):
return psycopg2.connect(self.dsn)
@staticmethod
def _hash_password(password: str) -> str:
"""bcrypt 哈希密码。"""
import bcrypt
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
@staticmethod
def _verify_password(password: str, hashed: str) -> bool:
"""验证密码。"""
import bcrypt
try:
return bcrypt.checkpw(password.encode(), hashed.encode())
except Exception:
return False
def _make_token(self, phone: str) -> str:
"""生成 JWT token(HMAC-SHA256,无外部依赖)。"""
import hmac
header = json.dumps({"alg": "HS256", "typ": "JWT"}).encode()
payload = json.dumps({
"phone": phone,
"exp": (datetime.now(timezone.utc) + timedelta(hours=24)).isoformat(),
}).encode()
import base64
h = base64.urlsafe_b64encode(header).rstrip(b"=").decode()
p = base64.urlsafe_b64encode(payload).rstrip(b"=").decode()
sig = hmac.new(self.jwt_secret.encode(), f"{h}.{p}".encode(), hashlib.sha256).digest()
s = base64.urlsafe_b64encode(sig).rstrip(b"=").decode()
return f"{h}.{p}.{s}"
def verify_token(self, token: str) -> str | None:
"""验证 JWT token,返回手机号或 None。
Args:
token: JWT token
Returns:
手机号(验证成功)或 None(失败/过期)
"""
import hmac
import base64
try:
parts = token.split(".")
if len(parts) != 3:
return None
h, p, s = parts
expected_sig = hmac.new(self.jwt_secret.encode(), f"{h}.{p}".encode(), hashlib.sha256).digest()
expected = base64.urlsafe_b64encode(expected_sig).rstrip(b"=").decode()
if not hmac.compare_digest(s, expected):
return None
payload = json.loads(base64.urlsafe_b64decode(p + "=="))
exp = datetime.fromisoformat(payload["exp"])
if datetime.now(timezone.utc) > exp:
return None
return payload.get("phone")
except Exception:
return None
def register(self, phone: str, password: str, name: str = "") -> dict:
"""注册新用户。
Returns:
{"phone": str, "name": str, "token": str}
Raises:
ValueError: 手机号已注册
"""
if not re.match(r'^1[3-9]\d{9}$', phone):
raise ValueError("手机号格式不正确")
if len(password) < 6:
raise ValueError("密码至少 6 位")
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("SELECT 1 FROM users WHERE phone = %s", (phone,))
if cur.fetchone():
raise ValueError("该手机号已注册")
hashed = self._hash_password(password)
cur.execute(
"INSERT INTO users (phone, password, name) VALUES (%s, %s, %s) "
"RETURNING id",
(phone, hashed, name or ""),
)
return {"phone": phone, "name": name, "token": self._make_token(phone)}
def login(self, phone: str, password: str) -> dict:
"""登录验证。
Returns:
{"phone": str, "name": str, "token": str}
Raises:
ValueError: 手机号或密码错误 / 账号已停用
"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("SELECT password, name, status FROM users WHERE phone = %s", (phone,))
row = cur.fetchone()
if not row:
raise ValueError("手机号或密码错误")
if not self._verify_password(password, row[0]):
raise ValueError("手机号或密码错误")
if row[2] == "inactive":
raise ValueError("账号已被停用,请联系管理员")
return {"phone": phone, "name": row[1] or "", "token": self._make_token(phone)}
def reset_password(self, user_id: int, new_password: str) -> bool:
"""管理员重置用户密码。
Args:
user_id: 用户 ID
new_password: 新密码(明文,内部 bcrypt 哈希)
Returns:
是否成功
"""
if len(new_password) < 6:
raise ValueError("密码至少 6 位")
hashed = self._hash_password(new_password)
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE users SET password = %s, updated_at = now() WHERE id = %s",
(hashed, user_id),
)
return cur.rowcount > 0
def update_status(self, user_id: int, status: str) -> bool:
"""管理员激活/停用用户。
Args:
user_id: 用户 ID
status: 'active' 或 'inactive'
Returns:
是否成功
"""
if status not in ("active", "inactive"):
raise ValueError("状态应为 active 或 inactive")
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE users SET status = %s, updated_at = now() WHERE id = %s",
(status, user_id),
)
return cur.rowcount > 0
class TicketStore:
"""PostgreSQL 人工复核工单存储,含 SLA 和专家自动分配。"""
# SLA 状态阈值:剩余时间 < 25% 为 warning,过期为 overdue
SLA_WARNING_RATIO = 0.25
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
def _conn(self):
return psycopg2.connect(self.dsn)
def _gen_ticket_no(self, cur) -> str:
"""生成工单号 RT-YYYYMMDD-NNN,按日递增。"""
today = datetime.now(timezone.utc).strftime("%Y%m%d")
prefix = f"RT-{today}-"
cur.execute(
"SELECT ticket_no FROM review_tickets WHERE ticket_no LIKE %s ORDER BY ticket_no DESC LIMIT 1",
(prefix + "%",),
)
row = cur.fetchone()
if row:
seq = int(row[0].rsplit("-", 1)[1]) + 1
else:
seq = 1
return f"{prefix}{seq:03d}"
def _get_expert_rule(self, cur, risk_level: str) -> dict | None:
"""按风险等级查询专家分配规则。"""
cur.execute(
"SELECT expert_type, sla_hours, auto_assign FROM expert_rules WHERE risk_level = %s",
(risk_level,),
)
row = cur.fetchone()
if not row:
return None
return {"expert_type": row[0], "sla_hours": row[1], "auto_assign": row[2]}
def _calc_sla_status(self, deadline: datetime, now: datetime | None = None) -> str:
"""计算 SLA 状态:normal / warning / overdue。"""
now = now or datetime.now(timezone.utc)
if now >= deadline:
return "overdue"
total = (deadline - now).total_seconds()
# 无法获取原始总时长时,用 deadline 距 now 的比例近似
# 简化:剩余 < 6 小时标记 warning
if total < 6 * 3600:
return "warning"
return "normal"
def _row_to_dict(self, r: tuple) -> dict:
"""将查询行转为字典,含 SLA 实时计算。"""
d = {
"id": r[0], "ticket_no": r[1], "session_id": r[2], "question": r[3],
"risk_level": r[4], "conclusion": r[5], "contact": r[6], "status": r[7],
"priority": r[8], "assignee": r[9], "resolution": r[10],
"expert_type": r[11], "sla_deadline": str(r[12]) if r[12] else None,
"sla_status": r[13], "created_at": str(r[14]), "updated_at": str(r[15]),
}
# 实时计算 SLA 状态(覆盖数据库中的静态值)
if d["sla_deadline"] and d["status"] not in ("resolved", "closed"):
try:
deadline = datetime.fromisoformat(d["sla_deadline"])
d["sla_status"] = self._calc_sla_status(deadline)
except Exception:
pass
return d
_SELECT_COLS = (
"id, ticket_no, session_id, question, risk_level, conclusion, contact, "
"status, priority, assignee, resolution, expert_type, sla_deadline, sla_status, "
"created_at, updated_at"
)
def create(self, question: str, risk_level: str, conclusion: str | None = None,
session_id: str | None = None, contact: str | None = None,
priority: str = "normal") -> dict:
"""创建工单。高风险自动 urgent,按规则分配专家类型和 SLA。"""
if risk_level == "red":
priority = "urgent"
with self._conn() as conn:
with conn.cursor() as cur:
rule = self._get_expert_rule(cur, risk_level)
expert_type = rule["expert_type"] if rule else None
sla_hours = rule["sla_hours"] if rule else 24
auto_assign = rule["auto_assign"] if rule else None
ticket_no = self._gen_ticket_no(cur)
cur.execute(
"INSERT INTO review_tickets "
"(ticket_no, session_id, question, risk_level, conclusion, contact, status, priority, "
"expert_type, assignee, sla_deadline, sla_status) "
"VALUES (%s, %s, %s, %s, %s, %s, 'pending', %s, %s, %s, now() + %s * interval '1 hour', 'normal') "
"RETURNING id, created_at",
(ticket_no, session_id, redact(question), risk_level,
redact(conclusion) if conclusion else None,
redact(contact) if contact else None, priority,
expert_type, auto_assign, sla_hours),
)
row = cur.fetchone()
return {
"id": row[0], "ticket_no": ticket_no, "status": "pending",
"priority": priority, "expert_type": expert_type,
"assignee": auto_assign, "sla_hours": sla_hours,
"created_at": str(row[1]),
}
def list(self, status: str | None = None, limit: int = 20) -> List[dict]:
"""列出工单,支持按状态筛选,实时计算 SLA。"""
with self._conn() as conn:
with conn.cursor() as cur:
if status:
cur.execute(
f"SELECT {self._SELECT_COLS} FROM review_tickets "
"WHERE status = %s ORDER BY created_at DESC LIMIT %s",
(status, limit),
)
else:
cur.execute(
f"SELECT {self._SELECT_COLS} FROM review_tickets "
"ORDER BY created_at DESC LIMIT %s",
(limit,),
)
rows = cur.fetchall()
return [self._row_to_dict(r) for r in rows]
def get(self, ticket_no: str) -> dict | None:
"""查询单个工单详情,实时计算 SLA。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
f"SELECT {self._SELECT_COLS} FROM review_tickets WHERE ticket_no = %s",
(ticket_no,),
)
r = cur.fetchone()
if not r:
return None
return self._row_to_dict(r)
def update_status(self, ticket_no: str, status: str,
assignee: str | None = None, resolution: str | None = None) -> dict | None:
"""更新工单状态/分配/处理结果。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE review_tickets SET status = %s, assignee = COALESCE(%s, assignee), "
"resolution = COALESCE(%s, resolution), updated_at = now() "
"WHERE ticket_no = %s RETURNING id",
(status, assignee, resolution, ticket_no),
)
if not cur.fetchone():
return None
return self.get(ticket_no)
def list_expert_rules(self) -> List[dict]:
"""列出全部专家分配规则(管理用)。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, risk_level, expert_type, sla_hours, auto_assign, created_at "
"FROM expert_rules ORDER BY id"
)
rows = cur.fetchall()
return [
{"id": r[0], "risk_level": r[1], "expert_type": r[2],
"sla_hours": r[3], "auto_assign": r[4], "created_at": str(r[5])}
for r in rows
]
def update_expert_rule(self, risk_level: str, expert_type: str | None = None,
sla_hours: int | None = None, auto_assign: str | None = None) -> dict | None:
"""更新专家分配规则。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE expert_rules SET "
"expert_type = COALESCE(%s, expert_type), "
"sla_hours = COALESCE(%s, sla_hours), "
"auto_assign = %s "
"WHERE risk_level = %s RETURNING id, risk_level, expert_type, sla_hours, auto_assign",
(expert_type, sla_hours, auto_assign, risk_level),
)
r = cur.fetchone()
if not r:
return None
return {"id": r[0], "risk_level": r[1], "expert_type": r[2],
"sla_hours": r[3], "auto_assign": r[4]}
def extract_text(filename: str, data: bytes) -> str:
"""提取文件文本,支持 TXT/MD/CSV/JSON/PDF/DOCX/图片(OCR)。
图片 OCR 使用 Tesseract(本地运行,无需联网,隐私优先)。
"""
suffix = Path(filename or "").suffix.lower()
if suffix in {".txt", ".md", ".csv", ".json"}:
return data.decode("utf-8", errors="replace")
if suffix == ".pdf":
try:
import pdfplumber
import io
with pdfplumber.open(io.BytesIO(data)) as pdf:
return "\n\n".join(p.extract_text() or "" for p in pdf.pages)
except Exception:
return ""
if suffix == ".docx":
try:
import docx
import io
doc = docx.Document(io.BytesIO(data))
return "\n".join(p.text for p in doc.paragraphs if p.text.strip())
except Exception:
return ""
if suffix in {".png", ".jpg", ".jpeg", ".tif", ".tiff", ".bmp", ".webp"}:
try:
import pytesseract
from PIL import Image
import io
img = Image.open(io.BytesIO(data))
# 优先中英文混合识别
text = pytesseract.image_to_string(img, lang="chi_sim+eng")
return text.strip()
except Exception:
return ""
return ""
class ExpertStore:
"""PostgreSQL 专家库存储:专家信息、上下架管理。
分类:legal(法务)/decision(经营)/emotion(情绪)/life(生活)。
"""
CATEGORIES = ["legal", "decision", "emotion", "life"]
CATEGORY_LABELS = {
"legal": "法务", "decision": "经营",
"emotion": "情绪", "life": "生活",
}
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
def _conn(self):
return psycopg2.connect(self.dsn)
def list_experts(self, category: str | None = None,
status: str | None = None) -> list[dict]:
"""列出专家,支持按分类和状态筛选。"""
with self._conn() as conn:
with conn.cursor() as cur:
conditions = []
params = []
if category:
conditions.append("category = %s")
params.append(category)
if status:
conditions.append("status = %s")
params.append(status)
where = " AND ".join(conditions) if conditions else "TRUE"
cur.execute(
f"SELECT id, name, phone, category, title, bio, expertise, "
f"avatar_url, rate, rating_count, status, created_at "
f"FROM experts WHERE {where} ORDER BY rate DESC, id",
params,
)
rows = cur.fetchall()
return [
{"id": r[0], "name": r[1], "phone": r[2], "category": r[3],
"title": r[4], "bio": r[5], "expertise": r[6], "avatar_url": r[7],
"rate": r[8], "rating_count": r[9], "status": r[10],
"created_at": str(r[11])}
for r in rows
]
def get_expert(self, expert_id: int) -> dict | None:
"""获取单个专家详情。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, name, phone, category, title, bio, expertise, "
"avatar_url, rate, rating_count, status, created_at "
"FROM experts WHERE id = %s",
(expert_id,),
)
r = cur.fetchone()
if not r:
return None
return {"id": r[0], "name": r[1], "phone": r[2], "category": r[3],
"title": r[4], "bio": r[5], "expertise": r[6], "avatar_url": r[7],
"rate": r[8], "rating_count": r[9], "status": r[10],
"created_at": str(r[11])}
def update_status(self, expert_id: int, status: str) -> bool:
"""上架/下架专家。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE experts SET status = %s, updated_at = now() WHERE id = %s",
(status, expert_id),
)
return cur.rowcount > 0
# 可编辑字段白名单(不含 id/phone/password/rate/rating_count/created_at)
EDITABLE_FIELDS = {"name", "category", "title", "bio", "expertise", "avatar_url", "status"}
def update_expert(self, expert_id: int, **fields) -> dict | None:
"""更新专家信息(仅允许白名单字段)。
Args:
expert_id: 专家 ID
**fields: 可编辑字段 name/category/title/bio/expertise/avatar_url/status
Returns:
更新后的专家 dict,专家不存在返回 None
"""
# 过滤非法字段;avatar_url 允许设为空字符串(清除头像)
updates = {}
for k, v in fields.items():
if k not in self.EDITABLE_FIELDS:
continue
if v is None:
continue
if k == "avatar_url" and v == "":
updates[k] = None # 空字符串 → NULL,清除头像
else:
updates[k] = v
if not updates:
return self.get_expert(expert_id)
# 校验 category
if "category" in updates and updates["category"] not in self.CATEGORIES:
raise ValueError(f"分类应为 {self.CATEGORIES} 之一")
# 校验 status
if "status" in updates and updates["status"] not in ("active", "inactive"):
raise ValueError("状态应为 active 或 inactive")
set_clauses = [f"{k} = %s" for k in updates]
params = list(updates.values()) + [expert_id]
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
f"UPDATE experts SET {', '.join(set_clauses)}, updated_at = now() "
f"WHERE id = %s",
params,
)
if cur.rowcount == 0:
return None
return self.get_expert(expert_id)
def create_service(self, user_phone: str, expert_id: int,
category: str) -> dict:
"""用户选定专家,创建服务关系(待专家确认)。
Raises:
ValueError: 已存在服务关系
"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, status FROM expert_services "
"WHERE user_phone = %s AND expert_id = %s",
(user_phone, expert_id),
)
existing = cur.fetchone()
if existing:
if existing[1] in ("pending", "active"):
raise ValueError("已选定该专家,请勿重复操作")
# closed 状态则重新发起
cur.execute(
"UPDATE expert_services SET status = 'pending', "
"signed_at = NULL, closed_at = NULL, updated_at = now() "
"WHERE id = %s",
(existing[0],),
)
return {"service_id": existing[0], "status": "pending"}
cur.execute(
"INSERT INTO expert_services (user_phone, expert_id, category, status) "
"VALUES (%s, %s, %s, 'pending') RETURNING id",
(user_phone, expert_id, category),
)
service_id = cur.fetchone()[0]
return {"service_id": service_id, "status": "pending"}
def list_user_services(self, user_phone: str) -> list[dict]:
"""列出用户的所有服务关系。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT es.id, es.expert_id, es.category, es.status, "
"es.signed_at, es.created_at, e.name, e.title, e.bio "
"FROM expert_services es JOIN experts e ON es.expert_id = e.id "
"WHERE es.user_phone = %s ORDER BY es.created_at DESC",
(user_phone,),
)
rows = cur.fetchall()
return [
{"service_id": r[0], "expert_id": r[1], "category": r[2],
"status": r[3], "signed_at": str(r[4]) if r[4] else None,
"created_at": str(r[5]), "expert_name": r[6],
"expert_title": r[7], "expert_bio": r[8]}
for r in rows
]
def list_expert_services(self, expert_id: int,
status: str | None = None) -> list[dict]:
"""列出专家的服务关系(工作台用)。"""
with self._conn() as conn:
with conn.cursor() as cur:
if status:
cur.execute(
"SELECT es.id, es.user_phone, es.category, es.status, "
"es.signed_at, es.created_at "
"FROM expert_services es "
"WHERE es.expert_id = %s AND es.status = %s "
"ORDER BY es.created_at DESC",
(expert_id, status),
)
else:
cur.execute(
"SELECT es.id, es.user_phone, es.category, es.status, "
"es.signed_at, es.created_at "
"FROM expert_services es "
"WHERE es.expert_id = %s ORDER BY es.created_at DESC",
(expert_id,),
)
rows = cur.fetchall()
return [
{"service_id": r[0], "user_phone": r[1], "category": r[2],
"status": r[3], "signed_at": str(r[4]) if r[4] else None,
"created_at": str(r[5])}
for r in rows
]
def accept_service(self, service_id: int, expert_id: int) -> bool:
"""专家确认服务关系(签协议)。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE expert_services SET status = 'active', "
"signed_at = now(), updated_at = now() "
"WHERE id = %s AND expert_id = %s AND status = 'pending'",
(service_id, expert_id),
)
return cur.rowcount > 0
def close_service(self, service_id: int, *,
user_phone: str | None = None,
expert_id: int | None = None) -> bool:
"""结束服务关系(用户或专家均可取消,需校验归属)。
Args:
service_id: 服务关系 ID
user_phone: 调用者为用户时传入,校验归属
expert_id: 调用者为专家时传入,校验归属
Returns:
是否成功取消
"""
with self._conn() as conn:
with conn.cursor() as cur:
# 校验归属:必须是该服务的用户或专家
if user_phone is not None:
cur.execute(
"SELECT 1 FROM expert_services "
"WHERE id = %s AND user_phone = %s AND status IN ('pending', 'active')",
(service_id, user_phone),
)
elif expert_id is not None:
cur.execute(
"SELECT 1 FROM expert_services "
"WHERE id = %s AND expert_id = %s AND status IN ('pending', 'active')",
(service_id, expert_id),
)
else:
return False
if not cur.fetchone():
return False
cur.execute(
"UPDATE expert_services SET status = 'closed', "
"closed_at = now(), updated_at = now() WHERE id = %s",
(service_id,),
)
return cur.rowcount > 0
class ConsultationStore:
"""PostgreSQL 专家咨询存储:用户发起咨询,专家回复。"""
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
def _conn(self):
return psycopg2.connect(self.dsn)
def create(self, service_id: int, user_phone: str, expert_id: int,
category: str, topic: str, context: str = "",
question: str = "") -> dict:
"""创建咨询:将 AI 对话话题发给专家。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"INSERT INTO expert_consultations "
"(service_id, user_phone, expert_id, category, topic, context, question, status) "
"VALUES (%s, %s, %s, %s, %s, %s, %s, 'pending') RETURNING id, created_at",
(service_id, user_phone, expert_id, category, topic, context, question),
)
row = cur.fetchone()
return {"consultation_id": row[0], "created_at": str(row[1]), "status": "pending"}
def list_user_consultations(self, user_phone: str,
status: str | None = None) -> list[dict]:
"""列出用户的咨询。"""
with self._conn() as conn:
with conn.cursor() as cur:
if status:
cur.execute(
"SELECT c.id, c.service_id, c.expert_id, c.category, c.topic, "
"c.question, c.reply, c.status, c.replied_at, c.created_at, "
"e.name, e.title "
"FROM expert_consultations c JOIN experts e ON c.expert_id = e.id "
"WHERE c.user_phone = %s AND c.status = %s "
"ORDER BY c.created_at DESC",
(user_phone, status),
)
else:
cur.execute(
"SELECT c.id, c.service_id, c.expert_id, c.category, c.topic, "
"c.question, c.reply, c.status, c.replied_at, c.created_at, "
"e.name, e.title "
"FROM expert_consultations c JOIN experts e ON c.expert_id = e.id "
"WHERE c.user_phone = %s ORDER BY c.created_at DESC",
(user_phone,),
)
rows = cur.fetchall()
return [
{"id": r[0], "service_id": r[1], "expert_id": r[2], "category": r[3],
"topic": r[4], "question": r[5], "reply": r[6], "status": r[7],
"replied_at": str(r[8]) if r[8] else None,
"created_at": str(r[9]), "expert_name": r[10], "expert_title": r[11]}
for r in rows
]
def list_expert_consultations(self, expert_id: int,
status: str | None = None) -> list[dict]:
"""列出专家收到的咨询(工作台用)。"""
with self._conn() as conn:
with conn.cursor() as cur:
if status:
cur.execute(
"SELECT c.id, c.service_id, c.user_phone, c.category, c.topic, "
"c.context, c.question, c.reply, c.status, c.replied_at, c.created_at "
"FROM expert_consultations c "
"WHERE c.expert_id = %s AND c.status = %s "
"ORDER BY c.created_at DESC",
(expert_id, status),
)
else:
cur.execute(
"SELECT c.id, c.service_id, c.user_phone, c.category, c.topic, "
"c.context, c.question, c.reply, c.status, c.replied_at, c.created_at "
"FROM expert_consultations c "
"WHERE c.expert_id = %s ORDER BY c.created_at DESC",
(expert_id,),
)
rows = cur.fetchall()
return [
{"id": r[0], "service_id": r[1], "user_phone": r[2], "category": r[3],
"topic": r[4], "context": r[5], "question": r[6], "reply": r[7],
"status": r[8], "replied_at": str(r[9]) if r[9] else None,
"created_at": str(r[10])}
for r in rows
]
def reply(self, consultation_id: int, expert_id: int, reply_text: str) -> bool:
"""专家回复咨询。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE expert_consultations SET reply = %s, status = 'replied', "
"replied_at = now(), updated_at = now() "
"WHERE id = %s AND expert_id = %s AND status = 'pending'",
(reply_text, consultation_id, expert_id),
)
return cur.rowcount > 0
def get(self, consultation_id: int) -> dict | None:
"""获取单个咨询详情。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, service_id, user_phone, expert_id, category, "
"topic, context, question, reply, status, replied_at, created_at "
"FROM expert_consultations WHERE id = %s",
(consultation_id,),
)
r = cur.fetchone()
if not r:
return None
return {"id": r[0], "service_id": r[1], "user_phone": r[2], "expert_id": r[3],
"category": r[4], "topic": r[5], "context": r[6], "question": r[7],
"reply": r[8], "status": r[9],
"replied_at": str(r[10]) if r[10] else None,
"created_at": str(r[11])}
class RatingStore:
"""PostgreSQL 专家评价存储。"""
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
def _conn(self):
return psycopg2.connect(self.dsn)
def create(self, consultation_id: int, user_phone: str, expert_id: int,
rating: int, comment: str = "") -> dict:
"""创建评价,同时更新专家平均分。"""
if rating < 1 or rating > 5:
raise ValueError("评分应在 1-5 之间")
with self._conn() as conn:
with conn.cursor() as cur:
# 检查是否已评价
cur.execute(
"SELECT 1 FROM expert_ratings WHERE consultation_id = %s",
(consultation_id,),
)
if cur.fetchone():
raise ValueError("该咨询已评价")
cur.execute(
"INSERT INTO expert_ratings (consultation_id, user_phone, expert_id, rating, comment) "
"VALUES (%s, %s, %s, %s, %s) RETURNING id, created_at",
(consultation_id, user_phone, expert_id, rating, comment),
)
row = cur.fetchone()
# 更新专家平均分(rate 存储为整数,乘以100)
cur.execute(
"SELECT rate, rating_count FROM experts WHERE id = %s",
(expert_id,),
)
expert = cur.fetchone()
old_total = (expert[0] or 0) * (expert[1] or 0)
new_count = (expert[1] or 0) + 1
new_rate = (old_total + rating * 100) // new_count
cur.execute(
"UPDATE experts SET rate = %s, rating_count = %s, updated_at = now() "
"WHERE id = %s",
(new_rate, new_count, expert_id),
)
return {"rating_id": row[0], "created_at": str(row[1])}
def list_by_expert(self, expert_id: int, limit: int = 20) -> list[dict]:
"""列出专家的评价。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT r.id, r.consultation_id, r.user_phone, r.rating, r.comment, r.created_at "
"FROM expert_ratings r WHERE r.expert_id = %s "
"ORDER BY r.created_at DESC LIMIT %s",
(expert_id, limit),
)
rows = cur.fetchall()
return [
{"id": r[0], "consultation_id": r[1], "user_phone": r[2][:3] + "****" + r[2][-4:],
"rating": r[3], "comment": r[4], "created_at": str(r[5])}
for r in rows
]
class RegulationUpdateManager:
"""法规更新管理:定时检查、变更检测、版本快照、审核工单、回滚。
工作流:
1. check_updates():扫描知识库,对比 effective_date 和 updated_at,标记需更新文档
2. create_snapshot():更新前创建版本快照(存入 knowledge_doc_versions)
3. update_document():更新文档内容,关联快照
4. create_review_ticket():为变更生成人工审核工单
5. rollback():回滚到指定版本
数据源:支持手动导入 + 预留外部 API 接口(fetch_from_source)。
"""
def __init__(self, dsn: str | None = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
self._ensure_version_table()
def _conn(self):
return psycopg2.connect(self.dsn)
def _ensure_version_table(self):
"""确保版本快照表存在。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("""
CREATE TABLE IF NOT EXISTS knowledge_doc_versions (
id SERIAL PRIMARY KEY,
doc_id INTEGER NOT NULL REFERENCES knowledge_docs(id),
version_num INTEGER NOT NULL,
title TEXT NOT NULL,
content TEXT NOT NULL,
source TEXT,
effective_date DATE,
snapshot_reason TEXT,
created_at TIMESTAMPTZ NOT NULL DEFAULT now()
)
""")
cur.execute(
"CREATE INDEX IF NOT EXISTS idx_doc_versions_doc "
"ON knowledge_doc_versions(doc_id, version_num)"
)
def check_updates(self) -> List[dict]:
"""扫描知识库,返回需要更新的文档列表。
判定逻辑:
- effective_date 已过但 status 仍为 active(法规已生效未更新)
- updated_at 超过 90 天未更新(可能过期)
- status 为 active 但 effective_date 在未来(未生效)
"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute("""
SELECT id, title, content, source, doc_type, region,
effective_date, status, updated_at
FROM knowledge_docs
WHERE status = 'active'
ORDER BY updated_at ASC
""")
rows = cur.fetchall()
from datetime import datetime, timezone, timedelta
now = datetime.now(timezone.utc)
needs_update = []
for r in rows:
doc_id, title, content, source, doc_type, region, eff_date, status, updated_at = r
reasons = []
# 超过 90 天未更新
if updated_at and (now - updated_at.replace(tzinfo=timezone.utc) > timedelta(days=90)):
reasons.append("超过90天未更新")
# 法规已生效但内容可能过期
if eff_date and doc_type == "law":
# eff_date 可能是 date 或 datetime,统一转 datetime
if hasattr(eff_date, 'tzinfo') and eff_date.tzinfo:
eff_dt = eff_date
elif hasattr(eff_date, 'tzinfo'):
eff_dt = eff_date.replace(tzinfo=timezone.utc)
else:
# datetime.date 对象,无 tzinfo
eff_dt = datetime.combine(eff_date, datetime.min.time(), tzinfo=timezone.utc)
if (now - eff_dt) > timedelta(days=365):
reasons.append("生效超过1年,需确认是否最新版本")
if reasons:
needs_update.append({
"id": doc_id, "title": title, "source": source,
"doc_type": doc_type, "region": region,
"effective_date": str(eff_date) if eff_date else None,
"updated_at": str(updated_at) if updated_at else None,
"reasons": reasons,
})
return needs_update
def create_snapshot(self, doc_id: int, reason: str = "更新前备份") -> dict:
"""为文档创建版本快照。"""
with self._conn() as conn:
with conn.cursor() as cur:
# 获取当前版本号
cur.execute(
"SELECT COALESCE(MAX(version_num), 0) + 1 "
"FROM knowledge_doc_versions WHERE doc_id = %s",
(doc_id,),
)
version_num = cur.fetchone()[0]
# 读取当前文档
cur.execute(
"SELECT title, content, source, effective_date "
"FROM knowledge_docs WHERE id = %s",
(doc_id,),
)
doc = cur.fetchone()
if not doc:
raise ValueError(f"文档不存在: {doc_id}")
# 插入快照
cur.execute(
"INSERT INTO knowledge_doc_versions "
"(doc_id, version_num, title, content, source, effective_date, snapshot_reason) "
"VALUES (%s, %s, %s, %s, %s, %s, %s) RETURNING id, created_at",
(doc_id, version_num, doc[0], doc[1], doc[2], doc[3], reason),
)
row = cur.fetchone()
return {"snapshot_id": row[0], "doc_id": doc_id, "version_num": version_num,
"created_at": str(row[1])}
def update_document(self, doc_id: int, content: str, title: str | None = None,
effective_date: str | None = None, keywords: str | None = None,
snapshot_reason: str = "法规更新") -> dict:
"""更新文档内容,自动创建更新前快照。"""
# 创建快照
snapshot = self.create_snapshot(doc_id, reason=snapshot_reason)
# 更新文档
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"UPDATE knowledge_docs SET "
"content = %s, "
"title = COALESCE(%s, title), "
"effective_date = COALESCE(%s, effective_date), "
"keywords = COALESCE(%s, keywords), "
"updated_at = now() "
"WHERE id = %s RETURNING id, title, updated_at",
(content, title, effective_date, keywords, doc_id),
)
row = cur.fetchone()
if not row:
raise ValueError(f"文档不存在: {doc_id}")
return {"id": row[0], "title": row[1], "updated_at": str(row[2]),
"snapshot": snapshot}
def list_versions(self, doc_id: int) -> List[dict]:
"""列出文档的所有版本快照。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"SELECT id, version_num, title, snapshot_reason, created_at "
"FROM knowledge_doc_versions WHERE doc_id = %s "
"ORDER BY version_num DESC",
(doc_id,),
)
rows = cur.fetchall()
return [
{"snapshot_id": r[0], "version_num": r[1], "title": r[2],
"reason": r[3], "created_at": str(r[4])}
for r in rows
]
def rollback(self, doc_id: int, version_num: int) -> dict:
"""回滚文档到指定版本。"""
with self._conn() as conn:
with conn.cursor() as cur:
# 读取快照内容
cur.execute(
"SELECT title, content, source, effective_date "
"FROM knowledge_doc_versions WHERE doc_id = %s AND version_num = %s",
(doc_id, version_num),
)
snapshot = cur.fetchone()
if not snapshot:
raise ValueError(f"版本不存在: doc_id={doc_id}, version={version_num}")
# 回滚前创建当前版本快照
cur.execute(
"SELECT COALESCE(MAX(version_num), 0) + 1 "
"FROM knowledge_doc_versions WHERE doc_id = %s",
(doc_id,),
)
new_version = cur.fetchone()[0]
cur.execute(
"SELECT title, content, source, effective_date FROM knowledge_docs WHERE id = %s",
(doc_id,),
)
current = cur.fetchone()
if current:
cur.execute(
"INSERT INTO knowledge_doc_versions "
"(doc_id, version_num, title, content, source, effective_date, snapshot_reason) "
"VALUES (%s, %s, %s, %s, %s, %s, %s)",
(doc_id, new_version, current[0], current[1], current[2], current[3], "回滚前备份"),
)
# 恢复内容
cur.execute(
"UPDATE knowledge_docs SET "
"title = %s, content = %s, source = %s, effective_date = %s, "
"updated_at = now() WHERE id = %s RETURNING id, title, updated_at",
(snapshot[0], snapshot[1], snapshot[2], snapshot[3], doc_id),
)
row = cur.fetchone()
return {"id": row[0], "title": row[1], "updated_at": str(row[2]),
"rolled_back_to": version_num, "backup_version": new_version}
def fetch_from_source(self, source_name: str) -> List[dict]:
"""从外部数据源获取法规更新(预留接口)。
当前返回空列表,生产环境可对接:
- 北大法宝 API
- 中国裁判文书网
- 国家法律法规数据库
"""
# 预留:实际对接外部 API 时实现
return []
class DocumentGenerator:
"""法律文书生成器:催款函、通知、证据清单、劳动催告。
本地生成 DOCX,无需联网。阅后即焚模式默认禁止导出。
"""
TEMPLATES = {
"demand_letter": "催款函",
"labor_notice": "劳动催告书",
"termination_notice": "解除合同通知书",
"evidence_checklist": "证据清单",
"general_notice": "通知书",
}
def __init__(self, ephemeral_mode: bool = False, kimi_client=None):
"""ephemeral_mode 为 True 时禁止导出(阅后即焚)。
Args:
ephemeral_mode: 阅后即焚模式标记
kimi_client: KimiClient 实例,传入时启用 AI 生成正文
"""
self.ephemeral_mode = ephemeral_mode
self.kimi_client = kimi_client
def generate_ai_content(self, template: str, fields: dict) -> str | None:
"""AI 根据模板类型和字段生成文书正文。
Returns:
生成的正文文本,失败时返回 None(降级到模板默认内容)
"""
if not self.kimi_client:
return None
title = self.TEMPLATES.get(template, "通知书")
prompt = (
f"你是法律文书撰写专家。请根据以下信息生成一份{title}的正文内容。\n\n"
f"已知信息:\n"
)
for k, v in fields.items():
if k not in ("content", "items", "signer", "date"):
prompt += f"- {k}:{v}\n"
if "items" in fields:
prompt += f"- 证据项:{', '.join(fields['items'])}\n"
prompt += (
"\n要求:\n"
"1. 正文 200-400 字,语气正式、法律用语规范\n"
"2. 引用相关法律依据(如适用)\n"
"3. 明确要求对方在指定期限内履行义务\n"
"4. 只返回正文内容,不要标题和落款\n"
)
try:
return self.kimi_client.chat(
[{"role": "system", "content": "你是法律文书撰写助手,服务对象是中小企业老板(原告方),生成正式规范的文书正文。"},
{"role": "user", "content": prompt}],
temperature=0.3,
timeout=30,
)
except Exception:
return None
def generate(self, template: str, fields: dict) -> bytes:
"""生成 DOCX 文件,返回二进制内容。
Args:
template: 模板类型(demand_letter/labor_notice/termination_notice/
evidence_checklist/general_notice)
fields: 文书字段,如 {
"creditor": "债权人名称",
"debtor": "债务人名称",
"amount": "欠款金额",
"date": "日期",
"content": "正文内容",
"items": ["证据项1", "证据项2"], # 证据清单用
}
Returns:
DOCX 文件二进制内容
Raises:
ValueError: 阅后即焚模式或未知模板
"""
if self.ephemeral_mode:
raise ValueError("阅后即焚模式下禁止导出文件")
if template not in self.TEMPLATES:
raise ValueError(f"未知模板: {template},可选: {list(self.TEMPLATES.keys())}")
# AI 生成正文(如果未提供 content 且有 kimi_client)
if not fields.get("content") and self.kimi_client:
ai_content = self.generate_ai_content(template, fields)
if ai_content:
fields = {**fields, "content": ai_content}
import docx
from docx.shared import Pt, Cm
from docx.enum.text import WD_ALIGN_PARAGRAPH
import io
doc = docx.Document()
# 标题
title = self.TEMPLATES[template]
heading = doc.add_heading(title, level=1)
heading.alignment = WD_ALIGN_PARAGRAPH.CENTER
# 模板特定内容
if template == "demand_letter":
self._fill_demand_letter(doc, fields)
elif template == "labor_notice":
self._fill_labor_notice(doc, fields)
elif template == "termination_notice":
self._fill_termination_notice(doc, fields)
elif template == "evidence_checklist":
self._fill_evidence_checklist(doc, fields)
else:
self._fill_general_notice(doc, fields)
# 落款
doc.add_paragraph("")
sign_date = fields.get("date", "")
signer = fields.get("signer", "")
if sign_date:
p = doc.add_paragraph(sign_date)
p.alignment = WD_ALIGN_PARAGRAPH.RIGHT
if signer:
p = doc.add_paragraph(f"{signer}(签章)")
p.alignment = WD_ALIGN_PARAGRAPH.RIGHT
# 导出为二进制
buf = io.BytesIO()
doc.save(buf)
return buf.getvalue()
def _fill_demand_letter(self, doc, fields: dict):
"""催款函。"""
doc.add_paragraph(f"{fields.get('debtor', '___')}:")
doc.add_paragraph("")
body = fields.get("content") or (
f"贵方于 {fields.get('contract_date', '___')} 与我方签订"
f"《{fields.get('contract_name', '___')}》,约定贵方应于"
f"{fields.get('due_date', '___')} 支付 {fields.get('amount', '___')}。"
f"截至本函发出之日,贵方尚未履行付款义务,欠款金额为"
f"{fields.get('amount', '___')}。"
)
doc.add_paragraph(body)
doc.add_paragraph("")
doc.add_paragraph(
f"请贵方在收到本函后 {fields.get('deadline_days', '7')} 日内付清上述款项。"
"如逾期未付,我方将依法采取进一步措施,包括但不限于提起诉讼、"
"申请财产保全等,由此产生的费用由贵方承担。"
)
doc.add_paragraph("")
doc.add_paragraph(f"债权人:{fields.get('creditor', '___')}")
def _fill_labor_notice(self, doc, fields: dict):
"""劳动催告书。"""
doc.add_paragraph(f"{fields.get('employee', '___')} 同志:")
doc.add_paragraph("")
body = fields.get("content") or (
f"您于 {fields.get('incident_date', '___')} 发生"
f"{fields.get('incident', '___')} 事项。根据相关法律法规及公司规章制度,"
f"现通知您:{fields.get('notice_content', '___')}。"
)
doc.add_paragraph(body)
doc.add_paragraph("")
doc.add_paragraph(
"请在收到本通知后 3 日内予以回复。如有异议,"
"可依法申请劳动仲裁或向劳动行政部门投诉。"
)
doc.add_paragraph("")
doc.add_paragraph(f"用人单位:{fields.get('employer', '___')}")
def _fill_termination_notice(self, doc, fields: dict):
"""解除合同通知书。"""
doc.add_paragraph(f"{fields.get('counterparty', '___')}:")
doc.add_paragraph("")
body = fields.get("content") or (
f"我方与贵方于 {fields.get('contract_date', '___')} 签订的"
f"《{fields.get('contract_name', '___')}》,因贵方"
f"{fields.get('breach_reason', '___')},已构成根本违约。"
f"根据合同第 {fields.get('clause', '___')} 条及相关法律规定,"
"我方现正式通知贵方:解除上述合同。"
)
doc.add_paragraph(body)
doc.add_paragraph("")
doc.add_paragraph(
"请贵方在收到本通知后 10 日内与我方办理结算事宜。"
"我方保留追究贵方违约责任的权利。"
)
doc.add_paragraph("")
doc.add_paragraph(f"通知方:{fields.get('notifier', '___')}")
def _fill_evidence_checklist(self, doc, fields: dict):
"""证据清单。"""
doc.add_paragraph(f"案件:{fields.get('case_name', '___')}")
doc.add_paragraph("")
doc.add_paragraph("以下为需收集保存的证据清单:")
doc.add_paragraph("")
items = fields.get("items", [])
if not items:
items = ["合同原件", "往来邮件/聊天记录", "付款凭证",
"发票", "通知函件", "其他相关证据"]
for i, item in enumerate(items, 1):
doc.add_paragraph(f"{i}. {item}", style="List Number")
doc.add_paragraph("")
doc.add_paragraph("注:请确保证据原件完好,复印件与原件核对一致。")
def _fill_general_notice(self, doc, fields: dict):
"""通用通知书。"""
doc.add_paragraph(f"{fields.get('recipient', '___')}:")
doc.add_paragraph("")
body = fields.get("content", "")
doc.add_paragraph(body)
doc.add_paragraph("")
doc.add_paragraph(f"通知方:{fields.get('notifier', '___')}")
class FactFinder:
"""事实追问器:分析用户问题,识别缺失关键事实,生成追问,多轮对话后汇总结论。
工作流:
1. analyze():分析问题类型,识别缺失的关键事实维度
2. generate_questions():针对缺失维度生成结构化追问
3. should_ask_more():根据用户回答判断是否还需追问
4. summarize():事实充分后,汇总已知事实供 AI 逐层分析
每个场景(legal/decision/life/emotion)有不同的关键事实维度。
最多追问 2 轮,每轮 1-3 个问题,避免过度打扰用户。
"""
# 各场景的关键事实维度
FACT_DIMENSIONS: dict[str, list[dict]] = {
"legal": [
{"key": "contract", "desc": "是否有书面合同/协议",
"questions": ["双方是否签订了书面合同或协议?", "合同中是否有相关条款约定?"]},
{"key": "timeline", "desc": "事件发生时间和当前状态",
"questions": ["事件是什么时候发生的?", "目前处于什么阶段(协商/纠纷/诉讼/执行)?"]},
{"key": "evidence", "desc": "已掌握的证据",
"questions": ["目前有哪些证据(合同/邮件/聊天记录/付款凭证)?", "是否有书面记录?"]},
{"key": "amount", "desc": "涉及金额",
"questions": ["涉及的金额大约是多少?", "是否已经造成实际损失?"]},
{"key": "party", "desc": "对方当事人情况",
"questions": ["对方是个人还是企业?", "对方目前的态度如何(配合/回避/对抗)?"]},
],
"decision": [
{"key": "scale", "desc": "决策规模和影响",
"questions": ["这个决策涉及的金额/规模大约是多少?", "对公司整体业务的影响程度如何?"]},
{"key": "cashflow", "desc": "现金流状况",
"questions": ["公司目前的现金流能支撑多长时间?", "是否有备用资金来源?"]},
{"key": "alternatives", "desc": "是否有备选方案",
"questions": ["除了当前方案,是否考虑过其他选择?", "不做的后果是什么?"]},
{"key": "timeline", "desc": "决策时间窗口",
"questions": ["这个决策需要在什么时间前做出?", "是否有外部截止日期?"]},
{"key": "market", "desc": "市场和竞争状况",
"questions": ["目标市场的增长趋势如何?", "主要竞争对手的情况怎样?"]},
{"key": "team", "desc": "团队和管理能力",
"questions": ["现有团队能否支撑这个决策?", "是否需要新增关键人员?"]},
{"key": "risk_tolerance", "desc": "风险承受能力",
"questions": ["如果这个决策失败,最坏的结果是什么?", "公司能承受多大的损失?"]},
{"key": "profitability", "desc": "盈利能力",
"questions": ["预期利润率大约是多少?", "投资回收期预计多长?"]},
],
"life": [
{"key": "context", "desc": "具体情境",
"questions": ["能具体描述一下当前的情况吗?", "最希望解决的核心问题是什么?"]},
{"key": "constraint", "desc": "时间和资源限制",
"questions": ["目前有多少时间可以投入?", "有哪些资源可以利用?"]},
],
"emotion": [
{"key": "feeling", "desc": "当前情绪状态",
"questions": ["现在最强烈的感受是什么?", "这种情况持续多久了?"]},
{"key": "trigger", "desc": "触发事件",
"questions": ["是什么事情让你有这种感觉?", "最近发生了什么变化?"]},
],
}
# 问题类型关键词 → 优先追问维度
TYPE_PRIORITY: dict[str, list[str]] = {
"劳动": ["contract", "timeline", "party"],
"旷工": ["contract", "timeline", "party"],
"合同": ["contract", "amount", "timeline"],
"欠款": ["amount", "evidence", "timeline"],
"催款": ["amount", "evidence", "party"],
"税务": ["amount", "timeline", "evidence"],
"诉讼": ["timeline", "evidence", "amount"],
"工伤": ["timeline", "evidence", "contract"],
"违约": ["contract", "amount", "evidence"],
"股东": ["contract", "party", "amount"],
"解雇": ["contract", "timeline", "party"],
"裁员": ["contract", "amount", "party"],
"工资": ["contract", "amount", "timeline"],
"社保": ["contract", "timeline", "amount"],
"决策": ["scale", "cashflow", "alternatives"],
"扩张": ["scale", "cashflow", "market"],
"开店": ["scale", "cashflow", "market"],
"分店": ["scale", "cashflow", "market"],
"新业务": ["scale", "cashflow", "market"],
"融资": ["scale", "cashflow", "profitability"],
"投资": ["scale", "cashflow", "profitability"],
"合伙": ["scale", "team", "contract"],
"入股": ["scale", "team", "contract"],
"股权": ["scale", "team", "contract"],
"降价": ["profitability", "market", "cashflow"],
"涨价": ["profitability", "market", "alternatives"],
"促销": ["profitability", "market", "scale"],
"招人": ["team", "cashflow", "scale"],
"招聘": ["team", "cashflow", "scale"],
"招": ["team", "cashflow", "scale"],
"招个": ["team", "cashflow", "scale"],
"招一位": ["team", "cashflow", "scale"],
"总监": ["team", "cashflow", "scale"],
"经理": ["team", "cashflow", "scale"],
"主管": ["team", "cashflow", "scale"],
"员工": ["team", "cashflow", "scale"],
"裁员": ["team", "cashflow", "scale"],
"减员": ["team", "cashflow", "scale"],
"收缩": ["cashflow", "scale", "alternatives"],
"关店": ["cashflow", "scale", "alternatives"],
"转型": ["market", "cashflow", "team"],
"外包": ["scale", "cashflow", "alternatives"],
"情绪": ["feeling", "trigger"],
"压力": ["feeling", "trigger"],
"焦虑": ["feeling", "trigger"],
"失眠": ["feeling", "trigger"],
}
def __init__(self, max_rounds: int = 2, max_questions_per_round: int = 3):
self.max_rounds = max_rounds
self.max_questions_per_round = max_questions_per_round
def analyze(self, question: str, scene: str) -> dict:
"""分析用户问题,返回问题类型和缺失的事实维度。
Returns:
{
"question_type": str, # 问题类型(劳动/合同/欠款/...)
"scene": str,
"priority_dims": list[str], # 优先追问的维度 key
"missing_dims": list[dict], # 缺失的事实维度详情
}
"""
# 识别问题类型
question_type = "general"
for keyword, dims in self.TYPE_PRIORITY.items():
if keyword in question:
question_type = keyword
break
# 获取该场景的事实维度
all_dims = self.FACT_DIMENSIONS.get(scene, self.FACT_DIMENSIONS["legal"])
# 按优先级排序
priority_keys = self.TYPE_PRIORITY.get(question_type, [])
priority_dims = [d["key"] for d in all_dims if d["key"] in priority_keys]
# 补充未在优先级中的维度
other_dims = [d["key"] for d in all_dims if d["key"] not in priority_keys]
ordered_keys = priority_dims + other_dims
# 按优先级顺序构建 missing_dims(而非 all_dims 原始顺序)
dim_map = {d["key"]: d for d in all_dims}
missing_dims = [dim_map[k] for k in ordered_keys if k in dim_map]
return {
"question_type": question_type,
"scene": scene,
"priority_dims": priority_dims,
"missing_dims": missing_dims,
}
def generate_questions(self, analysis: dict, round_num: int,
previous_answers: list[dict] = None) -> list[dict]:
"""生成追问问题。
Args:
analysis: analyze() 的返回结果
round_num: 当前追问轮次(1 开始)
previous_answers: 之前轮次的用户回答
Returns:
[{"dim_key": str, "dim_desc": str, "question": str}]
"""
previous_answers = previous_answers or []
# 已回答的维度不再追问
answered_keys = set()
for ans in previous_answers:
answered_keys.add(ans.get("dim_key"))
missing = [d for d in analysis["missing_dims"] if d["key"] not in answered_keys]
# 取前 max_questions_per_round 个
questions = []
for dim in missing[:self.max_questions_per_round]:
# 每个维度取第一个问题(简洁)
q = dim["questions"][0] if dim["questions"] else dim["desc"]
questions.append({
"dim_key": dim["key"],
"dim_desc": dim["desc"],
"question": q,
})
return questions
def should_ask_more_ai(self, question: str, scene: str,
previous_answers: list[dict],
kimi_client) -> bool | None:
"""AI 判断已有事实是否足够分析。
Returns:
True: 还需追问
False: 事实已充分
None: AI 不可用,降级到规则
"""
if not kimi_client:
return None
scene_desc = {
"legal": "法律", "decision": "经营决策",
"life": "生活", "emotion": "情绪疏导",
}.get(scene, "通用")
prompt = (
f"你是{scene_desc}顾问,服务对象是中小企业老板。判断已有事实是否足够给出有价值的建议。\n\n"
f"用户问题:{question}\n\n"
)
if previous_answers:
prompt += "已知事实:\n"
for ans in previous_answers:
prompt += f"- {ans.get('dim_desc', '')}:{ans.get('answer', '')}\n"
else:
prompt += "已知事实:无\n"
prompt += (
"\n请判断:基于以上事实,是否能给出有针对性、可执行的建议?\n"
"如果事实已充分,返回 {\"need_more\": false}\n"
"如果还需追问,返回 {\"need_more\": true}\n"
"只返回 JSON,不要其他内容。"
)
try:
answer = kimi_client.chat(
[{"role": "system", "content": "你是事实充分性判断助手,只返回 JSON。"},
{"role": "user", "content": prompt}],
temperature=0.1,
timeout=10,
)
answer = answer.strip()
if answer.startswith("```"):
answer = answer.split("\n", 1)[1] if "\n" in answer else answer[3:]
if answer.endswith("```"):
answer = answer[:-3]
answer = answer.replace("\u201c", '"').replace("\u201d", '"').strip()
start = answer.find("{")
end = answer.rfind("}")
if start == -1 or end == -1:
return None
data = json.loads(answer[start:end + 1])
return bool(data.get("need_more", True))
except Exception:
return None
def should_ask_more(self, analysis: dict, round_num: int,
previous_answers: list[dict],
kimi_client=None, question: str = "",
scene: str = "legal") -> bool:
"""判断是否还需要继续追问。
优先用 AI 判断事实充分性,降级到规则:
- 已达最大轮次 → False
- 优先维度都已回答 → False
- 无优先维度时,默认追问前 3 个维度
"""
if round_num >= self.max_rounds:
return False
# 优先 AI 判断
if kimi_client and question:
ai_result = self.should_ask_more_ai(question, scene, previous_answers, kimi_client)
if ai_result is not None:
return ai_result
# 降级到规则
answered_keys = set(ans.get("dim_key") for ans in previous_answers)
priority_dims = analysis.get("priority_dims", [])
# 无优先维度时,用前 3 个维度作为默认追问
if not priority_dims:
all_dims = analysis.get("missing_dims", [])
priority_dims = [d["key"] for d in all_dims[:3]]
# 优先维度都已回答
if all(d in answered_keys for d in priority_dims):
return False
# 还有优先维度未回答
return True
def summarize_facts(self, question: str, analysis: dict,
answers: list[dict]) -> str:
"""汇总已知事实,供 AI 逐层分析。
Returns:
结构化的事实摘要文本,拼接到 system prompt 后
"""
scene = analysis.get("scene", "legal")
lines = [f"用户原始问题:{question}", "", "已知事实:"]
for ans in answers:
dim_desc = ans.get("dim_desc", "")
user_answer = ans.get("answer", "")
lines.append(f"- {dim_desc}:{user_answer}")
# 未回答的维度
answered_keys = set(ans.get("dim_key") for ans in answers)
unanswered = [d for d in analysis["missing_dims"] if d["key"] not in answered_keys]
if unanswered:
lines.append("")
lines.append("未确认的事实(分析时需标注假设):")
for d in unanswered:
lines.append(f"- {d['desc']}")
lines.append("")
# 按场景给出不同的分析指令
scene_instruction = {
"legal": "请基于以上已知事实进行逐层分析:先确认事实,再分析法律风险和依据,最后给出结论和行动建议。对未确认的事实明确标注为假设。",
"decision": "请基于以上已知事实进行逐层分析:先确认事实,再进行情景推演(收益/风险/现金流/推荐程度),最后给出明确推荐和第一步行动。对未确认的事实明确标注为假设。",
"life": "请基于以上已知事实给出务实、具体、可执行的建议。先确认已知情况,再给出今天能做的一件小事。对未确认的事实明确标注为假设。避免说教,尊重老板的时间。",
"emotion": "请基于以上已知事实进行回应。先承接当前状态,再给出温和的下一步建议。若检测到自伤、绝望等危机信号,立即提示联系心理援助热线 400-161-9995 或就医。",
}
lines.append(scene_instruction.get(scene, "请基于以上已知事实进行逐层分析,对未确认的事实明确标注为假设。"))
return "\n".join(lines)
def generate_ai_questions(self, question: str, scene: str,
kimi_client, previous_answers: list[dict] = None,
max_questions: int = 3, history: list[dict] = None) -> list[dict]:
"""用 AI 分析问题,生成针对性追问问题。
Args:
question: 用户原始问题
scene: 场景(legal/decision/life/emotion)
kimi_client: KimiClient 实例
previous_answers: 之前轮次的用户回答
max_questions: 最多生成几个问题
Returns:
[{"dim_key": str, "dim_desc": str, "question": str}]
失败时返回空列表(调用方降级到规则匹配)
"""
previous_answers = previous_answers or []
# 按场景定制追问引导
scene_config = {
"legal": {
"desc": "法律/法务",
"role": "你是中国执业律师,服务对象是中小企业老板(雇主),所有追问必须从老板/雇主视角出发",
"dim_examples": "合同、金额、时间、证据、对方态度、是否协商过",
"focus": "影响法律结论的关键事实",
},
"decision": {
"desc": "经营决策",
"role": "你是中小企业经营顾问,服务对象是企业老板,所有追问从老板视角出发",
"dim_examples": "规模金额、现金流、市场状况、团队能力、备选方案、时间窗口",
"focus": "影响决策推演的关键事实",
},
"life": {
"desc": "生活助理",
"role": "你是生活管理顾问,擅长家庭、健康、出行等实用建议,服务对象是企业老板",
"dim_examples": "具体状况、涉及人员、时间安排、当前困难、已有尝试",
"focus": "让建议更具体可执行的关键细节,而非泛泛的'描述情况'",
},
"emotion": {
"desc": "情绪疏导",
"role": "你是心理咨询师,服务对象是企业老板",
"dim_examples": "触发事件、持续时间、当前感受、已有应对、社会支持",
"focus": "理解情绪状态和触发因素的关键信息",
},
}
config = scene_config.get(scene, scene_config["legal"])
prompt = (
f"{config['role']}。用户提出了一个{config['desc']}问题,"
f"在给出完整建议之前,需要先确认几个关键事实。\n\n"
f"用户问题:{question}\n\n"
)
# 注入对话历史(多轮上下文),让追问能理解当前问题的背景
if history:
prompt += "之前的对话历史(用于理解当前问题的背景):\n"
for h in history[-6:]:
role_label = "用户" if h.get("role") == "user" else "AI"
content = h.get("content", "")
if content:
prompt += f"{role_label}: {content[:200]}\n"
prompt += "\n注意:当前问题是在上述对话背景下的延续,追问应结合之前的上下文,不要当成孤立问题处理。\n\n"
if previous_answers:
prompt += "已知事实:\n"
for ans in previous_answers:
prompt += f"- {ans.get('dim_desc', '')}:{ans.get('answer', '')}\n"
prompt += "\n"
prompt += (
f"请生成最多 {max_questions} 个最关键的追问,帮助更准确地分析这个问题。\n\n"
f"要求:\n"
f"1. **追问必须与用户问题直接相关,不要偏离主题**。"
f"例如用户问'社保未缴全',追问应该是关于社保缴纳情况、欠缴金额、是否被查等,"
f"而不是问合同违约或股东纠纷等无关问题\n"
f"2. 每个问题针对一个关键事实维度(如{config['dim_examples']}等)\n"
f"3. 问题必须具体、有针对性,针对用户问题中的实际情境提问\n"
f"4. 不要问'能具体描述一下情况吗'这类泛泛的问题,要问具体的细节\n"
f"5. 优先问对{config['focus']}影响最大的事实\n"
f"6. 不要问用户已经回答过的事实\n"
f"7. 如果已有足够信息,返回空列表\n"
f"8. **所有追问必须从老板/雇主视角出发**(用户是中小企业老板,不是员工)。"
f"例如劳动用工场景,应问'公司与该员工签的劳动合同中解除条款怎么约定的',而非'您与公司签订的劳动合同'\n\n"
f"请以 JSON 数组格式返回,每个元素包含 dim_key(维度英文标识)、dim_desc(维度中文描述)、question(追问问题):\n"
f'[{{"dim_key": "amount", "dim_desc": "涉及金额", "question": "涉及的金额大约是多少?"}}]\n'
)
try:
answer = kimi_client.chat(
[{"role": "system", "content": "你是专业顾问助手,服务对象是中小企业老板。所有追问必须从老板/雇主视角出发,不要用员工视角提问。只返回 JSON 数组,不要其他内容。"},
{"role": "user", "content": prompt}],
temperature=0.3,
timeout=30,
)
# 解析 JSON(容错:提取第一个 JSON 数组)
answer = answer.strip()
# 去除可能的 markdown 代码块标记
if answer.startswith("```"):
answer = answer.split("\n", 1)[1] if "\n" in answer else answer[3:]
if answer.endswith("```"):
answer = answer[:-3]
# 修复中文引号(常见 LLM 输出问题)
answer = answer.replace("\u201c", '"').replace("\u201d", '"')
answer = answer.strip()
# 尝试提取 JSON 数组
start = answer.find("[")
end = answer.rfind("]")
if start == -1 or end == -1:
return []
json_str = answer[start:end + 1]
questions = json.loads(json_str)
# 校验格式
result = []
for q in questions[:max_questions]:
if isinstance(q, dict) and "question" in q:
result.append({
"dim_key": q.get("dim_key", f"ai_{len(result)}"),
"dim_desc": q.get("dim_desc", "补充信息"),
"question": q["question"],
})
return result
except Exception:
return []
def build_analysis_prompt(self, question: str, scene: str,
answers: list[dict],
kimi_client=None, history: list[dict] = None) -> tuple[str, dict]:
"""完整的追问流程:分析 → 生成追问 → 汇总事实。
优先用 AI 生成针对性追问,Kimi 不可用时降级到规则匹配。
如果不需要追问(事实已充分或已达最大轮次),返回汇总 prompt。
Args:
kimi_client: KimiClient 实例,传入时启用 AI 追问;None 时用规则匹配
Returns:
(prompt_or_questions, meta)
- 需要追问:prompt_or_questions = 追问问题列表, meta = {"need_ask": True, ...}
- 不需要追问:prompt_or_questions = 事实汇总 prompt, meta = {"need_ask": False, ...}
"""
analysis = self.analyze(question, scene)
# answers 格式: [{"dim_key", "dim_desc", "answer"}, ...]
round_num = len(answers) // self.max_questions_per_round + 1
if self.should_ask_more(analysis, round_num, answers,
kimi_client=kimi_client, question=question,
scene=scene):
# 优先 AI 生成追问
questions = []
if kimi_client:
questions = self.generate_ai_questions(
question, scene, kimi_client, answers, self.max_questions_per_round,
history=history,
)
# AI 不可用或返回空,降级到规则匹配
if not questions:
questions = self.generate_questions(analysis, round_num, answers)
if questions:
return questions, {
"need_ask": True,
"round": round_num,
"questions": questions,
"analysis": analysis,
"source": "ai" if kimi_client else "rule",
}
# 不需要追问,汇总事实
fact_summary = self.summarize_facts(question, analysis, answers)
return fact_summary, {
"need_ask": False,
"analysis": analysis,
"fact_summary": fact_summary,
}
class KimiClient:
"""Kimi/Moonshot OpenAI-compatible client. API key is read only from env.
支持超时重试和降级标记。调用方可通过 chat_with_fallback 获取推理位置。
"""
def __init__(self, api_key: str | None = None, model: str | None = None, base_url: str | None = None):
self.api_key = api_key or os.getenv("KIMI_API_KEY")
self.model = model or os.getenv("KIMI_MODEL", "moonshot-v1-8k")
self.base_url = (base_url or os.getenv("KIMI_BASE_URL", "https://api.moonshot.cn/v1")).rstrip("/")
if not self.api_key:
raise RuntimeError("KIMI_API_KEY is not configured")
def chat(self, messages: list[dict], temperature: float = 0.2, timeout: int = 60) -> str:
"""调用云端模型,超时 60s。"""
body = json.dumps({"model": self.model, "messages": messages, "temperature": temperature}).encode()
req = Request(self.base_url + "/chat/completions", data=body, headers={"Authorization": "Bearer " + self.api_key, "Content-Type": "application/json"})
with urlopen(req, timeout=timeout) as response:
payload = json.loads(response.read().decode())
return payload["choices"][0]["message"]["content"]
def chat_stream(self, messages: list[dict], temperature: float = 0.2, timeout: int = 120):
"""流式调用云端模型,逐 token 返回生成器。
Yields:
str: 每个 token 片段(delta.content)
Raises:
Exception: API 错误或超时
"""
body = json.dumps({
"model": self.model,
"messages": messages,
"temperature": temperature,
"stream": True,
}).encode()
req = Request(
self.base_url + "/chat/completions",
data=body,
headers={
"Authorization": "Bearer " + self.api_key,
"Content-Type": "application/json",
"Accept": "text/event-stream",
},
)
with urlopen(req, timeout=timeout) as response:
buffer = b""
for chunk in iter(lambda: response.read(1024), b""):
buffer += chunk
# SSE 事件以 \n\n 分隔
while b"\n\n" in buffer:
event_data, buffer = buffer.split(b"\n\n", 1)
for line in event_data.split(b"\n"):
line = line.strip()
if not line.startswith(b"data: "):
continue
data = line[6:]
if data == b"[DONE]":
return
try:
payload = json.loads(data)
delta = payload.get("choices", [{}])[0].get("delta", {})
content = delta.get("content")
if content:
yield content
except json.JSONDecodeError:
continue
def chat_with_fallback(self, messages: list[dict], local_fallback: str = "",
temperature: float = 0.2, retries: int = 1) -> dict:
"""调用云端模型,失败时降级到本地结论。
返回 {"answer": str, "inference_location": "cloud"|"local", "model": str, "fallback_used": bool}
"""
for attempt in range(retries + 1):
try:
answer = self.chat(messages, temperature=temperature)
return {"answer": answer, "inference_location": "cloud",
"model": self.model, "fallback_used": False}
except Exception as e:
if attempt < retries:
continue
# 降级到本地
return {"answer": local_fallback, "inference_location": "local",
"model": "local-triage", "fallback_used": True, "error": str(e)}
# 场景 system prompt 映射:按模块路由不同人设,前端通过 scene 参数选择
SCENE_PROMPTS: dict[str, str] = {
"legal": (
"你是中国中小企业老板的法务参谋,服务对象是企业老板(雇主),所有分析和建议必须从老板/雇主视角出发。"
"先给风险等级(绿色/黄色/橙色/红色),"
"再给结论、立即行动、需保存证据、禁止事项和是否需要律师复核。"
"覆盖劳动用工、合同审查、债权债务、股东纠纷、税务风险。"
"劳动用工场景中,你是帮老板处理员工问题的律师,不是帮员工维权的律师。"
"严禁编造法律依据;不确定时明确说明。"
),
"decision": (
"你是中国中小企业老板的经营决策助手。面对项目承接、扩张收缩、招人、降价、"
"业务线调整、融资和合伙等决策,采用情景推演:列出 2-3 个可行方案,"
"分别给出预期收益、主要风险、现金流影响和推荐程度(高/中/低)。"
"最后给出明确推荐和今天可执行的第一步。不替老板做最终决定,但要把利弊讲透。"
),
"life": (
"你是中国中小企业老板的生活助理,帮助降低老板的非经营决策负担。"
"覆盖家庭沟通、子女教育、父母养老、睡眠与压力、出差日程、健康习惯、商务礼仪。"
"回答务实、具体、可执行,给出今天能做的一件小事。避免说教,尊重老板的时间。"
),
"emotion": (
"你是老板的情绪价值引擎。根据用户选择的模式调整回复风格:"
"冷静分析模式:事实、风险、方案优先,克制客观;"
"陪伴疏导模式:先承接情绪,再给出下一步,温和不评判;"
"强力推动模式:把拖延事项拆成今天可完成的小动作,直接有力。"
"若检测到自伤、绝望等危机信号,立即提示联系心理援助热线 400-161-9995 或就医。"
),
}
def build_dynamic_prompt(scene: str, question: str, kimi_client=None,
fact_summary: str | None = None) -> str:
"""基于场景+问题动态生成定制 system prompt。
AI 根据用户具体问题调整 prompt 侧重,降级到 SCENE_PROMPTS。
Args:
scene: 场景(legal/decision/life/emotion)
question: 用户问题
kimi_client: KimiClient 实例
fact_summary: 已知事实摘要(多轮追问后)
Returns:
定制化 system prompt
"""
base_prompt = SCENE_PROMPTS.get(scene, SCENE_PROMPTS["legal"])
if not kimi_client:
return base_prompt
scene_desc = {
"legal": "法务参谋",
"decision": "经营决策助手",
"life": "生活助理",
"emotion": "情绪价值引擎",
}.get(scene, "顾问")
prompt = (
f"你是中国中小企业老板的{scene_desc}。以下是你的基础人设:\n{base_prompt}\n\n"
f"用户问题:{question}\n\n"
)
if fact_summary:
prompt += f"已知事实:\n{fact_summary}\n\n"
prompt += (
"请基于用户的具体问题,生成一段定制化的回复指引(50-100字),"
"明确本次回复应侧重哪些方面、采用什么结构、避免什么。\n"
"只返回指引文本,不要其他内容。"
)
try:
guidance = kimi_client.chat(
[{"role": "system", "content": "你是 prompt 工程师,根据用户问题定制回复指引。"},
{"role": "user", "content": prompt}],
temperature=0.3,
timeout=10,
)
return base_prompt + "\n\n本次回复指引:" + guidance.strip()
except Exception:
return base_prompt
# 情绪模式前缀,拼接到 emotion 场景 prompt 前
EMOTION_MODE_PREFIX: dict[str, str] = {
"calm": "当前模式:冷静分析。",
"companion": "当前模式:陪伴疏导。",
"push": "当前模式:强力推动。",
}
def build_emotion_prefix(emotion_mode: str, user_message: str = "",
kimi_client=None) -> str:
"""根据用户首条消息动态生成情绪模式前缀。
AI 根据用户情绪强度调整语气,降级到固定 EMOTION_MODE_PREFIX。
Args:
emotion_mode: 情绪模式(calm/companion/push)
user_message: 用户首条消息(用于感知情绪强度)
kimi_client: KimiClient 实例
Returns:
情绪模式前缀文本
"""
base_prefix = EMOTION_MODE_PREFIX.get(emotion_mode, "")
if not kimi_client or not user_message:
return base_prefix
mode_desc = {
"calm": "冷静分析(事实、风险、方案优先,克制客观)",
"companion": "陪伴疏导(先承接情绪,再给出下一步,温和不评判)",
"push": "强力推动(把拖延拆成今天可完成的小动作,直接有力)",
}.get(emotion_mode, "")
prompt = (
f"用户选择了「{mode_desc}」模式。\n"
f"用户消息:{user_message}\n\n"
"请生成一段 30-50 字的模式前缀指令,根据用户当前情绪状态调整回复语气。\n"
"只返回指令文本,不要其他内容。"
)
try:
prefix = kimi_client.chat(
[{"role": "system", "content": "你是情绪对话风格调节助手。"},
{"role": "user", "content": prompt}],
temperature=0.4,
timeout=10,
)
return f"当前模式:{mode_desc.split('(')[0]}。{prefix.strip()}"
except Exception:
return base_prefix
def extract_profile_signals(question: str, scene: str, kimi_client=None,
fact_answers: list[dict] | None = None) -> dict:
"""AI 从用户问题中提取抽象画像信号(非对话内容)。
隐私原则:只提取用户特征(行业/偏好/话题方向),不提取具体问题内容。
即使阅后即焚模式也调用此方法,画像信号不包含敏感内容。
Returns:
{
"industry": "零售/制造/科技/服务/...",
"company_size": "micro/small/medium/large",
"role": "owner/manager/ceo/...",
"common_topics": ["劳动用工", "扩张", ...],
"risk_tolerance": "conservative/moderate/aggressive",
"communication_style": "concise/detailed/structured",
"decision_style": "data-driven/intuitive/collaborative",
"key_facts": {"company_type": "有限公司", ...},
}
AI 不可用时返回空 dict。
"""
if not kimi_client:
return {}
scene_desc = {
"legal": "法律/法务",
"decision": "经营决策",
"life": "生活助理",
"emotion": "情绪疏导",
}.get(scene, "通用")
prompt = (
f"你是用户画像分析专家。从用户的{scene_desc}问题中提取抽象用户特征。\n\n"
f"用户问题:{question}\n\n"
)
if fact_answers:
prompt += "用户补充的事实:\n"
for ans in fact_answers:
prompt += f"- {ans.get('dim_desc', '')}:{ans.get('answer', '')}\n"
prompt += "\n"
prompt += (
"请提取以下抽象特征(只提取能从问题中合理推断的,不确定的留空):\n"
"- industry: 行业(如 零售/制造/科技/服务/餐饮/物流/教育/医疗)\n"
"- company_size: 企业规模(micro<10人/small<50/medium<200/large>200)\n"
"- role: 用户角色(如 owner/manager/ceo/director)\n"
"- common_topics: 常问话题(1-3个,如 劳动用工/扩张/合同审查/税务/融资)\n"
"- risk_tolerance: 风险偏好(conservative/moderate/aggressive)\n"
"- communication_style: 沟通偏好(concise/detailed/structured)\n"
"- decision_style: 决策风格(data-driven/intuitive/collaborative)\n"
"- key_facts: 已知关键事实(如 {\"company_type\": \"有限公司\", \"region\": \"上海\"})\n\n"
"只返回 JSON,不要其他内容。不确定的字段留空字符串或空数组。\n"
'{"industry": "", "company_size": "", "role": "", "common_topics": [], '
'"risk_tolerance": "", "communication_style": "", "decision_style": "", "key_facts": {}}'
)
try:
answer = kimi_client.chat(
[{"role": "system", "content": "你是用户画像分析助手,只返回 JSON。"},
{"role": "user", "content": prompt}],
temperature=0.1,
timeout=10,
)
answer = answer.strip()
if answer.startswith("```"):
answer = answer.split("\n", 1)[1] if "\n" in answer else answer[3:]
if answer.endswith("```"):
answer = answer[:-3]
answer = answer.replace("\u201c", '"').replace("\u201d", '"').strip()
start = answer.find("{")
end = answer.rfind("}")
if start == -1 or end == -1:
return {}
data = json.loads(answer[start:end + 1])
# 过滤空值
return {k: v for k, v in data.items() if v}
except Exception:
return {}
def build_profile_context(profile: UserProfile) -> str:
"""将用户画像转为 prompt 注入文本。
Args:
profile: 用户画像
Returns:
画像上下文文本,无画像时返回空字符串
"""
return profile.to_context()
# ===== 对话历史落库 =====
class CryptoUtil:
"""消息加密工具:保密模式用系统密钥加密存储,查看历史时用登录密码做二次验证。
使用 AES-256-GCM(Fernet)加密,密钥从环境变量 MESSAGE_ENCRYPTION_KEY 派生。
生产环境必须设置该环境变量;开发环境自动生成临时密钥(重启后旧密文不可解密)。
"""
def __init__(self):
self._fernet = None
self._is_default_key = False
if Fernet is None:
# cryptography 未安装,降级为不加密(开发环境)
return
key = os.getenv("MESSAGE_ENCRYPTION_KEY")
if not key:
# 开发环境:基于固定种子生成确定性密钥,避免每次重启丢失
key = base64.urlsafe_b64encode(
hashlib.sha256(b"bossai-dev-encryption-key").digest()
).decode()
self._is_default_key = True
self._fernet = Fernet(key.encode() if isinstance(key, str) else key)
def encrypt(self, plaintext: str) -> str:
"""加密明文,返回 base64 密文。cryptography 缺失时降级为明文。"""
if self._fernet is None:
return plaintext
return self._fernet.encrypt(plaintext.encode("utf-8")).decode("utf-8")
def decrypt(self, ciphertext: str) -> str:
"""解密密文,返回明文。cryptography 缺失时直接返回原文。"""
if self._fernet is None:
return ciphertext
return self._fernet.decrypt(ciphertext.encode("utf-8")).decode("utf-8")
class ChatStore:
"""PostgreSQL 对话历史存储。
- 普通模式(normal):明文存储消息+风险卡+元数据,可直接查看历史。
- 保密模式(confidential):加密存储消息正文,查看历史需登录密码二次验证后解密。
- 阅后即焚(ephemeral):不落库,仅内存临时会话。
"""
def __init__(self, dsn: str | None = None, crypto: "CryptoUtil | None" = None):
self.dsn = dsn or os.getenv(
"DATABASE_URL", "host=localhost port=5432 dbname=mcp user=postgres"
)
self.crypto = crypto or CryptoUtil()
def _conn(self):
return psycopg2.connect(self.dsn)
def save_message(
self,
session_id: str,
phone: str,
mode: str,
role: str,
content: str,
scene: str = "",
risk_level: str = "",
risk_conclusion: str = "",
risk_actions: list | None = None,
risk_evidence: list | None = None,
risk_escalation: bool = False,
inference_location: str = "",
model_name: str = "",
fallback_used: bool = False,
rag_context: list | None = None,
) -> int | None:
"""保存一条对话消息。
Args:
session_id: 会话 ID
phone: 用户手机号
mode: 隐私模式(normal/confidential/ephemeral)
role: 消息角色(user/assistant)
content: 消息正文
scene: 场景(legal/decision/life/emotion)
risk_level: 风险等级(green/amber/orange/red)
risk_conclusion: 风险结论
risk_actions: 行动建议列表
risk_evidence: 证据清单列表
risk_escalation: 是否建议升级
inference_location: 推理位置(cloud/local)
model_name: 模型名称
fallback_used: 是否使用了降级
rag_context: 知识库引用文档列表(仅 assistant 消息)
Returns:
消息 ID,ephemeral 模式返回 None
"""
if mode == "ephemeral":
return None # 阅后即焚不落库
# 保密模式加密消息正文和风险结论
if mode == "confidential":
stored_content = self.crypto.encrypt(content)
stored_risk_conclusion = self.crypto.encrypt(risk_conclusion) if risk_conclusion else ""
else:
stored_content = content
stored_risk_conclusion = risk_conclusion
risk_data = json.dumps({
"level": risk_level,
"conclusion": stored_risk_conclusion,
"actions": risk_actions or [],
"evidence": risk_evidence or [],
"escalation": risk_escalation,
}, ensure_ascii=False)
# 知识库引用摘要(只存标题/类型/地区,不存全文)
rag_summary = []
if rag_context:
for doc in rag_context:
rag_summary.append({
"title": doc.get("title", ""),
"doc_type": doc.get("doc_type", ""),
"region": doc.get("region"),
})
meta_data = json.dumps({
"scene": scene,
"inference_location": inference_location,
"model": model_name,
"fallback_used": fallback_used,
"rag_context": rag_summary,
}, ensure_ascii=False)
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""INSERT INTO chat_messages
(session_id, phone, mode, role, content, risk_data, meta_data, created_at)
VALUES (%s, %s, %s, %s, %s, %s, %s, now())
RETURNING id""",
(session_id, phone, mode, role, stored_content, risk_data, meta_data),
)
row = cur.fetchone()
return row[0] if row else None
def list_sessions(self, phone: str, mode: str | None = None) -> list:
"""列出用户的会话(按时间倒序,不含消息正文)。
Args:
phone: 用户手机号
mode: 可选模式过滤
Returns:
会话摘要列表 [{session_id, mode, scene, message_count, last_message_at, first_message_at}]
"""
query = """
SELECT session_id, mode,
COUNT(*) AS msg_count,
MIN(created_at) AS first_at,
MAX(created_at) AS last_at,
(array_agg(meta_data->>'scene' ORDER BY created_at DESC))[1] AS scene,
(array_agg(content ORDER BY created_at ASC) FILTER (WHERE role = 'user'))[1] AS first_question
FROM chat_messages
WHERE phone = %s
"""
params: list = [phone]
if mode:
query += " AND mode = %s"
params.append(mode)
query += " GROUP BY session_id, mode ORDER BY last_at DESC"
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(query, params)
rows = cur.fetchall()
# 场景中文映射(保密模式无法解密内容,用场景标签生成标题)
SCENE_LABELS = {"legal": "法务参谋", "decision": "经营决策",
"life": "生活助理", "emotion": "情绪中心"}
results = []
for r in rows:
sid, smode, msg_count, first_at, last_at, scene, first_q = r
if smode == "confidential":
# 保密模式:内容已加密,用场景标签 + 日期生成标题
scene_label = SCENE_LABELS.get(scene or "", "保密对话")
date_str = first_at.strftime("%m/%d") if first_at else ""
title = f"{scene_label} · {date_str}" if date_str else scene_label
else:
# 普通模式:取首条用户消息前 30 字作为标题
title = (first_q[:30] + "…") if first_q and len(first_q) > 30 else (first_q or "未命名对话")
results.append({
"session_id": sid,
"mode": smode,
"message_count": msg_count,
"first_message_at": first_at.isoformat() if first_at else None,
"last_message_at": last_at.isoformat() if last_at else None,
"scene": scene,
"title": title,
})
return results
def get_messages(self, session_id: str, phone: str, decrypt: bool = True) -> list:
"""获取会话的所有消息(按时间正序)。
Args:
session_id: 会话 ID
phone: 用户手机号(权限校验)
decrypt: 是否解密保密模式消息(查看历史时传 True)
Returns:
消息列表 [{id, role, content, mode, risk_data, meta_data, created_at}]
"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"""SELECT id, role, content, mode, risk_data, meta_data, created_at
FROM chat_messages
WHERE session_id = %s AND phone = %s
ORDER BY created_at ASC""",
(session_id, phone),
)
rows = cur.fetchall()
messages = []
for r in rows:
msg_id, role, content, msg_mode, risk_data, meta_data, created_at = r
risk = risk_data if isinstance(risk_data, dict) else (json.loads(risk_data) if risk_data else {})
meta = meta_data if isinstance(meta_data, dict) else (json.loads(meta_data) if meta_data else {})
# 保密模式解密
if msg_mode == "confidential" and decrypt:
try:
content = self.crypto.decrypt(content)
if risk.get("conclusion"):
risk["conclusion"] = self.crypto.decrypt(risk["conclusion"])
except Exception:
content = "[解密失败]"
risk["conclusion"] = "[解密失败]"
messages.append({
"id": msg_id,
"role": role,
"content": content,
"mode": msg_mode,
"risk": risk,
"meta": meta,
"created_at": created_at.isoformat() if created_at else None,
})
return messages
def delete_session(self, session_id: str, phone: str) -> int:
"""删除会话的所有消息(用户主动删除历史)。"""
with self._conn() as conn:
with conn.cursor() as cur:
cur.execute(
"DELETE FROM chat_messages WHERE session_id = %s AND phone = %s",
(session_id, phone),
)
return cur.rowcount