diff --git a/.env.example b/.env.example index b8dc1de..5b846be 100644 --- a/.env.example +++ b/.env.example @@ -2,3 +2,6 @@ PUBLIC_IP=182.92.86.220 TURN_SECRET=change-me-to-a-long-random-string TURN_URLS=turn:182.92.86.220:3478?transport=udp,turn:182.92.86.220:3478?transport=tcp +S3_ACCESS_KEY=rustfsadmin +S3_SECRET_KEY=rustfsadmin +S3_BUCKET=ai-video diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..166905e --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +# Local state created by docker-compose services. +/data/ +/logs/ diff --git a/backend/.env.example b/backend/.env.example index 6bd4600..b2fce9a 100644 --- a/backend/.env.example +++ b/backend/.env.example @@ -12,6 +12,15 @@ PORT=8000 # 前端开发地址,允许跨域(公网部署时加上实际前端 origin) CORS_ORIGINS=http://localhost:3000,http://127.0.0.1:3000 +# ---- RustFS / S3-compatible storage ---- +S3_ENDPOINT_URL=http://localhost:9000 +S3_ACCESS_KEY=rustfsadmin +S3_SECRET_KEY=rustfsadmin +S3_BUCKET=ai-video +S3_REGION=us-east-1 +KNOWLEDGE_MAX_FILE_BYTES=20971520 +KNOWLEDGE_TOP_K=5 + # ---- WebRTC TURN(公网跨网语音预览;本地开发留空) ---- # 与 docker compose --profile remote 的 coturn 配套。云安全组需放行 UDP 3478 与 49152-49200。 # PUBLIC_IP 填云主机公网 IP(compose 里给 coturn --external-ip 用,见项目根 .env)。 diff --git a/backend/README.md b/backend/README.md index 350447d..8a59f90 100644 --- a/backend/README.md +++ b/backend/README.md @@ -134,8 +134,9 @@ docker compose up # 前台起 pg + api(:8000)+ ui(:3030),日志 docker compose up -d # 后台起;看日志 docker compose logs -f api docker compose down # 停止全部 -# 可选:对象存储 / 后台任务 -docker compose --profile data up # + rustfs(S3) / redis +# 知识库依赖 RustFS；默认 compose 会一起启动。Redis 仍是可选服务。 +docker compose up -d postgres rustfs api ui +docker compose --profile data up -d redis # 可选:公网部署(WebRTC 需 TURN) docker compose --profile remote up -d ``` @@ -143,6 +144,20 @@ docker compose --profile remote up -d > 首次 `up` 会构建 api 镜像(装全量 `requirements.txt`,含 pipecat,较慢)。 > 之后改 Python 代码靠 `--reload` 热更新,不用重建;只有改 `requirements.txt` 才 `docker compose build api`。 +## 极简知识库 MVP + +知识库入口为前端「组件 / 知识库」。使用前先在「组件 / 模型」配置并启用 +一个 Embedding 资源，然后： + +1. 创建知识库并选择 Embedding 模型。 +2. 上传 PDF、DOCX、TXT、Markdown、CSV、JSON 或 HTML，或者直接添加文字。 +3. 文档状态会从 `pending` 变为 `processing`，最终进入 `ready` 或 `failed`。 +4. 失败记录会保留，可在页面重试；完成后可预览分块并使用「检索测试」验证召回。 +5. 在提示词助手的 pipeline 模式下选择知识库。每轮 LLM 推理前会自动检索并注入相关片段。 + +当前后台处理使用 FastAPI 进程内任务，适合 MVP。服务重启时未完成任务会转为失败， +可在页面重试；需要多实例或高吞吐时再迁移到 Redis/ARQ worker。 + ## 待联调 / TODO - [ ] 联调 Pipecat 1.3.0 语音链路与各 OpenAI 兼容服务 diff --git a/backend/app.py b/backend/app.py index 8926e59..78abb73 100644 --- a/backend/app.py +++ b/backend/app.py @@ -17,6 +17,7 @@ from contextlib import asynccontextmanager import settings import uvicorn from db.session import sync_default_tools, sync_interface_definitions +from services.knowledge import recover_interrupted_documents from fastapi import FastAPI from fastapi.middleware.cors import CORSMiddleware @@ -38,6 +39,7 @@ from routes import ( async def lifespan(_app: FastAPI): await sync_interface_definitions() await sync_default_tools() + await recover_interrupted_documents() yield diff --git a/backend/db/models.py b/backend/db/models.py index 9b92029..c2b3823 100644 --- a/backend/db/models.py +++ b/backend/db/models.py @@ -22,6 +22,7 @@ from sqlalchemy import ( ) from sqlalchemy.dialects.postgresql import JSONB from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column +from pgvector.sqlalchemy import Vector class Base(DeclarativeBase): @@ -93,6 +94,49 @@ class KnowledgeBase(Base): ) +class KnowledgeDocument(Base): + """A file or pasted text ingested into one knowledge base.""" + + __tablename__ = "knowledge_documents" + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + knowledge_base_id: Mapped[str] = mapped_column( + String(40), ForeignKey("knowledge_bases.id", ondelete="CASCADE"), index=True + ) + name: Mapped[str] = mapped_column(String(255)) + source_type: Mapped[str] = mapped_column(String(16)) # file|text + mime_type: Mapped[str] = mapped_column(String(128), default="text/plain") + storage_key: Mapped[str | None] = mapped_column(String(1024), nullable=True) + size_bytes: Mapped[int] = mapped_column(Integer, default=0) + status: Mapped[str] = mapped_column(String(16), default="processing") + error_message: Mapped[str] = mapped_column(String(2048), default="") + chunk_count: Mapped[int] = mapped_column(Integer, default=0) + created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) + updated_at: Mapped[datetime] = mapped_column( + DateTime(timezone=True), server_default=func.now(), onupdate=func.now() + ) + + +class KnowledgeChunk(Base): + """Searchable text chunk; Vector has no fixed dimension to support configured models.""" + + __tablename__ = "knowledge_chunks" + __table_args__ = ( + UniqueConstraint("document_id", "chunk_index", name="uq_knowledge_chunk_position"), + ) + + id: Mapped[str] = mapped_column(String(40), primary_key=True) + knowledge_base_id: Mapped[str] = mapped_column( + String(40), ForeignKey("knowledge_bases.id", ondelete="CASCADE"), index=True + ) + document_id: Mapped[str] = mapped_column( + String(40), ForeignKey("knowledge_documents.id", ondelete="CASCADE"), index=True + ) + chunk_index: Mapped[int] = mapped_column(Integer) + content: Mapped[str] = mapped_column(Text) + embedding: Mapped[list[float]] = mapped_column(Vector()) + + class Assistant(Base): """助手(单表,无版本化)。type 为可变普通列,5 种类型共用此表。 diff --git a/backend/migrations/versions/20260712_0005_add_knowledge_documents.py b/backend/migrations/versions/20260712_0005_add_knowledge_documents.py new file mode 100644 index 0000000..9258a0d --- /dev/null +++ b/backend/migrations/versions/20260712_0005_add_knowledge_documents.py @@ -0,0 +1,55 @@ +"""add knowledge documents and vector chunks + +Revision ID: 20260712_0005 +Revises: 20260712_0004 +""" +from collections.abc import Sequence + +from alembic import op +import sqlalchemy as sa +from pgvector.sqlalchemy import Vector + +revision: str = "20260712_0005" +down_revision: str | Sequence[str] | None = "20260712_0004" +branch_labels = None +depends_on = None + + +def upgrade() -> None: + op.execute("CREATE EXTENSION IF NOT EXISTS vector") + op.create_table( + "knowledge_documents", + sa.Column("id", sa.String(40), primary_key=True), + sa.Column("knowledge_base_id", sa.String(40), nullable=False), + sa.Column("name", sa.String(255), nullable=False), + sa.Column("source_type", sa.String(16), nullable=False), + sa.Column("mime_type", sa.String(128), server_default="text/plain", nullable=False), + sa.Column("storage_key", sa.String(1024), nullable=True), + sa.Column("size_bytes", sa.Integer(), server_default="0", nullable=False), + sa.Column("status", sa.String(16), server_default="processing", nullable=False), + sa.Column("error_message", sa.String(2048), server_default="", nullable=False), + sa.Column("chunk_count", sa.Integer(), server_default="0", nullable=False), + sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False), + sa.ForeignKeyConstraint(["knowledge_base_id"], ["knowledge_bases.id"], ondelete="CASCADE"), + ) + op.create_index("ix_knowledge_documents_knowledge_base_id", "knowledge_documents", ["knowledge_base_id"]) + op.create_table( + "knowledge_chunks", + sa.Column("id", sa.String(40), primary_key=True), + sa.Column("knowledge_base_id", sa.String(40), nullable=False), + sa.Column("document_id", sa.String(40), nullable=False), + sa.Column("chunk_index", sa.Integer(), nullable=False), + sa.Column("content", sa.Text(), nullable=False), + sa.Column("embedding", Vector(), nullable=False), + sa.ForeignKeyConstraint(["knowledge_base_id"], ["knowledge_bases.id"], ondelete="CASCADE"), + sa.ForeignKeyConstraint(["document_id"], ["knowledge_documents.id"], ondelete="CASCADE"), + sa.UniqueConstraint("document_id", "chunk_index", name="uq_knowledge_chunk_position"), + ) + op.create_index("ix_knowledge_chunks_knowledge_base_id", "knowledge_chunks", ["knowledge_base_id"]) + op.create_index("ix_knowledge_chunks_document_id", "knowledge_chunks", ["document_id"]) + + +def downgrade() -> None: + op.drop_table("knowledge_chunks") + op.drop_table("knowledge_documents") diff --git a/backend/models.py b/backend/models.py index cb23143..8b6612c 100644 --- a/backend/models.py +++ b/backend/models.py @@ -75,6 +75,7 @@ class AssistantConfig(BaseModel): # Prompt assistant reusable tools. Execution remains type-specific in the pipeline. tools: list[RuntimeTool] = Field(default_factory=list) + knowledge_base_id: str | None = None # workflow 类型:节点图(nodes/edges)。非 workflow 为空,引擎据此决定是否启用。 graph: dict = {} diff --git a/backend/requirements.txt b/backend/requirements.txt index ca612b3..b2c0792 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -24,3 +24,8 @@ sqlalchemy[asyncio]>=2.0 alembic>=1.13 asyncpg greenlet # SQLAlchemy 异步运行时必需(部分平台不会自动带上) +pgvector +boto3 +python-multipart +pypdf +python-docx diff --git a/backend/routes/assistants.py b/backend/routes/assistants.py index 967478a..b457a5c 100644 --- a/backend/routes/assistants.py +++ b/backend/routes/assistants.py @@ -6,6 +6,7 @@ from db.models import ( Assistant, AssistantModelBinding, AssistantToolBinding, + KnowledgeBase, ModelResource, Tool, ) @@ -52,6 +53,14 @@ async def _validate_vision_model( raise HTTPException(400, "视觉模型必须支持图片输入") +async def _validate_knowledge_base(session: AsyncSession, body: AssistantUpsert) -> None: + if body.runtime_mode != "pipeline" or body.type not in {"prompt", "workflow"}: + body.knowledge_base_id = None + return + if body.knowledge_base_id and not await session.get(KnowledgeBase, body.knowledge_base_id): + raise HTTPException(400, "知识库不存在") + + async def _sync_bindings( session: AsyncSession, assistant_id: str, resource_ids: dict[str, str] ) -> None: @@ -166,6 +175,7 @@ async def create_assistant( ): _validate_workflow(body) await _validate_vision_model(session, body) + await _validate_knowledge_base(session, body) data = body.model_dump() resource_ids = data.pop("model_resource_ids") tool_ids = data.pop("tool_ids") @@ -235,6 +245,7 @@ async def update_assistant( raise HTTPException(404, "助手不存在") _validate_workflow(body) await _validate_vision_model(session, body) + await _validate_knowledge_base(session, body) data = body.model_dump() resource_ids = data.pop("model_resource_ids") tool_ids = data.pop("tool_ids") diff --git a/backend/routes/knowledge_bases.py b/backend/routes/knowledge_bases.py index e9cfbf0..f23755d 100644 --- a/backend/routes/knowledge_bases.py +++ b/backend/routes/knowledge_bases.py @@ -6,11 +6,16 @@ KB 自身引用一个 Embedding 模型资源。被助手引用时禁止删除 import uuid -from db.models import KnowledgeBase, ModelResource +from db.models import Assistant, KnowledgeBase, KnowledgeChunk, KnowledgeDocument, ModelResource from db.session import get_session -from fastapi import APIRouter, Depends, HTTPException -from schemas import KnowledgeBaseOut, KnowledgeBaseUpsert +from fastapi import APIRouter, BackgroundTasks, Depends, File, Form, HTTPException, UploadFile +from schemas import ( + KnowledgeBaseOut, KnowledgeBaseUpsert, KnowledgeChunkOut, KnowledgeDocumentOut, + KnowledgeSearchIn, KnowledgeTextIn, +) from services.auth import require_admin +from services.knowledge import create_document, delete_storage_object, process_document, search +import settings from sqlalchemy import select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -43,6 +48,16 @@ def _to_out(kb: KnowledgeBase) -> KnowledgeBaseOut: ) +def _document_out(document: KnowledgeDocument) -> KnowledgeDocumentOut: + return KnowledgeDocumentOut( + id=document.id, knowledge_base_id=document.knowledge_base_id, + name=document.name, source_type=document.source_type, mime_type=document.mime_type, + size_bytes=document.size_bytes, status=document.status, + error_message=document.error_message, chunk_count=document.chunk_count, + created_at=document.created_at.isoformat() if document.created_at else None, + ) + + @router.get("", response_model=list[KnowledgeBaseOut]) async def list_knowledge_bases(session: AsyncSession = Depends(get_session)): rows = ( @@ -83,6 +98,12 @@ async def update_knowledge_base( if not kb: raise HTTPException(404, "知识库不存在") await _validate_embedding_resource(session, body.embedding_model_resource_id) + if kb.embedding_model_resource_id != body.embedding_model_resource_id: + has_document = (await session.execute( + select(KnowledgeDocument.id).where(KnowledgeDocument.knowledge_base_id == kb_id).limit(1) + )).scalar_one_or_none() + if has_document: + raise HTTPException(409, "知识库已有文档，不能更换 Embedding 模型；请新建知识库") for k, v in body.model_dump().items(): setattr(kb, k, v) await session.commit() @@ -97,6 +118,19 @@ async def delete_knowledge_base( kb = await session.get(KnowledgeBase, kb_id) if not kb: raise HTTPException(404, "知识库不存在") + referenced = (await session.execute( + select(Assistant.id).where(Assistant.knowledge_base_id == kb_id).limit(1) + )).scalar_one_or_none() + if referenced: + raise HTTPException(409, "知识库正被助手引用,无法删除") + documents = (await session.execute( + select(KnowledgeDocument).where(KnowledgeDocument.knowledge_base_id == kb_id) + )).scalars().all() + for document in documents: + try: + await delete_storage_object(document) + except Exception: + pass try: await session.delete(kb) await session.commit() @@ -105,3 +139,119 @@ async def delete_knowledge_base( await session.rollback() raise HTTPException(409, "知识库正被助手引用,无法删除") return {"ok": True} + + +@router.get("/{kb_id}/documents", response_model=list[KnowledgeDocumentOut]) +async def list_documents(kb_id: str, session: AsyncSession = Depends(get_session)): + if not await session.get(KnowledgeBase, kb_id): + raise HTTPException(404, "知识库不存在") + rows = (await session.execute( + select(KnowledgeDocument).where(KnowledgeDocument.knowledge_base_id == kb_id) + .order_by(KnowledgeDocument.created_at.desc()) + )).scalars().all() + return [_document_out(row) for row in rows] + + +@router.post("/{kb_id}/documents/text", response_model=KnowledgeDocumentOut, status_code=202) +async def add_text( + kb_id: str, body: KnowledgeTextIn, background_tasks: BackgroundTasks, + session: AsyncSession = Depends(get_session), +): + kb = await session.get(KnowledgeBase, kb_id) + if not kb: + raise HTTPException(404, "知识库不存在") + try: + document = await create_document( + session, kb, name=body.name, source_type="text", + raw_data=body.content.encode("utf-8"), mime_type="text/plain", + ) + background_tasks.add_task(process_document, document.id) + return _document_out(document) + except Exception as exc: + raise HTTPException(400, f"文字入库失败: {exc}") from exc + + +@router.post("/{kb_id}/documents/file", response_model=KnowledgeDocumentOut, status_code=202) +async def add_file( + kb_id: str, + background_tasks: BackgroundTasks, + file: UploadFile = File(...), + name: str = Form(default=""), + session: AsyncSession = Depends(get_session), +): + kb = await session.get(KnowledgeBase, kb_id) + if not kb: + raise HTTPException(404, "知识库不存在") + data = await file.read(settings.KNOWLEDGE_MAX_FILE_BYTES + 1) + if len(data) > settings.KNOWLEDGE_MAX_FILE_BYTES: + limit_mb = settings.KNOWLEDGE_MAX_FILE_BYTES // 1024 // 1024 + raise HTTPException(413, f"文件不能超过 {limit_mb} MB") + filename = file.filename or "未命名文件" + try: + document = await create_document( + session, kb, name=name.strip() or filename, source_type="file", + raw_data=data, mime_type=file.content_type or "application/octet-stream", + ) + background_tasks.add_task(process_document, document.id) + return _document_out(document) + except Exception as exc: + raise HTTPException(400, f"文件入库失败: {exc}") from exc + + +@router.delete("/{kb_id}/documents/{document_id}") +async def delete_document(kb_id: str, document_id: str, session: AsyncSession = Depends(get_session)): + document = await session.get(KnowledgeDocument, document_id) + if not document or document.knowledge_base_id != kb_id: + raise HTTPException(404, "文档不存在") + try: + await delete_storage_object(document) + finally: + await session.delete(document) + await session.commit() + return {"ok": True} + + +@router.post( + "/{kb_id}/documents/{document_id}/retry", + response_model=KnowledgeDocumentOut, + status_code=202, +) +async def retry_document( + kb_id: str, document_id: str, background_tasks: BackgroundTasks, + session: AsyncSession = Depends(get_session), +): + document = await session.get(KnowledgeDocument, document_id) + if not document or document.knowledge_base_id != kb_id: + raise HTTPException(404, "文档不存在") + if document.status in {"pending", "processing"}: + raise HTTPException(409, "文档正在处理中") + document.status = "pending" + document.error_message = "" + await session.commit() + await session.refresh(document) + background_tasks.add_task(process_document, document.id) + return _document_out(document) + + +@router.get("/{kb_id}/documents/{document_id}/chunks", response_model=list[KnowledgeChunkOut]) +async def list_document_chunks( + kb_id: str, document_id: str, session: AsyncSession = Depends(get_session) +): + document = await session.get(KnowledgeDocument, document_id) + if not document or document.knowledge_base_id != kb_id: + raise HTTPException(404, "文档不存在") + chunks = (await session.execute( + select(KnowledgeChunk).where(KnowledgeChunk.document_id == document_id) + .order_by(KnowledgeChunk.chunk_index).limit(100) + )).scalars().all() + return [KnowledgeChunkOut(id=row.id, chunk_index=row.chunk_index, content=row.content) for row in chunks] + + +@router.post("/{kb_id}/search") +async def search_knowledge_base( + kb_id: str, body: KnowledgeSearchIn, session: AsyncSession = Depends(get_session) +): + try: + return await search(session, kb_id, body.query, body.top_k) + except Exception as exc: + raise HTTPException(400, f"检索失败: {exc}") from exc diff --git a/backend/schemas.py b/backend/schemas.py index e87f2b5..4d1da95 100644 --- a/backend/schemas.py +++ b/backend/schemas.py @@ -192,6 +192,35 @@ class KnowledgeBaseOut(KnowledgeBaseUpsert): updated_at: str | None = None +class KnowledgeTextIn(CamelModel): + name: str = Field(min_length=1, max_length=255) + content: str = Field(min_length=1) + + +class KnowledgeDocumentOut(CamelModel): + id: str + knowledge_base_id: str + name: str + source_type: str + mime_type: str + size_bytes: int + status: str + error_message: str = "" + chunk_count: int + created_at: str | None = None + + +class KnowledgeSearchIn(CamelModel): + query: str = Field(min_length=1) + top_k: int = Field(default=5, ge=1, le=20) + + +class KnowledgeChunkOut(CamelModel): + id: str + chunk_index: int + content: str + + # ---------- 接口定义驱动的统一模型资源 ---------- class InterfaceDefinitionOut(CamelModel): interface_type: str diff --git a/backend/services/config_resolver.py b/backend/services/config_resolver.py index c4b34cb..7a78a09 100644 --- a/backend/services/config_resolver.py +++ b/backend/services/config_resolver.py @@ -137,6 +137,7 @@ async def resolve_runtime_config( enableInterrupt=assistant.enable_interrupt, turnConfig=assistant.turn_config or {}, tools=await _tools_for(session, assistant), + knowledge_base_id=assistant.knowledge_base_id, # workflow 图:仅 workflow 类型非空,引擎据此启用图驱动对话 graph=(assistant.graph or {}) if assistant.type == "workflow" else {}, # 外部托管类型连接信息(DB 存真 key,直接注入) diff --git a/backend/services/knowledge.py b/backend/services/knowledge.py new file mode 100644 index 0000000..d128884 --- /dev/null +++ b/backend/services/knowledge.py @@ -0,0 +1,235 @@ +"""Minimal knowledge ingestion and retrieval, isolated from the voice pipeline.""" + +import asyncio +from io import BytesIO +from pathlib import Path +import re +import uuid + +import boto3 +from botocore.config import Config +from botocore.exceptions import ClientError +from docx import Document as DocxDocument +from openai import AsyncOpenAI +from pypdf import PdfReader +from sqlalchemy import delete, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +import settings +from db.models import KnowledgeBase, KnowledgeChunk, KnowledgeDocument, ModelResource + + +def _s3_client(): + return boto3.client( + "s3", + endpoint_url=settings.S3_ENDPOINT_URL, + aws_access_key_id=settings.S3_ACCESS_KEY, + aws_secret_access_key=settings.S3_SECRET_KEY, + region_name=settings.S3_REGION, + config=Config(signature_version="s3v4", s3={"addressing_style": "path"}), + ) + + +def _ensure_bucket_and_put(key: str, data: bytes, mime_type: str) -> None: + client = _s3_client() + try: + client.head_bucket(Bucket=settings.S3_BUCKET) + except ClientError as exc: + status = exc.response.get("ResponseMetadata", {}).get("HTTPStatusCode") + if status != 404: + raise + client.create_bucket(Bucket=settings.S3_BUCKET) + client.put_object(Bucket=settings.S3_BUCKET, Key=key, Body=data, ContentType=mime_type) + + +def _delete_object(key: str) -> None: + _s3_client().delete_object(Bucket=settings.S3_BUCKET, Key=key) + + +def _get_object(key: str) -> bytes: + response = _s3_client().get_object(Bucket=settings.S3_BUCKET, Key=key) + return response["Body"].read() + + +def extract_text(filename: str, data: bytes) -> str: + suffix = Path(filename).suffix.lower() + if suffix == ".pdf": + return "\n".join(page.extract_text() or "" for page in PdfReader(BytesIO(data)).pages) + if suffix == ".docx": + document = DocxDocument(BytesIO(data)) + return "\n".join(paragraph.text for paragraph in document.paragraphs) + if suffix in {".txt", ".md", ".csv", ".json", ".html", ".htm"}: + return data.decode("utf-8", errors="replace") + raise ValueError("暂仅支持 PDF、DOCX、TXT、Markdown、CSV、JSON 和 HTML 文件") + + +def split_text(text: str, chunk_size: int = 800, overlap: int = 120) -> list[str]: + normalized = re.sub(r"[ \t]+", " ", text).strip() + if not normalized: + return [] + chunks: list[str] = [] + start = 0 + while start < len(normalized): + end = min(start + chunk_size, len(normalized)) + if end < len(normalized): + boundary = max(normalized.rfind("\n", start, end), normalized.rfind("。", start, end)) + if boundary > start + chunk_size // 2: + end = boundary + 1 + chunks.append(normalized[start:end].strip()) + if end >= len(normalized): + break + start = max(end - overlap, start + 1) + return [chunk for chunk in chunks if chunk] + + +async def _embedding_resource(session: AsyncSession, kb: KnowledgeBase) -> ModelResource: + resource = ( + await session.get(ModelResource, kb.embedding_model_resource_id) + if kb.embedding_model_resource_id + else None + ) + if not resource or resource.capability != "Embedding" or not resource.enabled: + raise ValueError("请先为知识库选择已启用的 Embedding 模型") + return resource + + +async def _embed(session: AsyncSession, kb: KnowledgeBase, texts: list[str]) -> list[list[float]]: + resource = await _embedding_resource(session, kb) + values, secrets = resource.values or {}, resource.secrets or {} + client = AsyncOpenAI( + api_key=str(secrets.get("apiKey") or ""), + base_url=str(values.get("apiUrl") or "") or None, + ) + try: + response = await client.embeddings.create( + model=str(values.get("modelId") or "text-embedding-3-small"), input=texts + ) + return [item.embedding for item in response.data] + finally: + await client.close() + + +async def create_document( + session: AsyncSession, + kb: KnowledgeBase, + *, + name: str, + source_type: str, + raw_data: bytes, + mime_type: str = "text/plain", +) -> KnowledgeDocument: + document_id = f"doc_{uuid.uuid4().hex[:12]}" + safe_name = Path(name).name + extension = ".txt" if source_type == "text" else Path(safe_name).suffix + storage_key = f"knowledge/{kb.id}/{document_id}/source{extension}" + await asyncio.to_thread(_ensure_bucket_and_put, storage_key, raw_data, mime_type) + + document = KnowledgeDocument( + id=document_id, + knowledge_base_id=kb.id, + name=name, + source_type=source_type, + mime_type=mime_type, + storage_key=storage_key, + size_bytes=len(raw_data), + status="pending", + ) + session.add(document) + await session.commit() + await session.refresh(document) + return document + + +async def process_document(document_id: str) -> None: + """Process one persisted source. Safe to call again after a failure.""" + from db.session import SessionLocal + + async with SessionLocal() as session: + document = await session.get(KnowledgeDocument, document_id) + if not document: + return + kb = await session.get(KnowledgeBase, document.knowledge_base_id) + if not kb or not document.storage_key: + return + document.status = "processing" + document.error_message = "" + document.chunk_count = 0 + await session.execute(delete(KnowledgeChunk).where(KnowledgeChunk.document_id == document.id)) + await session.commit() + + try: + data = await asyncio.to_thread(_get_object, document.storage_key) + text = ( + data.decode("utf-8", errors="replace") + if document.source_type == "text" + else await asyncio.to_thread(extract_text, document.name, data) + ) + chunks = split_text(text) + if not chunks: + raise ValueError("文档中没有可入库的文字") + embeddings: list[list[float]] = [] + for start in range(0, len(chunks), 64): + embeddings.extend(await _embed(session, kb, chunks[start : start + 64])) + if len(embeddings) != len(chunks): + raise ValueError("Embedding 服务返回的向量数量与分块数量不一致") + for index, (content, embedding) in enumerate(zip(chunks, embeddings)): + session.add(KnowledgeChunk( + id=f"chunk_{uuid.uuid4().hex[:12]}", knowledge_base_id=kb.id, + document_id=document.id, chunk_index=index, content=content, embedding=embedding, + )) + document.chunk_count = len(chunks) + document.status = "ready" + await session.commit() + except Exception as exc: + await session.rollback() + document = await session.get(KnowledgeDocument, document_id) + if document: + document.status = "failed" + document.error_message = str(exc)[:2048] + document.chunk_count = 0 + await session.commit() + + +async def recover_interrupted_documents() -> None: + """Make restart-interrupted work visible and retryable.""" + from db.session import SessionLocal + + async with SessionLocal() as session: + await session.execute( + update(KnowledgeDocument) + .where(KnowledgeDocument.status.in_(["pending", "processing"])) + .values(status="failed", error_message="服务重启导致处理被中断，请点击重试") + ) + await session.commit() + + +async def search(session: AsyncSession, kb_id: str, query: str, top_k: int | None = None) -> list[dict]: + kb = await session.get(KnowledgeBase, kb_id) + if not kb: + return [] + has_chunks = (await session.execute( + select(KnowledgeChunk.id) + .join(KnowledgeDocument, KnowledgeDocument.id == KnowledgeChunk.document_id) + .where(KnowledgeChunk.knowledge_base_id == kb_id, KnowledgeDocument.status == "ready") + .limit(1) + )).scalar_one_or_none() + if not has_chunks: + return [] + query_embedding = (await _embed(session, kb, [query]))[0] + distance = KnowledgeChunk.embedding.cosine_distance(query_embedding) + rows = (await session.execute( + select(KnowledgeChunk, KnowledgeDocument.name, distance.label("distance")) + .join(KnowledgeDocument, KnowledgeDocument.id == KnowledgeChunk.document_id) + .where(KnowledgeChunk.knowledge_base_id == kb_id, KnowledgeDocument.status == "ready") + .order_by(distance) + .limit(top_k or settings.KNOWLEDGE_TOP_K) + )).all() + return [ + {"content": chunk.content, "document": name, "score": round(max(0.0, 1.0 - float(dist)), 4)} + for chunk, name, dist in rows + ] + + +async def delete_storage_object(document: KnowledgeDocument) -> None: + if document.storage_key: + await asyncio.to_thread(_delete_object, document.storage_key) diff --git a/backend/services/pipecat/pipeline.py b/backend/services/pipecat/pipeline.py index 03ca866..2f579e2 100644 --- a/backend/services/pipecat/pipeline.py +++ b/backend/services/pipecat/pipeline.py @@ -27,6 +27,8 @@ from services.pipecat.service_factory import ( create_stt, create_tts, ) +from db.session import SessionLocal +from services.knowledge import search as search_knowledge from pipecat.adapters.schemas.function_schema import FunctionSchema from pipecat.adapters.schemas.tools_schema import ToolsSchema @@ -36,6 +38,7 @@ from pipecat.frames.frames import ( InterruptionFrame, LLMFullResponseEndFrame, LLMFullResponseStartFrame, + LLMContextFrame, LLMTextFrame, LLMMessagesAppendFrame, OutputTransportMessageUrgentFrame, @@ -79,6 +82,12 @@ VISION_ANALYSIS_SYSTEM_PROMPT = ( "你是一个视觉理解模型。请只根据图片内容和用户问题给出准确、简洁的中文观察结果。" "如果画面不足以判断,请明确说明不确定。" ) +KNOWLEDGE_TOOL_NAME = "search_knowledge_base" +KNOWLEDGE_SYSTEM_HINT = ( + "你已连接内部知识库。系统会在每轮用户问题前自动提供相关资料；" + "回答资料事实时只根据检索内容，资料不足要明确说明。" +) +KNOWLEDGE_CONTEXT_MARKER = "" def _require(value: str, label: str) -> str: @@ -309,6 +318,54 @@ class ConversationHistoryProcessor(FrameProcessor): await self._recorder.record_transport_message(frame.message) +class KnowledgeRetrievalProcessor(FrameProcessor): + """Retrieve before local LLM inference without changing Pipecat internals.""" + + def __init__(self, knowledge_base_id: str | None): + super().__init__() + self._knowledge_base_id = knowledge_base_id + self._last_signature = "" + + async def process_frame(self, frame, direction: FrameDirection): + await super().process_frame(frame, direction) + if not self._knowledge_base_id or not isinstance(frame, LLMContextFrame): + await self.push_frame(frame, direction) + return + + messages = frame.context.get_messages() + user_messages = [message for message in messages if message.get("role") == "user"] + if not user_messages: + await self.push_frame(frame, direction) + return + query = str(user_messages[-1].get("content") or "").strip() + signature = f"{len(user_messages)}:{query}" + if not query or signature == self._last_signature: + await self.push_frame(frame, direction) + return + self._last_signature = signature + + try: + async with SessionLocal() as session: + results = await search_knowledge(session, self._knowledge_base_id, query) + except Exception as exc: + logger.warning(f"自动知识库检索失败: {exc}") + results = [] + + sources = "\n\n".join( + f"[{index + 1}] 来源：{item['document']}（相关度 {item['score']}）\n{item['content']}" + for index, item in enumerate(results) + ) or "未检索到相关资料。" + block = f"{KNOWLEDGE_CONTEXT_MARKER}\n当前问题的知识库检索结果：\n{sources}" + system_message = next((message for message in messages if message.get("role") == "system"), None) + if system_message is None: + messages.insert(0, {"role": "system", "content": block}) + else: + content = str(system_message.get("content") or "") + base = content.split(KNOWLEDGE_CONTEXT_MARKER, 1)[0].rstrip() + system_message["content"] = f"{base}\n\n{block}" if base else block + await self.push_frame(frame, direction) + + class PassthroughLLMAssistantAggregator(LLMAssistantAggregator): """聚合 LLM 回复进上下文,同时继续把回复帧交给下游 TTS。""" @@ -432,11 +489,12 @@ async def run_pipeline( call_end = CallEndCoordinator(queue_call_end) def with_vision_hint(text: str) -> str: - if not vision_enabled: - return text - if not text: - return VISION_SYSTEM_HINT - return f"{text}\n\n{VISION_SYSTEM_HINT}" + hints = [] + if vision_enabled: + hints.append(VISION_SYSTEM_HINT) + if cfg.knowledge_base_id: + hints.append(KNOWLEDGE_SYSTEM_HINT) + return "\n\n".join(part for part in [text, *hints] if part) context = LLMContext( messages=[{"role": "system", "content": with_vision_hint(system_content)}] @@ -460,6 +518,7 @@ async def run_pipeline( assistant_aggregator = PassthroughLLMAssistantAggregator(context) text_input = TextInputProcessor(should_ignore_input=lambda: call_end.ending) vision_capture = VisionCaptureProcessor() + knowledge_retrieval = KnowledgeRetrievalProcessor(cfg.knowledge_base_id) vision_native_mode = vision_enabled and _vision_uses_main_llm(cfg) vision_state: dict[str, str | None] = {"client_id": None} vision_schema = FunctionSchema( @@ -476,6 +535,27 @@ async def run_pipeline( }, required=["question"], ) + knowledge_schema = FunctionSchema( + name=KNOWLEDGE_TOOL_NAME, + description="在当前助手绑定的知识库中检索与问题最相关的资料片段。", + properties={ + "query": {"type": "string", "description": "用于检索的完整问题或关键词"} + }, + required=["query"], + ) + + async def search_bound_knowledge(params: FunctionCallParams): + query = str(params.arguments.get("query") or "").strip() + if not query or not cfg.knowledge_base_id: + await params.result_callback({"status": "error", "message": "检索问题为空或未绑定知识库"}) + return + try: + async with SessionLocal() as session: + results = await search_knowledge(session, cfg.knowledge_base_id, query) + await params.result_callback({"status": "ok", "results": results}) + except Exception as exc: + logger.exception(f"知识库检索失败: {exc}") + await params.result_callback({"status": "error", "message": "知识库检索暂时不可用"}) async def fetch_user_image(params: FunctionCallParams): question = str(params.arguments.get("question") or "请描述当前画面。") @@ -538,11 +618,15 @@ async def run_pipeline( if vision_enabled: llm.register_function(VISION_TOOL_NAME, fetch_user_image) + if cfg.knowledge_base_id: + llm.register_function(KNOWLEDGE_TOOL_NAME, search_bound_knowledge) def set_visible_tools(schemas: list[FunctionSchema] | None = None) -> None: tools = list(schemas or []) if vision_enabled: tools.append(vision_schema) + if cfg.knowledge_base_id: + tools.append(knowledge_schema) if tools: context.set_tools(ToolsSchema(standard_tools=tools)) else: @@ -561,6 +645,7 @@ async def run_pipeline( text_input, stt, user_aggregator, + knowledge_retrieval, llm, # Aggregate the streamed LLM text before TTS. On interruption, # Pipecat commits the generated prefix immediately instead of diff --git a/backend/settings.py b/backend/settings.py index 82c6537..e1f8700 100644 --- a/backend/settings.py +++ b/backend/settings.py @@ -46,3 +46,12 @@ TURN_SECRET = os.getenv("TURN_SECRET", "") TURN_USERNAME = os.getenv("TURN_USERNAME", "") TURN_PASSWORD = os.getenv("TURN_PASSWORD", "") TURN_CREDENTIAL_TTL = int(os.getenv("TURN_CREDENTIAL_TTL", "86400")) + +# ---- S3-compatible object storage (RustFS in local compose) ---- +S3_ENDPOINT_URL = os.getenv("S3_ENDPOINT_URL", "http://localhost:9000") +S3_ACCESS_KEY = os.getenv("S3_ACCESS_KEY", "rustfsadmin") +S3_SECRET_KEY = os.getenv("S3_SECRET_KEY", "rustfsadmin") +S3_BUCKET = os.getenv("S3_BUCKET", "ai-video") +S3_REGION = os.getenv("S3_REGION", "us-east-1") +KNOWLEDGE_MAX_FILE_BYTES = int(os.getenv("KNOWLEDGE_MAX_FILE_BYTES", "20971520")) +KNOWLEDGE_TOP_K = int(os.getenv("KNOWLEDGE_TOP_K", "5")) diff --git a/backend/tests/test_knowledge.py b/backend/tests/test_knowledge.py new file mode 100644 index 0000000..2739556 --- /dev/null +++ b/backend/tests/test_knowledge.py @@ -0,0 +1,23 @@ +import unittest + +from services.knowledge import extract_text, split_text + + +class KnowledgeTextTest(unittest.TestCase): + def test_extracts_utf8_text(self): + self.assertEqual(extract_text("说明.txt", "你好，知识库".encode()), "你好，知识库") + + def test_split_text_keeps_overlap_and_all_content(self): + text = "第一段。" * 250 + chunks = split_text(text, chunk_size=120, overlap=20) + self.assertGreater(len(chunks), 1) + self.assertTrue(all(chunk for chunk in chunks)) + self.assertLessEqual(max(map(len, chunks)), 120) + + def test_rejects_unsupported_binary_file(self): + with self.assertRaisesRegex(ValueError, "暂仅支持"): + extract_text("archive.zip", b"data") + + +if __name__ == "__main__": + unittest.main() diff --git a/docker-compose.yaml b/docker-compose.yaml index b6c1a24..a941085 100644 --- a/docker-compose.yaml +++ b/docker-compose.yaml @@ -52,6 +52,10 @@ services: # WebRTC TURN(公网部署:设 PUBLIC_IP + TURN_SECRET,并 docker compose --profile remote up) TURN_URLS: "${TURN_URLS:-}" TURN_SECRET: "${TURN_SECRET:-}" + S3_ENDPOINT_URL: "http://rustfs:9000" + S3_ACCESS_KEY: "${S3_ACCESS_KEY:-rustfsadmin}" + S3_SECRET_KEY: "${S3_SECRET_KEY:-rustfsadmin}" + S3_BUCKET: "${S3_BUCKET:-ai-video}" ports: - "8000:8000" depends_on: @@ -96,8 +100,8 @@ services: # RustFS:S3 兼容对象存储(MinIO 替代,Rust 实现) # 9000 = S3 API,9001 = Web 控制台。默认账号 rustfsadmin/rustfsadmin。 rustfs: + # image: registry.cn-hangzhou.aliyuncs.com/qiluo-images/rustfs:latest image: rustfs/rustfs:latest - profiles: ["data"] environment: RUSTFS_VOLUMES: /data # 单盘单节点(开发够用) RUSTFS_ADDRESS: 0.0.0.0:9000 # S3 API @@ -108,10 +112,11 @@ services: RUSTFS_ACCESS_KEY: "${S3_ACCESS_KEY:-rustfsadmin}" RUSTFS_SECRET_KEY: "${S3_SECRET_KEY:-rustfsadmin}" ports: - - "127.0.0.1:9000:9000" - - "127.0.0.1:9001:9001" + - "9000:9000" + - "9001:9001" volumes: - - rustfs-data:/data + - ./data/rustfs:/data + - ./logs/rustfs:/logs networks: [app-network] # ---- 可选(profile: remote):WebRTC 公网穿透 ---- @@ -172,7 +177,6 @@ services: volumes: postgres_data: redis_data: - rustfs-data: networks: app-network: diff --git a/frontend/src/components/pages/AssistantPage.tsx b/frontend/src/components/pages/AssistantPage.tsx index 41c31a7..80f9013 100644 --- a/frontend/src/components/pages/AssistantPage.tsx +++ b/frontend/src/components/pages/AssistantPage.tsx @@ -664,7 +664,7 @@ export function AssistantPage(props: AssistantPageProps) { ...(form.voice ? { TTS: form.voice } : {}), ...(form.realtimeModel ? { Realtime: form.realtimeModel } : {}), }, - knowledgeBaseId: form.knowledgeBase || null, + knowledgeBaseId: form.runtimeMode === "pipeline" ? form.knowledgeBase || null : null, toolIds: form.toolIds, prompt: form.prompt, }), @@ -1900,18 +1900,20 @@ export function AssistantPage(props: AssistantPageProps) { /> - } - title="知识库配置" - description="选择助手回答时可检索的业务知识来源" - > - updateForm("knowledgeBase", value)} - options={kbOptions} - noneLabel="无" - /> - + {form.runtimeMode === "pipeline" && ( + } + title="知识库配置" + description="选择助手回答时可检索的业务知识来源" + > + updateForm("knowledgeBase", value)} + options={kbOptions} + noneLabel="无" + /> + + )} } diff --git a/frontend/src/components/pages/ComponentsKnowledgePage.tsx b/frontend/src/components/pages/ComponentsKnowledgePage.tsx index ed7bcfb..f8c68e0 100644 --- a/frontend/src/components/pages/ComponentsKnowledgePage.tsx +++ b/frontend/src/components/pages/ComponentsKnowledgePage.tsx @@ -1,10 +1,221 @@ -import { PlaceholderPage } from "./PlaceholderPage"; +"use client"; + +import { useCallback, useEffect, useMemo, useState } from "react"; +import { + Database, Eye, FileText, Loader2, Pencil, Plus, RefreshCw, + Search, Trash2, Upload, +} from "lucide-react"; +import { Button } from "@/components/ui/button"; +import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; +import { + Dialog, DialogContent, DialogFooter, DialogHeader, DialogTitle, +} from "@/components/ui/dialog"; +import { Input } from "@/components/ui/input"; +import { Textarea } from "@/components/ui/textarea"; +import { + Select, SelectContent, SelectItem, SelectTrigger, SelectValue, +} from "@/components/ui/select"; +import { + knowledgeBasesApi, modelResourcesApi, type KnowledgeBase, + type KnowledgeChunk, type KnowledgeDocument, type KnowledgeSearchResult, + type ModelResource, +} from "@/lib/api"; + +type BaseDialogMode = "create" | "edit" | null; + +function formatBytes(bytes: number) { + if (bytes < 1024) return `${bytes} B`; + if (bytes < 1024 * 1024) return `${(bytes / 1024).toFixed(1)} KB`; + return `${(bytes / 1024 / 1024).toFixed(1)} MB`; +} + +function statusLabel(status: string) { + return ({ pending: "等待处理", processing: "处理中", ready: "已完成", failed: "失败" } as Record)[status] ?? status; +} + +function statusClass(status: string) { + if (status === "ready") return "bg-emerald-500/10 text-emerald-600"; + if (status === "failed") return "bg-destructive/10 text-destructive"; + return "bg-amber-500/10 text-amber-600"; +} export function ComponentsKnowledgePage() { + const [bases, setBases] = useState([]); + const [models, setModels] = useState([]); + const [selectedId, setSelectedId] = useState(""); + const [documents, setDocuments] = useState([]); + const [busy, setBusy] = useState(false); + const [error, setError] = useState(""); + const [baseDialog, setBaseDialog] = useState(null); + const [textOpen, setTextOpen] = useState(false); + const [previewOpen, setPreviewOpen] = useState(false); + const [previewTitle, setPreviewTitle] = useState(""); + const [chunks, setChunks] = useState([]); + const [name, setName] = useState(""); + const [description, setDescription] = useState(""); + const [embeddingId, setEmbeddingId] = useState(""); + const [textName, setTextName] = useState(""); + const [content, setContent] = useState(""); + const [documentQuery, setDocumentQuery] = useState(""); + const [testQuery, setTestQuery] = useState(""); + const [searching, setSearching] = useState(false); + const [searchResults, setSearchResults] = useState([]); + + const selected = bases.find((item) => item.id === selectedId); + const filteredDocuments = useMemo(() => { + const query = documentQuery.trim().toLowerCase(); + return query ? documents.filter((item) => item.name.toLowerCase().includes(query)) : documents; + }, [documentQuery, documents]); + + const loadBases = useCallback(async () => { + const [nextBases, resources] = await Promise.all([ + knowledgeBasesApi.list(), modelResourcesApi.list(), + ]); + setBases(nextBases); + setModels(resources.filter((item) => item.capability === "Embedding" && item.enabled)); + setSelectedId((current) => nextBases.some((item) => item.id === current) ? current : nextBases[0]?.id || ""); + }, []); + + const loadDocuments = useCallback(async (kbId: string) => { + setDocuments(await knowledgeBasesApi.documents(kbId)); + }, []); + + useEffect(() => { + // eslint-disable-next-line react-hooks/set-state-in-effect -- initial remote data load + void loadBases().catch((cause) => setError(cause instanceof Error ? cause.message : "加载失败")); + }, [loadBases]); + + useEffect(() => { + if (!selectedId) return; + // eslint-disable-next-line react-hooks/set-state-in-effect -- selection-driven remote data load + void loadDocuments(selectedId).catch((cause) => setError(cause.message)); + }, [loadDocuments, selectedId]); + + useEffect(() => { + if (!selectedId || !documents.some((item) => ["pending", "processing"].includes(item.status))) return; + const timer = window.setInterval(() => { + void loadDocuments(selectedId).catch(() => undefined); + }, 2000); + return () => window.clearInterval(timer); + }, [documents, loadDocuments, selectedId]); + + function openCreate() { + setName(""); setDescription(""); setEmbeddingId(""); setBaseDialog("create"); + } + + function openEdit() { + if (!selected) return; + setName(selected.name); setDescription(selected.description); + setEmbeddingId(selected.embeddingModelResourceId || ""); setBaseDialog("edit"); + } + + async function saveBase() { + setBusy(true); setError(""); + try { + const body = { name: name.trim(), description, embeddingModelResourceId: embeddingId || null }; + const saved = baseDialog === "edit" && selected + ? await knowledgeBasesApi.update(selected.id, body) + : await knowledgeBasesApi.create(body); + setBaseDialog(null); await loadBases(); setSelectedId(saved.id); + } catch (cause) { + setError(cause instanceof Error ? cause.message : "保存失败"); + } finally { setBusy(false); } + } + + async function removeBase() { + if (!selected || !window.confirm(`确定删除知识库“${selected.name}”及其全部文档吗？`)) return; + setBusy(true); setError(""); + try { await knowledgeBasesApi.remove(selected.id); setDocuments([]); await loadBases(); } + catch (cause) { setError(cause instanceof Error ? cause.message : "删除失败"); } + finally { setBusy(false); } + } + + async function addText() { + if (!selectedId) return; + setBusy(true); setError(""); + try { + await knowledgeBasesApi.addText(selectedId, { name: textName.trim(), content }); + setTextOpen(false); setTextName(""); setContent(""); await loadDocuments(selectedId); + } catch (cause) { setError(cause instanceof Error ? cause.message : "入库失败"); } + finally { setBusy(false); } + } + + async function addFile(file?: File) { + if (!file || !selectedId) return; + setBusy(true); setError(""); + try { await knowledgeBasesApi.addFile(selectedId, file); await loadDocuments(selectedId); } + catch (cause) { setError(cause instanceof Error ? cause.message : "上传失败"); } + finally { setBusy(false); } + } + + async function retryDocument(documentId: string) { + setError(""); + try { await knowledgeBasesApi.retryDocument(selectedId, documentId); await loadDocuments(selectedId); } + catch (cause) { setError(cause instanceof Error ? cause.message : "重试失败"); } + } + + async function removeDocument(document: KnowledgeDocument) { + if (!window.confirm(`确定删除“${document.name}”吗？`)) return; + try { await knowledgeBasesApi.removeDocument(selectedId, document.id); await loadDocuments(selectedId); } + catch (cause) { setError(cause instanceof Error ? cause.message : "删除失败"); } + } + + async function previewDocument(document: KnowledgeDocument) { + setPreviewTitle(document.name); setChunks([]); setPreviewOpen(true); + try { setChunks(await knowledgeBasesApi.chunks(selectedId, document.id)); } + catch (cause) { setError(cause instanceof Error ? cause.message : "读取分块失败"); } + } + + async function testSearch() { + if (!selectedId || !testQuery.trim()) return; + setSearching(true); setError(""); + try { setSearchResults(await knowledgeBasesApi.search(selectedId, testQuery.trim())); } + catch (cause) { setError(cause instanceof Error ? cause.message : "检索失败"); } + finally { setSearching(false); } + } + return ( - +
+
+

