433 lines
20 KiB
Python
433 lines
20 KiB
Python
import base64
|
||
import json
|
||
import uuid
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
|
||
from fastapi import APIRouter, Depends, HTTPException, UploadFile
|
||
from fastapi.responses import StreamingResponse
|
||
from sqlalchemy import delete, select
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.config import get_settings
|
||
from app.db import SessionLocal, get_db
|
||
from app.deps import get_current_user
|
||
from app.integrations.adp_client import AdpClient
|
||
from app.integrations.storage_client import StorageClient
|
||
from app.models import Attachment, Conversation, Feedback, Message, MessageReference, User
|
||
from app.schemas.chat import (
|
||
AttachmentResponse,
|
||
ChatStreamRequest,
|
||
ConversationCreate,
|
||
ConversationResponse,
|
||
FeedbackCreate,
|
||
MessageResponse,
|
||
)
|
||
from app.services.audit_service import write_audit
|
||
from app.services.document_service import build_unit_chain, resolve_chat_doc_scope
|
||
from app.services.user_service import JOB_TITLE_GROUP_ALIASES
|
||
|
||
|
||
def _attachment_data_url(att: Attachment) -> str | None:
|
||
storage_dir = Path(get_settings().local_storage_dir)
|
||
p = storage_dir / att.storage_key
|
||
if not p.exists():
|
||
return None
|
||
try:
|
||
raw = p.read_bytes()
|
||
except OSError:
|
||
return None
|
||
mime = att.file_type or "image/png"
|
||
return f"data:{mime};base64,{base64.b64encode(raw).decode()}"
|
||
|
||
router = APIRouter()
|
||
|
||
|
||
@router.get("/sessions", response_model=list[ConversationResponse])
|
||
def list_sessions(current_user: User = Depends(get_current_user), db: Session = Depends(get_db)) -> list[Conversation]:
|
||
return db.scalars(
|
||
select(Conversation).where(Conversation.user_id == current_user.id).order_by(Conversation.id.desc())
|
||
).all()
|
||
|
||
|
||
@router.post("/sessions", response_model=ConversationResponse)
|
||
def create_session(
|
||
payload: ConversationCreate,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> Conversation:
|
||
conversation = Conversation(user_id=current_user.id, title=payload.title)
|
||
db.add(conversation)
|
||
write_audit(db, "chat.session.create", actor_id=current_user.id, target_type="conversation")
|
||
db.commit()
|
||
db.refresh(conversation)
|
||
return conversation
|
||
|
||
|
||
@router.delete("/sessions/{session_id}")
|
||
def delete_session(
|
||
session_id: int,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
conversation = db.get(Conversation, session_id)
|
||
if not conversation or conversation.user_id != current_user.id:
|
||
raise HTTPException(status_code=404, detail="会话不存在")
|
||
message_ids = list(db.scalars(select(Message.id).where(Message.conversation_id == session_id)))
|
||
if message_ids:
|
||
db.execute(delete(MessageReference).where(MessageReference.message_id.in_(message_ids)))
|
||
db.execute(delete(Feedback).where(Feedback.message_id.in_(message_ids)))
|
||
db.execute(delete(Attachment).where(Attachment.message_id.in_(message_ids)))
|
||
db.execute(delete(Message).where(Message.conversation_id == session_id))
|
||
db.delete(conversation)
|
||
write_audit(db, "chat.session.delete", actor_id=current_user.id, target_type="conversation", target_id=str(session_id))
|
||
db.commit()
|
||
return {"message": "会话已删除"}
|
||
|
||
|
||
@router.get("/sessions/{session_id}/messages", response_model=list[MessageResponse])
|
||
def list_messages(
|
||
session_id: int,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> list[MessageResponse]:
|
||
conversation = db.get(Conversation, session_id)
|
||
if not conversation or conversation.user_id != current_user.id:
|
||
raise HTTPException(status_code=404, detail="会话不存在")
|
||
messages = db.scalars(select(Message).where(Message.conversation_id == session_id).order_by(Message.id)).all()
|
||
message_ids = [m.id for m in messages]
|
||
atts_by_msg: dict[int, list[Attachment]] = {}
|
||
if message_ids:
|
||
for att in db.scalars(select(Attachment).where(Attachment.message_id.in_(message_ids))):
|
||
atts_by_msg.setdefault(att.message_id, []).append(att)
|
||
return [
|
||
MessageResponse(
|
||
id=message.id,
|
||
conversation_id=message.conversation_id,
|
||
question=message.question,
|
||
answer=message.answer,
|
||
status=message.status,
|
||
token_usage=message.token_usage,
|
||
pu_usage=message.pu_usage,
|
||
created_at=message.created_at,
|
||
references=_dedup_refs(message.references),
|
||
attachments=[
|
||
AttachmentResponse(id=att.id, file_type=att.file_type, data_url=_attachment_data_url(att))
|
||
for att in atts_by_msg.get(message.id, [])
|
||
],
|
||
)
|
||
for message in messages
|
||
]
|
||
|
||
|
||
def _dedup_refs(refs) -> list[dict]:
|
||
seen: set[tuple[str, str]] = set()
|
||
out: list[dict] = []
|
||
for ref in refs:
|
||
title = (ref.source_title or "").strip()
|
||
if not title:
|
||
continue
|
||
key = (title, (ref.snippet or "").strip())
|
||
if key in seen:
|
||
continue
|
||
seen.add(key)
|
||
out.append({
|
||
"source_title": ref.source_title,
|
||
"snippet": ref.snippet,
|
||
"page_no": ref.page_no,
|
||
"section": ref.section,
|
||
"source_url": ref.source_url,
|
||
})
|
||
return out
|
||
|
||
|
||
@router.post("/upload-image")
|
||
def upload_image(file: UploadFile, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)) -> dict:
|
||
if file.content_type not in {"image/png", "image/jpeg", "image/jpg"}:
|
||
raise HTTPException(status_code=400, detail="仅支持 PNG/JPG 截图")
|
||
raw = file.file.read()
|
||
storage = StorageClient()
|
||
key = storage.save_bytes(raw, suffix=Path(file.filename or "upload.bin").suffix)
|
||
try:
|
||
remote_url = AdpClient().upload_chat_image(
|
||
raw, mime_type=file.content_type or "image/png", original_name=file.filename or ""
|
||
)
|
||
except Exception as exc:
|
||
raise HTTPException(status_code=502, detail=f"图片上传到 ADP 失败: {exc}") from exc
|
||
attachment = Attachment(
|
||
user_id=current_user.id,
|
||
file_type=file.content_type or "image",
|
||
storage_key=key,
|
||
remote_url=remote_url,
|
||
original_name=file.filename,
|
||
)
|
||
db.add(attachment)
|
||
db.flush()
|
||
write_audit(db, "chat.image.upload", actor_id=current_user.id, target_type="attachment", target_id=str(attachment.id))
|
||
db.commit()
|
||
return {"id": attachment.id, "url": storage.temporary_url(key)}
|
||
|
||
|
||
@router.post("/stream")
|
||
def stream_chat(
|
||
payload: ChatStreamRequest,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> StreamingResponse:
|
||
current_user_id = current_user.id
|
||
is_admin = current_user.is_superuser or any(
|
||
role.code in {"admin", "system_admin"} for role in current_user.roles
|
||
)
|
||
# ADP 检索层过滤:角色码 + 提问人 unit_chain(整条向上链,支持就近优先/上级回退) + 公开范围标签.
|
||
# 只传角色码时,其他单位的专属文档也会被 ADP 一并召回喂给 LLM,LLM 会在正文写"依据文档:《XX 单位规定》";
|
||
# 把整条链一起下发,是为了让本单位无对应制度时,上级单位的制度也在候选内,由 LLM 按"就近优先"选用.
|
||
# 兄弟单位(不在向上链条上的)始终不进 allowed_tags,不会被召回.超管不过滤.
|
||
unit_chain_list = build_unit_chain(current_user) if not current_user.is_superuser else []
|
||
if current_user.is_superuser:
|
||
allowed_tags: list[str] = []
|
||
else:
|
||
allowed_tags = [role.code for role in current_user.roles]
|
||
allowed_tags.extend(unit_chain_list)
|
||
allowed_tags.extend(["所有人可见", "通用", "全体员工"])
|
||
|
||
# role_hint 必须是业务岗位（如"总院中层副职""员工"），用来匹配制度里按职级分组的条款。
|
||
# 只取 User.job_title；权限角色（employee/admin/system_admin）和 is_superuser 不参与,
|
||
# 否则会把"系统管理员"塞进制度职级匹配, LLM 找不到对应分组就反问澄清。
|
||
role_hint = (current_user.job_title or "").strip()
|
||
|
||
# 按用户所属单位（含逐级向上回退）计算允许引用的文档 title 集合。
|
||
# 超管/管理员或用户没绑定单位 -> 不做过滤（allowed_titles=None）。
|
||
scope = {"allowed_titles": None, "own_unit": "", "chosen_unit": "", "fell_back": False, "public_only": False}
|
||
if not is_admin and current_user.unit is not None:
|
||
try:
|
||
adp_docs = (AdpClient().list_documents(page=1, page_size=200).get("list") or [])
|
||
scope = resolve_chat_doc_scope(current_user, adp_docs)
|
||
except Exception:
|
||
# ADP 拉取失败时不阻断问答，退化为不过滤
|
||
scope = {"allowed_titles": None, "own_unit": (current_user.unit.name or "").strip(),
|
||
"chosen_unit": "", "fell_back": False, "public_only": False}
|
||
elif current_user.unit is not None:
|
||
scope["own_unit"] = (current_user.unit.name or "").strip()
|
||
|
||
allowed_titles = scope["allowed_titles"]
|
||
|
||
if not (payload.question or "").strip() and not payload.attachment_ids:
|
||
raise HTTPException(status_code=400, detail="问题文本和图片至少需要一项")
|
||
|
||
image_urls: list[str] = []
|
||
attachments: list[Attachment] = []
|
||
if payload.attachment_ids:
|
||
attachments = list(db.scalars(
|
||
select(Attachment).where(
|
||
Attachment.id.in_(payload.attachment_ids),
|
||
Attachment.user_id == current_user_id,
|
||
)
|
||
))
|
||
if len(attachments) != len(payload.attachment_ids):
|
||
raise HTTPException(status_code=404, detail="附件不存在或无权访问")
|
||
for att in attachments:
|
||
if not att.remote_url:
|
||
raise HTTPException(status_code=500, detail=f"附件缺少远端 URL，请重新上传: {att.id}")
|
||
image_urls.append(att.remote_url)
|
||
|
||
conversation = db.get(Conversation, payload.conversation_id) if payload.conversation_id else None
|
||
if conversation and conversation.user_id != current_user_id:
|
||
raise HTTPException(status_code=404, detail="会话不存在")
|
||
if not conversation:
|
||
title_seed = (payload.question or "").strip()[:40] or ("[图片] " + (attachments[0].original_name or "提问"))
|
||
conversation = Conversation(
|
||
user_id=current_user_id,
|
||
title=title_seed[:60],
|
||
adp_conversation_id=uuid.uuid4().hex,
|
||
)
|
||
db.add(conversation)
|
||
db.flush()
|
||
elif not conversation.adp_conversation_id:
|
||
conversation.adp_conversation_id = uuid.uuid4().hex
|
||
db.flush()
|
||
|
||
message = Message(conversation_id=conversation.id, user_id=current_user_id, question=payload.question, status="streaming")
|
||
db.add(message)
|
||
conversation.last_message_at = datetime.utcnow()
|
||
db.flush()
|
||
message_id = message.id
|
||
conversation_id = conversation.id
|
||
adp_conversation_id = conversation.adp_conversation_id
|
||
for att in attachments:
|
||
att.message_id = message_id
|
||
db.commit()
|
||
|
||
raw_question = payload.question or ""
|
||
hint_lines: list[str] = []
|
||
# 引用格式规则始终注入,与提问人职位无关。
|
||
# finance_assistant.md 系统 prompt 里写着"引用由系统自动展示,不用手写",这里覆盖它;
|
||
# 独立成条防止 job_title=None 的账号(如系统管理员)拿不到规则,导致回答漏"依据文档"。
|
||
hint_lines.append(
|
||
"【引用格式】不要在正文开头写\"根据《XX规定》...\"这类引用开场;"
|
||
"把制度出处放在整段回答的最末尾,另起一行以\"依据文档:《文档全名》(文号,如有)\"的格式列出;"
|
||
"引用多份文档时每份单独一行。"
|
||
"此规则最高优先级:即使命中 FAQ/标准答案,或已在正文其他位置提到制度名,末尾仍必须补上此行;"
|
||
"若知识库无对应制度可引(纯操作步骤、报错处置、闲聊等),则写\"依据文档:无\"。"
|
||
)
|
||
if role_hint:
|
||
aliases = JOB_TITLE_GROUP_ALIASES.get(role_hint, [role_hint])
|
||
alias_list = "、".join(f"\"{a}\"" for a in aliases)
|
||
hint_lines.append(
|
||
f"【提问人职位:{role_hint}】请严格遵守以下输出规则:\n"
|
||
f"1) 若制度按职级/岗位分组列条款(如\"总院领导班子正职\"\"总院中层副职\"\"总助级\"等),"
|
||
f"只输出分组名中包含以下任一关键词的分组细则: {alias_list};不匹配的分组一律不要列出,也不要用于对比。\n"
|
||
f"2) 若制度不按职级分组,或该职位不涉及相关业务,请直接输出通用条款或明确说明\"该职位无相关规定\"。"
|
||
)
|
||
own_unit = scope.get("own_unit") or ""
|
||
chosen_unit = scope.get("chosen_unit") or ""
|
||
if own_unit and not is_admin:
|
||
if scope.get("fell_back") and chosen_unit:
|
||
hint_lines.append(
|
||
f"【提问人所属单位:{own_unit}】"
|
||
f"该单位无直接对应制度,请引用其上级单位「{chosen_unit}」的对应制度以及公开制度作答。"
|
||
)
|
||
elif chosen_unit:
|
||
hint_lines.append(
|
||
f"【提问人所属单位:{own_unit}】"
|
||
f"请优先引用本单位对应制度及公开制度作答,不要引用其他单位专属制度。"
|
||
)
|
||
elif scope.get("public_only"):
|
||
hint_lines.append(
|
||
f"【提问人所属单位:{own_unit}】"
|
||
f"当前单位链上均无对应制度,仅可引用公开制度作答,若信息不足请明确说明。"
|
||
)
|
||
# 硬约束:兜底 ADP 检索层过滤失效 -> 强制 LLM 拒用 unit_chain 之外单位的专属文档.
|
||
# 允许 unit_chain 内的上级单位文档(就近优先由上面 fell_back/chosen_unit 分支引导).
|
||
# 优先级最高,凌驾于 FAQ 命中等所有规则.命中 FAQ 但 FAQ 内容显然只覆盖链外单位时,也必须拒答.
|
||
allowed_units_str = "、".join(f"「{u}」" for u in unit_chain_list) if unit_chain_list else f"「{own_unit}」"
|
||
hint_lines.append(
|
||
f"【单位隔离硬约束·最高优先级】提问人所属单位:{own_unit}。"
|
||
f"允许的单位标签白名单(精确匹配,禁止子串/字面前缀扩展):{allowed_units_str}。"
|
||
f"允许引用的文档必须满足以下之一:"
|
||
f"(a) 标签值精确等于白名单中任一单位名;"
|
||
f"(b) 标签值为'通用/全体员工/所有人可见'等公开范围。"
|
||
f"特别注意:标签名之间的字面前缀相同不代表存在上下级关系。"
|
||
f"例如'中国建材总院母公司'与'中国建材总院北京分公司'虽同以'中国建材总院'开头,"
|
||
f"但两者是平级独立单位,非上下级关系,禁止互相引用对方专属文档。"
|
||
f"不满足则:"
|
||
f"(1) 禁止将其内容用于作答,禁止在'依据文档'中列出;"
|
||
f"(2) 若召回文档全部不满足(即白名单单位及公开范围均无对应制度),必须回复原文:"
|
||
f"'知识库中未找到 {own_unit} 及其上级单位对应的规定,建议咨询本单位财务部门。';"
|
||
f"(3) 此规则覆盖'FAQ 命中即返回标准答案'的短路,即使命中 FAQ 也须先判定单位适用性。"
|
||
)
|
||
if hint_lines:
|
||
adp_question = (
|
||
f"{raw_question}\n\n"
|
||
"---以下为回答约束,仅供生成时参考,不参与知识检索匹配---\n"
|
||
+ "\n".join(hint_lines)
|
||
)
|
||
else:
|
||
adp_question = raw_question
|
||
# 兜底:allowed_titles 是空集(public_only 且公开也无匹配) -> 单位链上和公开都没文档可用.
|
||
# 直接返回硬约束里那句拒答话术, 不再走 ADP 生成, 避免 LLM 自行发挥引用兄弟单位文档.
|
||
refuse_directly = (
|
||
not is_admin
|
||
and own_unit
|
||
and isinstance(allowed_titles, set)
|
||
and len(allowed_titles) == 0
|
||
)
|
||
refuse_text = (
|
||
f"知识库中未找到 {own_unit} 及其上级单位对应的规定,建议咨询本单位财务部门。"
|
||
if refuse_directly else ""
|
||
)
|
||
|
||
def event_stream():
|
||
answer_parts: list[str] = []
|
||
usage = {"token_usage": 0, "pu_usage": 0}
|
||
references: list[dict] = []
|
||
seen_refs: set[tuple[str, str]] = set()
|
||
stream_start = datetime.utcnow()
|
||
failed = False
|
||
error_message: str | None = None
|
||
if refuse_directly:
|
||
answer_parts.append(refuse_text)
|
||
yield f"event: answer.delta\ndata: {json.dumps(refuse_text, ensure_ascii=False)}\n\n"
|
||
yield f"event: answer.usage\ndata: {json.dumps(usage, ensure_ascii=False)}\n\n"
|
||
yield f"event: answer.done\ndata: {json.dumps({}, ensure_ascii=False)}\n\n"
|
||
try:
|
||
if not refuse_directly:
|
||
for event in AdpClient().stream_answer(
|
||
adp_question,
|
||
user_id=str(current_user_id),
|
||
conversation_id=adp_conversation_id,
|
||
allowed_tags=allowed_tags,
|
||
unit_name=own_unit,
|
||
unit_chain=unit_chain_list,
|
||
image_urls=image_urls,
|
||
):
|
||
event_name = event["event"]
|
||
if event_name == "answer.delta":
|
||
answer_parts.append(event["data"])
|
||
elif event_name == "answer.reference":
|
||
title = (event["data"].get("source_title") or "").strip()
|
||
if not title:
|
||
continue
|
||
# 按用户所属单位链过滤引用；allowed_titles=None 时不过滤
|
||
if allowed_titles is not None and title not in allowed_titles:
|
||
continue
|
||
key = (title, (event["data"].get("snippet") or "").strip())
|
||
if key in seen_refs:
|
||
continue
|
||
seen_refs.add(key)
|
||
references.append(event["data"])
|
||
elif event_name == "answer.usage":
|
||
usage.update(event["data"])
|
||
yield f"event: {event_name}\ndata: {json.dumps(event['data'], ensure_ascii=False)}\n\n"
|
||
except Exception as exc: # noqa: BLE001
|
||
failed = True
|
||
error_message = str(exc) or exc.__class__.__name__
|
||
yield f"event: answer.error\ndata: {json.dumps({'message': error_message}, ensure_ascii=False)}\n\n"
|
||
|
||
latency_ms = int((datetime.utcnow() - stream_start).total_seconds() * 1000)
|
||
save_db = SessionLocal()
|
||
try:
|
||
saved = save_db.get(Message, message_id)
|
||
if saved:
|
||
saved.answer = "".join(answer_parts)
|
||
saved.status = "failed" if failed else "done"
|
||
saved.token_usage = usage["token_usage"]
|
||
saved.pu_usage = usage["pu_usage"]
|
||
saved.latency_ms = latency_ms
|
||
if not failed:
|
||
for ref in references:
|
||
save_db.add(MessageReference(message_id=message_id, **ref))
|
||
write_audit(
|
||
save_db,
|
||
"chat.ask.failed" if failed else "chat.ask",
|
||
actor_id=current_user_id,
|
||
target_type="message",
|
||
target_id=str(message_id),
|
||
detail={"error": error_message} if failed and error_message else None,
|
||
)
|
||
save_db.commit()
|
||
finally:
|
||
save_db.close()
|
||
|
||
return StreamingResponse(
|
||
event_stream(),
|
||
media_type="text/event-stream",
|
||
headers={"X-Conversation-Id": str(conversation_id), "X-Message-Id": str(message_id)},
|
||
)
|
||
|
||
|
||
@router.post("/messages/{message_id}/feedback")
|
||
def create_feedback(
|
||
message_id: int,
|
||
payload: FeedbackCreate,
|
||
current_user: User = Depends(get_current_user),
|
||
db: Session = Depends(get_db),
|
||
) -> dict:
|
||
message = db.get(Message, message_id)
|
||
if not message or message.user_id != current_user.id:
|
||
raise HTTPException(status_code=404, detail="消息不存在")
|
||
feedback = Feedback(message_id=message_id, user_id=current_user.id, rating=payload.rating, comment=payload.comment)
|
||
db.add(feedback)
|
||
write_audit(db, "chat.feedback", actor_id=current_user.id, target_type="message", target_id=str(message_id))
|
||
db.commit()
|
||
return {"message": "反馈已提交"}
|