Compare commits

...
14 changed files with 573 additions and 81 deletions
+1
View File
@@ -53,6 +53,7 @@ def init_db():
("knowledge_docs", "embedding_model", "VARCHAR DEFAULT ''"),
("knowledge_docs", "chunk_count", "INTEGER DEFAULT 0"),
("knowledge_docs", "vectorized_at", "TIMESTAMP"),
("knowledge_docs", "error_message", "VARCHAR DEFAULT ''"),
("avatars", "owner_id", "VARCHAR DEFAULT ''"),
("authorizations", "takeover_enabled", "BOOLEAN DEFAULT 0"),
("authorizations", "takeover_mode", "VARCHAR DEFAULT 'immediate'"),
+2
View File
@@ -20,6 +20,7 @@ import routers.chat
import routers.takeover
from responses import ok
from services.chat_attachment_service import purge_expired_chat_attachments
from services.knowledge_vectorizer import knowledge_vectorizer
from services.token_billing import DEFAULT_TOKEN_GRANT, release_stale_reservations
logger = logging.getLogger(__name__)
@@ -131,6 +132,7 @@ def on_startup():
init_db()
seed()
knowledge_vectorizer.start()
# Release stale resources when startup is invoked again by a reload/test.
stop_takeover_scheduler()
+2
View File
@@ -190,6 +190,7 @@ class KnowledgeDoc(Base):
file_size = Column(Integer, default=0)
file_url = Column(String, default="")
status = Column(String, default="uploaded") # uploaded | parsing | ready | failed
error_message = Column(String, default="") # 建立索引失败原因
vectorized = Column(Boolean, default=False) # 是否已向量化
embedding_model = Column(String, default="") # 向量模型标识
chunk_count = Column(Integer, default=0) # 切片数量
@@ -205,6 +206,7 @@ class KnowledgeDoc(Base):
"fileSize": self.file_size,
"fileUrl": self.file_url,
"status": self.status,
"errorMessage": self.error_message or "",
"vectorized": bool(self.vectorized),
"embeddingModel": self.embedding_model,
"chunkCount": self.chunk_count,
+99 -1
View File
@@ -47,6 +47,25 @@ QA_SEMANTIC_THRESHOLD = 0.72
QA_MATCH_MARGIN = 0.06
KNOWLEDGE_MIN_SCORE = float(os.getenv("KNOWLEDGE_MIN_SCORE", "0.42"))
_IMAGE_ACCESS_DENIAL_PATTERNS = (
re.compile(
r"(?:我|目前|暂时|这里|本身|系统)?\s*(?:无法|不能|没法|不支持)\s*"
r"(?:直接)?\s*(?:查看|看到|看见|识别|读取|访问|打开|分析|理解)"
r"(?:\s*(?:或|、|/)\s*(?:查看|看到|看见|识别|读取|访问|打开|分析|理解))*\s*"
r"(?:你(?:发|提供|上传)的|这张|该|当前)?\s*(?:图片|图像|照片|影像|文件)"
),
re.compile(
r"(?:我|这里|目前|暂时)?\s*(?:看不到|看不见|未看到|没有看到|没收到|未收到)\s*"
r"(?:你(?:发|提供|上传)的|这张|该|当前)?\s*(?:图片|图像|照片|影像)"
),
re.compile(
r"\b(?:i\s+)?(?:can(?:not|'t)|am\s+unable\s+to)\s+(?:directly\s+)?"
r"(?:view|see|access|read|analy[sz]e|recogni[sz]e)\s+"
r"(?:the\s+|this\s+|your\s+)?(?:image|photo|picture|scan)\b",
re.IGNORECASE,
),
)
_WRITING_SYSTEM_PATTERNS = {
"han": re.compile(r"[\u3400-\u4dbf\u4e00-\u9fff]"),
"latin": re.compile(r"[A-Za-z\u00c0-\u024f]"),
@@ -178,6 +197,75 @@ def _image_retrieval_question(question: str, image_contexts: list[dict]) -> str:
return "\n".join(part for part in parts if part).strip()
def _answer_denies_available_image(answer: str) -> bool:
"""Reject only whole-image access denials, not uncertainty about one field."""
value = re.sub(r"\s+", " ", answer or "").strip()
return any(pattern.search(value) for pattern in _IMAGE_ACCESS_DENIAL_PATTERNS)
def _compact_context_text(value: Any, limit: int) -> str:
lines = [re.sub(r"\s+", " ", line).strip() for line in str(value or "").splitlines()]
text = "\n".join(line for line in lines if line).strip()
return text[:limit].rstrip()
def _grounded_image_fallback(question: str, image_contexts: list[dict]) -> str:
"""Build a safe answer from completed vision data when the chat model contradicts it."""
summaries: list[str] = []
facts: list[str] = []
excerpts: list[str] = []
warnings: list[str] = []
for context in image_contexts:
summary = _compact_context_text(context.get("summary"), 500)
if summary:
summaries.append(summary)
structured = context.get("structuredData") or {}
if isinstance(structured, dict):
for fact in structured.get("key_facts") or []:
value = _compact_context_text(fact, 300)
if value:
facts.append(value)
extracted = _compact_context_text(context.get("extractedText"), 900)
if extracted:
excerpts.append(extracted)
warning = _compact_context_text(context.get("warning"), 300)
if warning:
warnings.append(warning)
summaries = list(dict.fromkeys(summaries))
facts = list(dict.fromkeys(facts))[:6]
excerpts = list(dict.fromkeys(excerpts))
warnings = list(dict.fromkeys(warnings))
writing_system = _dominant_writing_system(question)
if writing_system == "latin":
parts = []
if summaries:
parts.append("From the image, I can confirm: " + " ".join(summaries))
if facts:
parts.append("Key details:\n" + "\n".join(
f"{index}. {fact}" for index, fact in enumerate(facts, 1)
))
elif excerpts:
parts.append("Visible text:\n" + excerpts[0])
if warnings:
parts.append("Please note: " + " ".join(warnings))
return "\n".join(parts).strip() or "The image is available, but there is not enough clear detail to confirm more."
parts = []
if summaries:
parts.append("从这张图中可以确认:" + ";".join(summaries).rstrip("。;") + "。")
if facts:
parts.append("其中比较明确的信息有:\n" + "\n".join(
f"{index}. {fact}" for index, fact in enumerate(facts, 1)
))
elif excerpts:
parts.append("图中可见的主要文字是:\n" + excerpts[0])
if warnings:
parts.append("需要注意:" + ";".join(warnings).rstrip("。;") + "。")
return "\n".join(parts).strip() or "这张图已经看到了,但目前能确认的清晰信息比较有限。"
def _run_billed_vision_call(
db: Session,
avatar: Avatar,
@@ -553,8 +641,10 @@ def _build_prompt(
if image_contexts:
image_material = json.dumps(image_contexts, ensure_ascii=False, default=str)
system += (
"\n以下是当前会话图片经过视觉识别后得到的资料:\n"
"\n当前会话图片已经成功读取并完成内容识别,以下资料就是可直接使用的图片内容:\n"
f"{image_material}"
"\n必须直接依据这些图片内容回答当前问题。禁止声称无法查看、看不到、未收到、无法识别、"
"无法读取或不能访问图片,也不要要求对方重新上传;只有资料明确标记读取失败时才可以请对方重发。"
"\n图片资料可能包含 OCR 错字、模糊内容或用户尚未确认的信息,只能按可见内容谨慎表达。"
"标准答题对中的事实优先级高于图片资料,知识库事实优先级高于模型推测;发生冲突时遵循更高优先级资料,"
"并自然提醒对方核对原图。不得声称看到了图片中不存在的内容。"
@@ -815,6 +905,14 @@ def _resolve_reply(
except Exception as exc:
release_reservation(db, reservation, str(exc))
raise
answer = str(answer or "").strip()
if image_contexts and _answer_denies_available_image(answer):
logger.warning(
"chat model contradicted ready image context avatar=%s source=%s",
avatar.id,
usage_source,
)
answer = _grounded_image_fallback(question, image_contexts)
result = {
"answer": answer,
"source": "qa" if matched else (
+49 -61
View File
@@ -1,8 +1,5 @@
import os
import json
import logging
import uuid
from datetime import datetime, timezone
from fastapi import APIRouter, UploadFile, File, Depends, Header, HTTPException
from pydantic import BaseModel
@@ -12,16 +9,16 @@ from database import get_db
from models import KnowledgeDoc, QAPair, KnowledgeChunk, Avatar, User
from responses import ok, fail
import embeddings
from services.knowledge_vectorizer import knowledge_vectorizer
router = APIRouter()
logger = logging.getLogger(__name__)
BASE_DIR = os.path.dirname(os.path.abspath(__file__))
UPLOAD_DIR = os.path.abspath(os.getenv("UPLOAD_DIR", os.path.join(BASE_DIR, "uploads")))
os.makedirs(UPLOAD_DIR, exist_ok=True)
ALLOWED_EXT = {".md", ".txt", ".pdf", ".doc", ".docx", ".xlsx"}
MAX_UPLOAD_BYTES = 10 * 1024 * 1024
MAX_UPLOAD_BYTES = 50 * 1024 * 1024
UPLOAD_CHUNK_BYTES = 1024 * 1024
class QAIn(BaseModel):
@@ -71,15 +68,6 @@ def list_docs(avatar_id: str, authorization: str = Header(None), db: Session = D
.order_by(KnowledgeDoc.created_at.desc())
.all()
)
# Older synchronous uploads could be interrupted after persisting "parsing".
# New uploads are committed only after indexing finishes, so these rows are stale.
stale_docs = [doc for doc in docs if doc.status == "parsing"]
if stale_docs:
for doc in stale_docs:
doc.status = "failed"
doc.vectorized = False
doc.chunk_count = 0
db.commit()
return ok([_doc_payload(d) for d in docs])
@@ -93,65 +81,65 @@ async def upload_doc(avatar_id: str, file: UploadFile = File(...), authorization
os.makedirs(avatar_dir, exist_ok=True)
stored = f"{uuid.uuid4().hex}{ext}"
path = os.path.join(avatar_dir, stored)
content = await file.read()
if len(content) > MAX_UPLOAD_BYTES:
return fail("文件不能超过 10MB", code=400)
with open(path, "wb") as f:
f.write(content)
file_size = 0
try:
# Stream large files to disk so a 100MB upload does not occupy 100MB RAM.
with open(path, "wb") as f:
while chunk := await file.read(UPLOAD_CHUNK_BYTES):
file_size += len(chunk)
if file_size > MAX_UPLOAD_BYTES:
raise ValueError("文件不能超过 50MB")
f.write(chunk)
except ValueError as exc:
if os.path.exists(path):
os.remove(path)
return fail(str(exc), code=400)
doc = KnowledgeDoc(
id=uuid.uuid4().hex,
avatar_id=avatar_id,
filename=file.filename,
file_type=ext.lstrip("."),
file_size=len(content),
file_size=file_size,
file_url=f"/api/files/{avatar_id}/{stored}",
status="parsing",
)
# Complete extraction and embedding before the first database commit so a
# process restart cannot leave a permanent "parsing" row behind.
try:
text = embeddings.extract_text(path, ext)
chunks = embeddings.chunk_text(text)
if not chunks:
raise ValueError("文档没有可建立索引的文字内容")
vectors = embeddings.embed(chunks)
if len(vectors) != len(chunks):
raise ValueError("向量服务返回数量与文档分段不一致")
doc.vectorized = True
doc.embedding_model = embeddings.MODEL
doc.chunk_count = len(chunks)
doc.vectorized_at = datetime.now(timezone.utc)
doc.status = "ready"
db.add(doc)
for i, (chunk, vector) in enumerate(zip(chunks, vectors)):
db.add(
KnowledgeChunk(
doc_id=doc.id,
avatar_id=avatar_id,
content=chunk,
vector=json.dumps(vector),
chunk_index=i,
embedding_model=embeddings.MODEL,
)
)
db.commit()
db.refresh(doc)
except Exception as exc:
db.rollback()
doc.status = "failed"
doc.vectorized = False
doc.embedding_model = ""
doc.chunk_count = 0
doc.vectorized_at = None
db.add(doc)
db.commit()
db.refresh(doc)
logger.exception("knowledge vectorization failed for %s: %s", doc.id, exc)
# Persist and acknowledge the upload first. Extraction and embeddings may take
# minutes for a PDF and must never consume the browser request timeout.
db.add(doc)
db.commit()
db.refresh(doc)
knowledge_vectorizer.enqueue(doc.id)
return ok(_doc_payload(doc))
@router.post("/avatar/{avatar_id}/knowledge/docs/{doc_id}/retry")
def retry_doc(avatar_id: str, doc_id: str, authorization: str = Header(None), db: Session = Depends(get_db)):
_require_owned_avatar(db, avatar_id, authorization)
doc = db.query(KnowledgeDoc).filter(
KnowledgeDoc.id == doc_id, KnowledgeDoc.avatar_id == avatar_id
).first()
if not doc:
return fail("文档不存在", code=404)
if doc.vectorized and doc.status == "ready":
return ok(_doc_payload(doc))
stored_name = os.path.basename(doc.file_url or "")
if not stored_name or not os.path.isfile(os.path.join(UPLOAD_DIR, avatar_id, stored_name)):
return fail("原文件不可用,请重新上传", code=400)
db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == doc.id).delete()
doc.status = "parsing"
doc.vectorized = False
doc.embedding_model = ""
doc.chunk_count = 0
doc.vectorized_at = None
doc.error_message = ""
db.commit()
db.refresh(doc)
knowledge_vectorizer.enqueue(doc.id)
return ok(_doc_payload(doc))
@router.delete("/avatar/{avatar_id}/knowledge/docs/{doc_id}")
def delete_doc(avatar_id: str, doc_id: str, authorization: str = Header(None), db: Session = Depends(get_db)):
_require_owned_avatar(db, avatar_id, authorization)
@@ -0,0 +1,125 @@
"""Durable, serial knowledge-document indexing for the avatar knowledge base."""
import json
import logging
import os
import queue
import threading
from datetime import datetime, timezone
from database import SessionLocal
from models import KnowledgeChunk, KnowledgeDoc
import embeddings
logger = logging.getLogger(__name__)
BACKEND_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
UPLOAD_DIR = os.path.abspath(
os.getenv("UPLOAD_DIR", os.path.join(BACKEND_DIR, "routers", "uploads"))
)
class KnowledgeVectorizer:
"""Indexes one document at a time so slow providers cannot block uploads."""
def __init__(self):
self._queue: queue.Queue[str] = queue.Queue()
self._queued: set[str] = set()
self._lock = threading.Lock()
self._thread: threading.Thread | None = None
def start(self):
if self._thread and self._thread.is_alive():
return
self._thread = threading.Thread(
target=self._run, name="knowledge-vectorizer", daemon=True
)
self._thread.start()
db = SessionLocal()
try:
# A process restart must not abandon documents already accepted by upload.
for (doc_id,) in db.query(KnowledgeDoc.id).filter(KnowledgeDoc.status == "parsing"):
self.enqueue(doc_id)
finally:
db.close()
def enqueue(self, doc_id: str):
with self._lock:
if doc_id in self._queued:
return
self._queued.add(doc_id)
self._queue.put(doc_id)
def _run(self):
while True:
doc_id = self._queue.get()
try:
self.vectorize_document(doc_id)
except Exception:
logger.exception("Unexpected knowledge vectorizer failure for %s", doc_id)
finally:
with self._lock:
self._queued.discard(doc_id)
self._queue.task_done()
def vectorize_document(self, doc_id: str):
db = SessionLocal()
try:
doc = db.get(KnowledgeDoc, doc_id)
if not doc or doc.status != "parsing":
return
stored_name = os.path.basename(doc.file_url or "")
path = os.path.join(UPLOAD_DIR, doc.avatar_id, stored_name)
if not stored_name or not os.path.isfile(path):
raise FileNotFoundError("原文件不可用,请重新上传")
text = embeddings.extract_text(path, f".{doc.file_type}")
chunks = embeddings.chunk_text(text)
if not chunks:
raise ValueError("文档没有可建立索引的文字内容")
vectors = embeddings.embed(chunks)
if len(vectors) != len(chunks):
raise ValueError("向量服务返回数量与文档分段不一致")
# Commit the document and every chunk together. Chat only sees complete indexes.
db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == doc.id).delete()
db.add_all(
[
KnowledgeChunk(
doc_id=doc.id,
avatar_id=doc.avatar_id,
content=chunk,
vector=json.dumps(vector),
chunk_index=index,
embedding_model=embeddings.MODEL,
)
for index, (chunk, vector) in enumerate(zip(chunks, vectors))
]
)
doc.vectorized = True
doc.embedding_model = embeddings.MODEL
doc.chunk_count = len(chunks)
doc.vectorized_at = datetime.now(timezone.utc)
doc.status = "ready"
doc.error_message = ""
db.commit()
logger.info("Knowledge document %s indexed with %s chunks", doc.id, len(chunks))
except Exception as exc:
db.rollback()
failed_doc = db.get(KnowledgeDoc, doc_id)
if failed_doc:
db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == failed_doc.id).delete()
failed_doc.status = "failed"
failed_doc.vectorized = False
failed_doc.embedding_model = ""
failed_doc.chunk_count = 0
failed_doc.vectorized_at = None
failed_doc.error_message = str(exc)[:300] or "建立知识索引失败"
db.commit()
logger.exception("Knowledge vectorization failed for %s: %s", doc_id, exc)
finally:
db.close()
knowledge_vectorizer = KnowledgeVectorizer()
@@ -50,8 +50,15 @@ BOXIM_IMAGE_MESSAGE_TYPE = 1
BOXIM_IMAGE_PROMPT = "请看看这张图片。"
BOXIM_IMAGE_UNAVAILABLE_REPLY = "这张图片我暂时没看清,麻烦重新发送一张清晰的原图。"
IMAGE_CONTEXT_LOOKBACK_SECONDS = 1800
IMAGE_REFERENCE_LOOKBACK_SECONDS = 172_800
MAX_RECENT_IMAGE_CONTEXTS = 3
_IMAGE_REFERENCE_PATTERN = re.compile(
r"(?:图片|图像|照片|截图|这张图|刚才.{0,8}图|病例|病历|检查单|检验单|化验单|报告|影像|"
r"\b(?:image|photo|picture|screenshot|scan|report)\b)",
re.IGNORECASE,
)
def _utcnow() -> datetime:
return datetime.utcnow()
@@ -129,6 +136,10 @@ def _event_prompt(event: TakeoverMessage) -> str:
return ""
def _references_recent_image(value: str) -> bool:
return bool(_IMAGE_REFERENCE_PATTERN.search(value or ""))
class TakeoverService:
"""Poll BOXIM, honor the owner grace period, then generate and send one reply."""
@@ -693,8 +704,14 @@ class TakeoverService:
event: TakeoverMessage,
current_source_ids: list[str],
) -> list[TakeoverMessage]:
"""Recover a recent image that an older deployment recorded without a task."""
threshold = event.send_time - timedelta(seconds=IMAGE_CONTEXT_LOOKBACK_SECONDS)
"""Recover missed images, or reuse a referenced image from the last two days."""
references_image = _references_recent_image(event.content)
lookback_seconds = (
IMAGE_REFERENCE_LOOKBACK_SECONDS
if references_image
else IMAGE_CONTEXT_LOOKBACK_SECONDS
)
threshold = event.send_time - timedelta(seconds=lookback_seconds)
candidates = (
db.query(TakeoverMessage)
.filter(
@@ -714,7 +731,15 @@ class TakeoverService:
if not candidates:
return []
handled_ids = set(current_source_ids)
current_ids = set(current_source_ids)
if references_image:
return [
image
for image in reversed(candidates)
if image.boxim_message_id not in current_ids
]
handled_ids = set(current_ids)
task_sources = (
db.query(TakeoverReplyTask.source_message_ids)
.filter(
@@ -12,6 +12,7 @@ from main import app
from models import ChatAttachment
from routers.chat import (
ChatIn,
_answer_denies_available_image,
_attachment_contexts,
_load_chat_attachments,
_resolve_reply,
@@ -274,6 +275,47 @@ def test_image_context_keeps_standard_answer_authoritative():
assert "标准答题对中的事实优先级高于图片资料" in system
def test_ready_image_context_never_returns_whole_image_access_denial():
avatar = SimpleNamespace(
id="avatar-vision",
name="测试分身",
description="产品顾问",
config={},
)
model = Mock(return_value="抱歉,我无法查看或识别图片,请重新上传。")
result = _resolve_reply(
None,
avatar,
"请看看这张图片",
[],
qa_pairs=[],
search_fn=Mock(return_value=[]),
model_client=model,
image_contexts=[{
"id": "attachment",
"filename": "report.jpg",
"category": "medical_document",
"summary": "一份耳鼻喉科门诊记录",
"extractedText": "主诉:咽痛三天",
"structuredData": {"key_facts": ["主诉为咽痛三天"]},
"warning": "请核对原始资料",
}],
)
assert result["source"] == "vision"
assert "一份耳鼻喉科门诊记录" in result["answer"]
assert "主诉为咽痛三天" in result["answer"]
assert "无法查看" not in result["answer"]
system = model.call_args.kwargs["messages"][0]["content"]
assert "当前会话图片已经成功读取" in system
assert "禁止声称无法查看" in system
def test_image_denial_detector_allows_uncertain_field_in_ready_image():
assert _answer_denies_available_image("我无法查看这张图片") is True
assert _answer_denies_available_image("图片中患者姓名无法辨认,主诉为咽痛三天。") is False
def test_attachment_context_does_not_expose_internal_fields():
row = SimpleNamespace(
id="attachment",
@@ -8,6 +8,7 @@ from database import SessionLocal
from main import app
from models import Avatar, KnowledgeChunk, KnowledgeDoc, QAPair
from routers.knowledge import _doc_payload
from services.knowledge_vectorizer import knowledge_vectorizer
client = TestClient(app)
@@ -31,14 +32,14 @@ def test_doc_payload_reports_whether_the_persisted_file_exists(tmp_path: Path):
assert _doc_payload(doc)["filePresent"] is True
def test_upload_marks_vectorization_failure_instead_of_staying_processing(
def test_upload_returns_before_background_vectorization(
tmp_path: Path,
authorization_context,
):
context = authorization_context
with (
patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
patch("routers.knowledge.embeddings.embed", side_effect=RuntimeError("provider unavailable")),
patch("routers.knowledge.knowledge_vectorizer.enqueue") as enqueue,
):
response = client.post(
f"/api/avatar/{context['avatar'].id}/knowledge/docs",
@@ -47,14 +48,15 @@ def test_upload_marks_vectorization_failure_instead_of_staying_processing(
)
payload = response.json()["data"]
assert payload["status"] == "failed"
assert payload["status"] == "parsing"
assert payload["vectorized"] is False
assert payload["chunkCount"] == 0
enqueue.assert_called_once_with(payload["id"])
db = SessionLocal()
try:
stored = db.query(KnowledgeDoc).filter(KnowledgeDoc.id == payload["id"]).one()
assert stored.status == "failed"
assert stored.status == "parsing"
assert db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == stored.id).count() == 0
db.delete(stored)
db.commit()
@@ -62,14 +64,37 @@ def test_upload_marks_vectorization_failure_instead_of_staying_processing(
db.close()
def test_markdown_upload_commits_ready_document_and_chunks_together(
def test_upload_rejects_oversize_file_before_queuing_indexing(
tmp_path: Path,
authorization_context,
):
context = authorization_context
with (
patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
patch("routers.knowledge.embeddings.embed", return_value=[[1.0, 0.0]]),
patch("routers.knowledge.MAX_UPLOAD_BYTES", 4),
patch("routers.knowledge.knowledge_vectorizer.enqueue") as enqueue,
):
response = client.post(
f"/api/avatar/{context['avatar'].id}/knowledge/docs",
headers=context["owner_headers"],
files={"file": ("oversize.md", b"12345", "text/markdown")},
)
payload = response.json()
assert payload["code"] == 400
assert payload["message"] == "文件不能超过 50MB"
enqueue.assert_not_called()
assert not list((tmp_path / context["avatar"].id).glob("*"))
def test_background_vectorizer_commits_ready_document_and_chunks_together(
tmp_path: Path,
authorization_context,
):
context = authorization_context
with (
patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
patch("routers.knowledge.knowledge_vectorizer.enqueue"),
):
response = client.post(
f"/api/avatar/{context['avatar'].id}/knowledge/docs",
@@ -78,14 +103,19 @@ def test_markdown_upload_commits_ready_document_and_chunks_together(
)
payload = response.json()["data"]
assert payload["status"] == "ready"
assert payload["vectorized"] is True
assert payload["chunkCount"] == 1
assert payload["status"] == "parsing"
with (
patch("services.knowledge_vectorizer.UPLOAD_DIR", str(tmp_path)),
patch("services.knowledge_vectorizer.embeddings.embed", return_value=[[1.0, 0.0]]),
):
knowledge_vectorizer.vectorize_document(payload["id"])
db = SessionLocal()
try:
stored = db.query(KnowledgeDoc).filter(KnowledgeDoc.id == payload["id"]).one()
assert stored.status == "ready"
assert stored.vectorized is True
assert stored.chunk_count == 1
assert db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == stored.id).count() == 1
db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == stored.id).delete()
db.delete(stored)
@@ -94,6 +124,87 @@ def test_markdown_upload_commits_ready_document_and_chunks_together(
db.close()
def test_background_vectorizer_keeps_failure_reason_for_retry(
tmp_path: Path,
authorization_context,
):
context = authorization_context
with (
patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
patch("routers.knowledge.knowledge_vectorizer.enqueue"),
):
response = client.post(
f"/api/avatar/{context['avatar'].id}/knowledge/docs",
headers=context["owner_headers"],
files={"file": ("knowledge.md", b"# Knowledge\n\nTest content", "text/markdown")},
)
payload = response.json()["data"]
with (
patch("services.knowledge_vectorizer.UPLOAD_DIR", str(tmp_path)),
patch("services.knowledge_vectorizer.embeddings.embed", side_effect=RuntimeError("provider unavailable")),
):
knowledge_vectorizer.vectorize_document(payload["id"])
db = SessionLocal()
try:
stored = db.query(KnowledgeDoc).filter(KnowledgeDoc.id == payload["id"]).one()
assert stored.status == "failed"
assert stored.error_message == "provider unavailable"
assert db.query(KnowledgeChunk).filter(KnowledgeChunk.doc_id == stored.id).count() == 0
db.delete(stored)
db.commit()
finally:
db.close()
def test_retry_queues_a_failed_document_again(
tmp_path: Path,
authorization_context,
):
context = authorization_context
document_id = f"retry-doc-{context['suffix']}"
avatar_dir = tmp_path / context["avatar"].id
avatar_dir.mkdir()
(avatar_dir / "retry.md").write_text("retry content", encoding="utf-8")
db = SessionLocal()
try:
db.add(
KnowledgeDoc(
id=document_id,
avatar_id=context["avatar"].id,
filename="retry.md",
file_type="md",
file_url=f"/api/files/{context['avatar'].id}/retry.md",
status="failed",
error_message="provider unavailable",
)
)
db.commit()
finally:
db.close()
with (
patch("routers.knowledge.UPLOAD_DIR", str(tmp_path)),
patch("routers.knowledge.knowledge_vectorizer.enqueue") as enqueue,
):
response = client.post(
f"/api/avatar/{context['avatar'].id}/knowledge/docs/{document_id}/retry",
headers=context["owner_headers"],
)
payload = response.json()["data"]
assert payload["status"] == "parsing"
assert payload["errorMessage"] == ""
enqueue.assert_called_once_with(document_id)
db = SessionLocal()
try:
db.query(KnowledgeDoc).filter(KnowledgeDoc.id == document_id).delete()
db.commit()
finally:
db.close()
def test_each_avatar_has_an_independent_document_and_qa_scope(authorization_context):
context = authorization_context
first_avatar_id = context["avatar"].id
@@ -363,6 +363,54 @@ async def test_followup_text_recovers_recent_image_recorded_without_task(service
db.close()
@pytest.mark.asyncio
async def test_explicit_followup_reuses_handled_image_within_two_days(service_context):
session_factory, service, boxim, clock = service_context
await service.poll_and_process_messages()
image_message = {
"id": 116,
"localId": 116,
"sendId": 200,
"recvId": 100,
"sendTime": clock.millis(),
"type": 1,
"content": json.dumps({"originUrl": "https://cdn.example/handled-case.png"}),
}
boxim.messages.append(image_message)
await service.poll_messages()
db = session_factory()
try:
image_task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="116").one()
image_task.status = "sent"
image_task.sent_at = clock.now()
db.commit()
finally:
db.close()
clock.advance(47 * 60 * 60)
boxim.messages.append(
{
"id": 117,
"localId": 117,
"sendId": 200,
"recvId": 100,
"sendTime": clock.millis(),
"type": 0,
"content": "重新看一下刚才那张病例图片",
}
)
await service.poll_messages()
db = session_factory()
try:
task = db.query(TakeoverReplyTask).filter_by(trigger_message_id="117").one()
assert task.source_message_ids == ["116", "117"]
assert task.prompt == "请看看这张图片。\n重新看一下刚才那张病例图片"
finally:
db.close()
@pytest.mark.asyncio
async def test_invalid_image_message_is_recorded_but_not_scheduled(service_context):
session_factory, service, boxim, clock = service_context
@@ -117,7 +117,7 @@ location /api/ {
proxy_set_header X-Forwarded-Proto $scheme;
proxy_buffering off;
proxy_read_timeout 300s;
client_max_body_size 20m;
client_max_body_size 100m;
}
```
+4
View File
@@ -23,6 +23,10 @@ http {
root /usr/share/nginx/html;
index index.html;
# Keep the application gateway aligned with the production edge gateway.
# Without this Nginx rejects ordinary PDF uploads with HTTP 413 before
# FastAPI can return its user-facing file-size validation message.
client_max_body_size 100m;
# SPA 兜底(hash 路由下深链接也可正常加载)
location / {
+6 -1
View File
@@ -305,6 +305,7 @@ export interface KnowledgeDoc {
vectorized?: boolean
embeddingModel?: string
chunkCount?: number
errorMessage?: string
createdAt: string
}
@@ -335,7 +336,8 @@ export const uploadKnowledgeDoc = (avatarId: string, file: File) => {
const form = new FormData()
form.append('file', file)
return request.post<KnowledgeDoc>(`/avatar/${avatarId}/knowledge/docs`, form, {
headers: { 'Content-Type': 'multipart/form-data' }
headers: { 'Content-Type': 'multipart/form-data' },
timeout: 120000
})
}
@@ -343,6 +345,9 @@ export const uploadKnowledgeDoc = (avatarId: string, file: File) => {
export const deleteKnowledgeDoc = (avatarId: string, docId: string) =>
request.delete(`/avatar/${avatarId}/knowledge/docs/${docId}`)
export const retryKnowledgeDoc = (avatarId: string, docId: string) =>
request.post<KnowledgeDoc>(`/avatar/${avatarId}/knowledge/docs/${docId}/retry`)
// 标准问答对列表
export const getQAPairs = (avatarId: string) =>
request.get<QAPair[]>(`/avatar/${avatarId}/knowledge/qa`)
@@ -27,7 +27,7 @@
<p class="upload-hint">支持 MD / TXT / PDF / DOC / DOCX / XLSX,上传后自动向量化</p>
<input ref="fileInput" type="file" accept=".md,.txt,.pdf,.doc,.docx,.xlsx" class="hidden-input" @change="onFileChange" />
</div>
<p v-if="uploading" class="uploading-text">上传并向量化中…</p>
<p v-if="uploading" class="uploading-text">文件上传中…</p>
<p v-if="uploadError" class="error-text">{{ uploadError }}</p>
</div>
@@ -42,7 +42,10 @@
<p class="card-meta">{{ doc.fileType.toUpperCase() }} · {{ formatSize(doc.fileSize) }} · {{ formatDate(doc.createdAt) }}</p>
<p class="card-detail">{{ documentState(doc).detail }}</p>
</div>
<button class="card-delete" @click="removeDoc(doc.id)">删除</button>
<div class="card-actions">
<button v-if="documentState(doc).tone === 'failed'" class="card-retry" @click="retryDoc(doc.id)">重新索引</button>
<button class="card-delete" @click="removeDoc(doc.id)">删除</button>
</div>
</article>
</div>
<div v-else class="card-empty">📂 暂无文档,先上传一个知识文件</div>
@@ -78,7 +81,7 @@
</template>
<script setup lang="ts">
import { ref, onMounted, computed } from 'vue'
import { ref, onMounted, onUnmounted, computed } from 'vue'
import { useRoute, useRouter } from 'vue-router'
import { useAvatarStore } from '@/store/avatar'
import { pickScopedAvatarId, unwrapListData } from '@/utils/avatar-page-data.js'
@@ -87,6 +90,7 @@ import {
getKnowledgeDocs,
uploadKnowledgeDoc,
deleteKnowledgeDoc,
retryKnowledgeDoc,
getQAPairs,
deleteQAPair,
searchKnowledge,
@@ -107,6 +111,7 @@ const uploading = ref(false)
const uploadError = ref('')
const dragOver = ref(false)
const fileInput = ref<HTMLInputElement | null>(null)
let documentPollingTimer: ReturnType<typeof setInterval> | undefined
const query = ref('')
const searching = ref(false)
@@ -123,7 +128,26 @@ const documentState = (doc: any) => {
if (['uploaded', 'parsing'].includes(String(doc.status || '').toLowerCase())) {
return { tone: 'pending', label: '处理中', detail: '正在解析并建立知识索引' }
}
return { tone: 'failed', label: '处理失败', detail: '未能建立知识索引,请删除后重新上传' }
return { tone: 'failed', label: '处理失败', detail: doc.errorMessage || '未能建立知识索引,请重新索引或重新上传' }
}
const hasPendingDocuments = () => docs.value.some((doc) =>
['uploaded', 'parsing'].includes(String(doc.status || '').toLowerCase())
)
const stopDocumentPolling = () => {
if (documentPollingTimer) {
clearInterval(documentPollingTimer)
documentPollingTimer = undefined
}
}
const startDocumentPolling = () => {
if (documentPollingTimer || !hasPendingDocuments()) return
documentPollingTimer = setInterval(async () => {
await loadDocs()
if (!hasPendingDocuments()) stopDocumentPolling()
}, 2000)
}
const loadDocs = async () => {
@@ -131,6 +155,7 @@ const loadDocs = async () => {
try {
const res: any = await getKnowledgeDocs(avatarId.value)
docs.value = unwrapListData(res)
startDocumentPolling()
} catch (e) {
console.error(e)
}
@@ -182,6 +207,17 @@ const doUpload = async (file: File) => {
}
}
const retryDoc = async (id: string) => {
if (!avatarId.value) return
uploadError.value = ''
try {
await retryKnowledgeDoc(avatarId.value, id)
await loadDocs()
} catch (e: any) {
uploadError.value = e?.message || '重新索引失败'
}
}
const removeDoc = async (id: string) => {
if (!avatarId.value) return
await deleteKnowledgeDoc(avatarId.value, id)
@@ -257,6 +293,8 @@ onMounted(async () => {
if (avatarId.value) store.currentAvatarId = avatarId.value
await Promise.all([loadDocs(), loadQA()])
})
onUnmounted(stopDocumentPolling)
</script>
<style scoped>
@@ -300,7 +338,10 @@ onMounted(async () => {
.status-pill.missing { color: #B91C1C; background: #FEF2F2; }
.status-pill.failed { color: #B91C1C; background: #FEF2F2; }
.card-meta, .card-detail { margin: 5px 0 0; color: #9398AE; font-size: 11px; line-height: 1.4; }.card-detail { color: #8B6B58; }
.card-delete { flex: 0 0 auto; align-self: center; border: 0; color: #EF4444; background: #FEF2F2; border-radius: 8px; padding: 7px 9px; font-size: 12px; cursor: pointer; }
.card-actions { flex: 0 0 auto; display: flex; flex-direction: column; align-items: stretch; gap: 6px; }
.card-delete, .card-retry { align-self: center; border: 0; border-radius: 8px; padding: 7px 9px; font-size: 12px; cursor: pointer; white-space: nowrap; }
.card-delete { color: #EF4444; background: #FEF2F2; }
.card-retry { color: #C15F18; background: #FFF3E6; }
.card-empty { padding: 42px 16px; border: 1px dashed #F1D9C3; border-radius: 16px; color: #9398AE; background: #fff; font-size: 14px; text-align: center; }
.qa-card { align-items: stretch; text-align: left; }.qa-card.qa-disabled { opacity: .58; }
.qa-card .card-content,