知识库

管理 pipeline 助手可检索的文件与文字资料。

+ +
+ {error &&
{error}
} +
+ 知识库列表 + {bases.length === 0 &&

还没有知识库。

} + {bases.map((base) => ( + + ))} + + +
+
{selected?.name || "请选择知识库"}{selected?.description &&

{selected.description}

}
{selected &&
}
 + + {selected &&
setDocumentQuery(event.target.value)}/>
} +
+ {selected && filteredDocuments.length === 0 &&

暂无文档，上传文件或添加文字开始入库。

} + {filteredDocuments.map((document) => ( +
+
{document.name}{statusLabel(document.status)}
{document.sourceType === "file" ? "文件" : "文字"} · {formatBytes(document.sizeBytes)} · {document.chunkCount} 个分块
{document.status === "failed" && }{document.status === "ready" && }
+ {document.status === "processing" &&
正在抽取、切分并生成向量…
} + {document.errorMessage &&

{document.errorMessage}

} +
+ ))} +
+ + + + {selected && 检索测试

直接验证当前知识库能否召回正确片段。

setTestQuery(event.target.value)} onKeyDown={(event) => { if (event.key === "Enter") void testSearch(); }}/>
{searchResults.map((result, index) =>
来源：{result.document}相关度 {result.score}

{result.content}

)}
} +
+
+ + !open && setBaseDialog(null)}>{baseDialog === "edit" ? "编辑知识库" : "新建知识库"}
setName(event.target.value)}/>