feat: 全局「第二大脑」AI 助手(跨书跨对话 + bm_* 工具 + DSML 兜底)

- Conversation 加 kind(book/assistant)、book_id 可空;SQLite 老表迁移保留数据
- rag/engine.py: search_all_books 跨书混合检索;citations 带 book_title 出处
- app/assistant/service.py: 知识库概览 + 第二大脑 persona + 原生工具
  (bm_list_conversations / bm_read_conversation / bm_search_books),并入 MCP agent 循环
- chat.py: create/list 支持 kind;send_message 委托助手管线
- llm/hub.py: 解析 DeepSeek DSML 文本工具调用(模型非确定性输出标准/DSML)
- mcp/agent.py: 工具轮中段追加强制文本收尾,防模型反复调工具不出答案
- 前端: 侧栏「AI 助手」入口 + 抽取共享 ChatPanel 复用书/助手两套对话 UI

Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
2026-10-25 03:21:22 +08:00
co-authored by Claude Code
parent fcdc3d4f93
commit 2cda214170
11 changed files with 1414 additions and 471 deletions
+41 -16
View File
@@ -16,6 +16,7 @@ from app.rag.engine import (
estimate_tokens, match_chapters, _trim_messages,
)
from app.mcp.agent import agent_stream, build_mcp_context, execute_mcp_tool
from app.assistant.service import stream_assistant
logger = logging.getLogger(__name__)
@@ -23,7 +24,9 @@ router = APIRouter(prefix="/chat", tags=["chat"])
class CreateConversationRequest(BaseModel):
book_id: str
# kind="book" 绑定单本书(book_id 必填);kind="assistant" 为全局「第二大脑」对话
kind: str = "book"
book_id: str | None = None
title: str | None = None
model: str = "claude"
@@ -38,7 +41,8 @@ class SendMessageRequest(BaseModel):
class ConversationOut(BaseModel):
id: str
book_id: str
kind: str
book_id: str | None
title: str | None
model: str
created_at: str
@@ -62,29 +66,42 @@ class MessageOut(BaseModel):
@router.get("/conversations", response_model=list[ConversationOut])
async def list_conversations(book_id: str | None = None, db: AsyncSession = Depends(get_db)):
async def list_conversations(
book_id: str | None = None,
kind: str | None = None,
db: AsyncSession = Depends(get_db),
):
q = select(Conversation).order_by(Conversation.updated_at.desc())
if book_id:
q = q.where(Conversation.book_id == book_id)
if kind:
q = q.where(Conversation.kind == kind)
result = await db.execute(q)
convs = result.scalars().all()
return [ConversationOut(
id=c.id, book_id=c.book_id, title=c.title, model=c.model,
id=c.id, kind=c.kind, book_id=c.book_id, title=c.title, model=c.model,
created_at=c.created_at.isoformat() if c.created_at else "",
) for c in convs]
@router.post("/conversations", response_model=ConversationOut)
async def create_conversation(req: CreateConversationRequest, db: AsyncSession = Depends(get_db)):
book = await db.get(Book, req.book_id)
if not book:
raise HTTPException(404, "书不存在")
conv = Conversation(book_id=req.book_id, title=req.title or f"与《{book.title}》的对话", model=req.model)
if req.kind == "assistant":
# 全局「第二大脑」对话:不绑定书
book_id = None
title = req.title or "与 AI 助手的对话"
else:
book = await db.get(Book, req.book_id)
if not book:
raise HTTPException(404, "书不存在")
book_id = req.book_id
title = req.title or f"与《{book.title}》的对话"
conv = Conversation(book_id=book_id, title=title, model=req.model, kind=req.kind)
db.add(conv)
await db.commit()
await db.refresh(conv)
return ConversationOut(
id=conv.id, book_id=conv.book_id, title=conv.title, model=conv.model,
id=conv.id, kind=conv.kind, book_id=conv.book_id, title=conv.title, model=conv.model,
created_at=conv.created_at.isoformat() if conv.created_at else "",
)
@@ -107,10 +124,6 @@ async def send_message(conv_id: str, req: SendMessageRequest, db: AsyncSession =
if not conv:
raise HTTPException(404, "会话不存在")
book = await db.get(Book, conv.book_id)
if not book:
raise HTTPException(404, "书不存在")
model_name = req.model or conv.model
# 重试:复用被重试的用户消息(不新建),新回答作为它的新版本
@@ -125,6 +138,17 @@ async def send_message(conv_id: str, req: SendMessageRequest, db: AsyncSession =
db.add(user_msg)
await db.commit()
# 全局「第二大脑」对话:book_id 为空,走独立管线(跨书检索 + 知识库概览 + 原生工具 + MCP)
if conv.kind == "assistant":
provider = await resolve_provider(db, model_name)
if provider is None:
raise HTTPException(400, f"模型 {model_name} 未配置,请先在页面「管理模型」中添加或选择可用模型")
return await stream_assistant(conv, user_msg, req.content, db, provider)
book = await db.get(Book, conv.book_id)
if not book:
raise HTTPException(404, "书不存在")
chapters = (await db.execute(
select(Chapter).where(Chapter.book_id == book.id).order_by(Chapter.order_idx)
)).scalars().all()
@@ -325,7 +349,7 @@ async def branch_conversation(conv_id: str, from_message_id: str = Query(...), d
conv = await db.get(Conversation, conv_id)
if not conv:
raise HTTPException(404, "会话不存在")
branch = Conversation(book_id=conv.book_id, title=f"{conv.title or '对话'} (分支)", model=conv.model)
branch = Conversation(book_id=conv.book_id, title=f"{conv.title or '对话'} (分支)", model=conv.model, kind=conv.kind)
db.add(branch)
await db.flush()
@@ -364,9 +388,10 @@ async def export_conversation(conv_id: str, fmt: str = "markdown", db: AsyncSess
select(Message).where(Message.conversation_id == conv_id).order_by(Message.created_at)
)
msgs = result.scalars().all()
book = await db.get(Book, conv.book_id)
book = await db.get(Book, conv.book_id) if conv.book_id else None
lines = [f"# {conv.title or '对话'}", f"书籍: {book.title if book else '未知'}\n"]
book_label = "全局助手" if conv.kind == "assistant" else (book.title if book else "未知")
lines = [f"# {conv.title or '对话'}", f"书籍: {book_label}\n"]
for m in msgs:
role_label = "🧑 用户" if m.role == "user" else "🤖 AI"
# 把 mermaid mindmap 转成 flowchart LR:老版阅读器渲染 mindmap 是放射状粗线,很难读
View File
+466
View File
@@ -0,0 +1,466 @@
"""assistant — 全局 AI 助手(第二大脑)管线
与单书对话相对:不绑定某本书。每次请求注入「知识库概览」+ 跨书检索段,
并给模型提供三个原生内部工具(bm_*,走与 MCP 工具同一套 function-calling 循环):
- bm_list_conversations:遍历全部对话(摸清时间/精力去向)
- bm_read_conversation:读取某次对话完整原文
- bm_search_books:跨全部书籍检索原文
外部信息继续走 MCP 工具。不支持的 provider(Anthropic 未接工具循环)则纯流式回答,
行为与单书对话一致。
"""
from __future__ import annotations
import json
import logging
from typing import AsyncGenerator, Awaitable
from fastapi.responses import StreamingResponse
from sqlalchemy import select, func, or_
from sqlalchemy.ext.asyncio import AsyncSession
from app.llm.hub import LLMProvider
from app.mcp.agent import build_mcp_context, execute_mcp_tool, agent_stream
from app.models.book import Book, Chapter, Segment
from app.models.chat import Conversation, Message
from app.rag.engine import (
search_all_books, parse_citations, estimate_tokens, _trim_messages,
)
logger = logging.getLogger(__name__)
NATIVE_PREFIX = "bm_"
NATIVE_TOOL_RESULT_CHARS = 6000 # 单条原生工具结果上限,与 MCP 的 TOOL_RESULT_CHARS 一致
DIGEST_BUDGET_TOKENS = 6000 # 「知识库概览」system 注入预算
# ======================= 知识库概览 =======================
async def _conversation_stats(db: AsyncSession, conv_ids: list[str]) -> dict[str, dict]:
"""每会话的消息数与最后一条消息时间。"""
if not conv_ids:
return {}
rows = await db.execute(
select(Message.conversation_id, func.count(Message.id), func.max(Message.created_at))
.where(Message.conversation_id.in_(conv_ids))
.group_by(Message.conversation_id)
)
return {r[0]: {"count": r[1], "last": r[2]} for r in rows.all()}
async def _recent_message_digest(db: AsyncSession, conv_id: str, limit: int = 2) -> str:
"""会话末尾最近几条消息的紧凑摘要(预算内注入,帮模型感知聊了什么)。"""
rows = await db.execute(
select(Message)
.where(Message.conversation_id == conv_id)
.order_by(Message.created_at.desc())
.limit(limit)
)
msgs = list(rows.scalars().all())
msgs.reverse()
parts = []
for m in msgs:
who = "🧑 用户" if m.role == "user" else "🤖 AI"
text = (m.content or "").replace("\n", " ")[:120]
parts.append(f" {who}:{text}")
return "\n".join(parts)
def _fmt_time(dt) -> str:
return dt.strftime("%Y-%m-%d %H:%M") if dt else ""
async def build_assistant_context(db: AsyncSession) -> tuple[str, list[Book]]:
"""生成「知识库概览」文本 + 全书列表。
概览包含:书库规模、每本书的对话/消息/活跃情况、最近对话摘要、模型使用统计。
全程按 token 预算截断,避免每次请求塞大量原文。
"""
books = (await db.execute(select(Book).order_by(Book.created_at.desc()))).scalars().all()
book_map = {b.id: b for b in books}
convs = (await db.execute(
select(Conversation).order_by(Conversation.updated_at.desc())
)).scalars().all()
conv_stats = await _conversation_stats(db, [c.id for c in convs])
# 每本书聚合:会话数 / 消息数 / 最后活跃
per_book: dict[str, dict] = {}
for b in books:
per_book[b.id] = {"convs": 0, "msgs": 0, "last": None}
for c in convs:
st = per_book.setdefault(c.book_id, {"convs": 0, "msgs": 0, "last": None})
st["convs"] += 1
cs = conv_stats.get(c.id)
if cs:
st["msgs"] += cs["count"]
if cs["last"] and (st["last"] is None or cs["last"] > st["last"]):
st["last"] = cs["last"]
model_rows = (await db.execute(
select(Message.model, func.count(Message.id))
.group_by(Message.model)
.order_by(func.count(Message.id).desc())
)).all()
model_usage = [(r[0], r[1]) for r in model_rows if r[0]]
total_msgs = sum(st["msgs"] for st in per_book.values())
lines: list[str] = []
lines.append("## 知识库概览(自动生成)")
lines.append(f"- 共 {len(books)} 本书、{len(convs)} 次对话、{total_msgs} 条消息。")
if books:
lines.append("\n### 书籍")
for b in books:
st = per_book.get(b.id, {})
last = _fmt_time(st.get("last"))
suffix = f",最后活跃 {last}" if last else ""
lines.append(
f"- 《{b.title or b.filename}》({b.author or '未知作者'},约 {b.char_count:,} 字):"
f"{st.get('convs', 0)} 次对话 · {st.get('msgs', 0)} 条消息{suffix}"
)
used = estimate_tokens("\n".join(lines))
if convs:
conv_lines: list[str] = ["\n### 最近对话"]
for c in convs:
cs = conv_stats.get(c.id)
if not cs:
continue
book = book_map.get(c.book_id)
src = f"《{book.title or book.filename}》" if book else "全局助手"
head = f"- {src} / {c.title or '对话'}({cs['count']} 条消息,最后活跃 {_fmt_time(cs['last'])})"
recent = await _recent_message_digest(db, c.id)
if used + estimate_tokens(head) + estimate_tokens(recent) > DIGEST_BUDGET_TOKENS:
break # 预算内从新到旧追加,超预算即停
conv_lines.append(head)
if recent:
conv_lines.append(recent)
used += estimate_tokens(head) + estimate_tokens(recent)
if len(conv_lines) > 1:
lines.extend(conv_lines)
if model_usage:
lines.append("\n### 模型使用")
for name, cnt in model_usage:
lines.append(f"- {name or '未知'} × {cnt}")
return "\n".join(lines), books
# ======================= Prompt 组装 =======================
def build_assistant_prompt(
digest: str,
segments: list[Segment],
history: list[dict],
query: str,
) -> list[dict]:
system = (
"# 你是 BookMate 的第二大脑助手,用户跨书籍、跨对话的综合思考伙伴。\n\n"
"你的职责:\n"
"1. 梳理总结:帮用户看清时间和精力消耗在哪些书、哪些主题、哪些对话上,以及究竟获得了什么;\n"
"2. 连接知识:跨书籍、跨对话建立概念联系,把零散内容串成体系;\n"
"3. 回顾沉淀:复盘最近读了什么、讨论了什么、收获了什么,指出待办与存疑之处;\n"
"4. 需要书中具体内容时检索并引用原文;需要外部信息(时事/数据/网络)时调用外部工具。\n\n"
)
system += digest + "\n"
if segments:
system += "\n## 与问题直接相关的全书检索段落(含出处)\n"
for i, seg in enumerate(segments, 1):
bt = getattr(seg, "_book_title", None)
ct = getattr(seg, "_chapter_title", None)
src = f"{bt}{'·' + ct if ct else ''}" if bt else seg.book_id
system += f"\n[[ref:{i}]] [{src}] {seg.text[:800].replace(chr(10), ' ')}\n"
else:
system += (
"\n## 检索说明\n"
"本次没有检索到与问题直接匹配的原文。请基于上方知识库概览与你的判断回答,"
"并明确说明哪些是推断。\n"
)
system += (
"\n## 回答要求\n"
"1. 区分信息来源:「书中原文」带 [[ref:N]] 引用;「对话记录」说明是哪次对话;「外部工具」注明来源与时效\n"
"2. 引用原文时用 [[ref:N]](N 对应上方段落编号),同一段只引用一次,整体引用控制在 5 条左右\n"
"3. 用中文回答;检索不到就诚实说明,不要编造统计数字或原文\n"
"4. 涉及精力/时间/收获的梳理时,尽量给出有依据的归纳,标注哪些来自对话记录、哪些是推断"
)
messages = [{"role": "system", "content": system}]
if history:
messages.extend(history[-10:])
messages.append({"role": "user", "content": query})
return messages
# ======================= 原生内部工具 =======================
NATIVE_TOOLS: list[dict] = [
{
"type": "function",
"function": {
"name": "bm_search_books",
"description": "在用户的全部书籍中做全文检索,返回相关段落(每条带书名与章节)。当用户问题需要书中具体原文/出处时调用。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "要检索的问题或关键词(中文亦可)"},
"top_k": {"type": "integer", "description": "返回段落条数,默认 6,最大 15"},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "bm_read_conversation",
"description": "读取某一次完整对话的原文(该对话所有用户与 AI 消息)。conversation_id 从 bm_list_conversations 或知识库概览中获得。",
"parameters": {
"type": "object",
"properties": {
"conversation_id": {"type": "string", "description": "要读取的对话 id"},
},
"required": ["conversation_id"],
},
},
},
{
"type": "function",
"function": {
"name": "bm_list_conversations",
"description": "列出用户的全部对话(可按书名/标题/消息关键词过滤),含每段对话的书名、消息数与最后活跃时间。用于梳理时间和精力都花在哪些书、哪些话题上。",
"parameters": {
"type": "object",
"properties": {
"query": {"type": "string", "description": "可选过滤关键词(匹配书名、标题、消息内容)"},
"limit": {"type": "integer", "description": "最多返回的对话数,默认 20,最大 50"},
},
},
},
},
]
async def _tool_search_books(db: AsyncSession, args: dict) -> str:
query = str(args.get("query") or "").strip()
if not query:
return "[bm_search_books] 缺少 query 参数"
top_k = max(1, min(int(args.get("top_k") or 6), 15))
books = (await db.execute(select(Book))).scalars().all()
segs = await search_all_books(db, query, top_k=top_k, books=books)
if not segs:
return "[bm_search_books] 没有检索到相关段落"
ch_ids = {s.chapter_id for s in segs}
chapters = (await db.execute(select(Chapter).where(Chapter.id.in_(ch_ids)))).scalars().all()
ch_map = {c.id: c for c in chapters}
book_map = {b.id: b for b in books}
lines = []
for i, s in enumerate(segs, 1):
book = book_map.get(s.book_id)
ch = ch_map.get(s.chapter_id)
if book:
src = f"《{book.title or book.filename}》{(' / ' + ch.title) if ch and ch.title else ''}"
else:
src = s.book_id
lines.append(f"[{i}] {src}:{s.text[:300].replace(chr(10), ' ')}")
return "\n".join(lines)
async def _tool_read_conversation(db: AsyncSession, args: dict) -> str:
conv_id = str(args.get("conversation_id") or "").strip()
conv = await db.get(Conversation, conv_id)
if not conv:
return "[bm_read_conversation] 对话不存在"
book = await db.get(Book, conv.book_id) if conv.book_id else None
msgs = (await db.execute(
select(Message).where(Message.conversation_id == conv_id).order_by(Message.created_at)
)).scalars().all()
title = f"《{book.title or book.filename}》" if book else "全局助手"
lines = [f"# {title} / {conv.title or '对话'}(共 {len(msgs)} 条消息)"]
for m in msgs:
who = "🧑 用户" if m.role == "user" else "🤖 AI"
lines.append(f"{who}:{(m.content or '')[:600].replace(chr(10), ' ')}")
return "\n".join(lines)[:NATIVE_TOOL_RESULT_CHARS]
async def _tool_list_conversations(db: AsyncSession, args: dict) -> str:
query = str(args.get("query") or "").strip()
limit = max(1, min(int(args.get("limit") or 20), 50))
q = select(Conversation).order_by(Conversation.updated_at.desc())
if query:
books_match = select(Book.id).where(
or_(Book.title.contains(query), Book.filename.contains(query))
)
msgs_match = select(Message.conversation_id).where(Message.content.contains(query))
q = q.where(or_(
Conversation.title.contains(query),
Conversation.book_id.in_(books_match),
Conversation.id.in_(msgs_match),
))
convs = (await db.execute(q.limit(limit))).scalars().all()
if not convs:
return "[bm_list_conversations] 没有匹配的对话"
book_map = {b.id: b for b in (await db.execute(select(Book))).scalars().all()}
stats = await _conversation_stats(db, [c.id for c in convs])
lines = []
for c in convs:
book = book_map.get(c.book_id)
src = f"《{book.title or book.filename}》" if book else "全局助手"
st = stats.get(c.id, {})
last = _fmt_time(st.get("last"))
lines.append(
f"- [{c.id}] {src} / {c.title or '对话'}({st.get('count', 0)} 条消息"
+ (f",最后活跃 {last}" if last else "")
+ ")"
)
return "\n".join(lines)
async def execute_native_tool(db: AsyncSession, name: str, args: dict) -> str:
if name == "bm_search_books":
return await _tool_search_books(db, args)
if name == "bm_read_conversation":
return await _tool_read_conversation(db, args)
if name == "bm_list_conversations":
return await _tool_list_conversations(db, args)
return f"[内部工具] 未知工具:{name}"
def _dispatch_tool(db: AsyncSession, name: str, args: dict, mcp_ctx) -> Awaitable[str]:
"""按前缀分发:bm_ 内部工具,其余交给 MCP。返回可等待的工具结果。"""
if name.startswith(NATIVE_PREFIX):
return execute_native_tool(db, name, args)
if mcp_ctx is not None:
return execute_mcp_tool(mcp_ctx, name, args)
return _noop_tool_result(f"[工具] 无法执行:{name}")
async def _noop_tool_result(text: str) -> str:
return text
def _describe_tool(name: str, mcp_ctx) -> str:
if name.startswith(NATIVE_PREFIX):
return name
return mcp_ctx.display_name(name) if mcp_ctx is not None else name
# ======================= 流式回答 =======================
async def _annotate_segments(db: AsyncSession, books: list[Book], segments: list[Segment]) -> list[Segment]:
"""给跨书检索段临时标注书名/章节名,供 prompt 出处与 citations 使用。"""
if not segments:
return segments
book_map = {b.id: b for b in books}
ch_ids = {s.chapter_id for s in segments}
chapters = (await db.execute(select(Chapter).where(Chapter.id.in_(ch_ids)))).scalars().all()
ch_map = {c.id: c for c in chapters}
for s in segments:
b = book_map.get(s.book_id)
s._book_title = (b.title or b.filename) if b else None
c = ch_map.get(s.chapter_id)
s._chapter_title = c.title if c else None
return segments
async def stream_assistant(
conv: Conversation,
user_msg: Message,
content: str,
db: AsyncSession,
provider: LLMProvider,
) -> StreamingResponse:
"""全局助手对话的 SSE 流式回答。SSE 事件格式与单书对话完全一致。"""
digest, books = await build_assistant_context(db)
segments = await _annotate_segments(
db, books, await search_all_books(db, content, top_k=10, books=books)
)
# 本会话历史:正常发送时当前 user 消息是最后一条,去掉它;
# 重试时它可能是中间某条,历史只取它之前(旧版本回答不进 prompt)
history_rows = (await db.execute(
select(Message).where(Message.conversation_id == conv.id).order_by(Message.created_at)
)).scalars().all()
ids = {m.id for m in history_rows}
if user_msg.id in ids:
idx = next(i for i, m in enumerate(history_rows) if m.id == user_msg.id)
history = [{"role": m.role, "content": m.content} for m in history_rows[:idx]]
else:
history = [{"role": m.role, "content": m.content} for m in history_rows[:-1]]
messages = build_assistant_prompt(digest, segments, history, content)
mcp_ctx = await build_mcp_context(db) if provider.supports_tools else None
tools: list[dict] = []
if provider.supports_tools:
tools = list(NATIVE_TOOLS)
if mcp_ctx is not None:
tools.extend(mcp_ctx.ai_tools)
system = messages[0]["content"]
system += (
"\n\n## 内部工具\n"
"你有三个内部工具可以调用:\n"
"- bm_list_conversations:列出全部对话(可按关键词过滤),摸清时间精力去向;\n"
"- bm_read_conversation:读取某次对话的完整原文;\n"
"- bm_search_books:跨全部书籍检索原文段落。\n"
"涉及「读了什么 / 收获 / 时间精力去向」的梳理时,先调用工具核实再回答。\n"
)
if mcp_ctx is not None:
system += (
"\n## 外部信息工具\n"
f"需要书中/对话之外的外部信息(最新动态、时事、网络资料、行业数据等)时,"
f"可调用:{', '.join(mcp_ctx.names)}\n"
"- 外部信息要说明来源与时效;回答主体仍以书中内容与对话记录为准。\n"
)
messages[0]["content"] = system
total_tokens = estimate_tokens("".join(m["content"] for m in messages))
if total_tokens > provider.max_context * 0.8:
messages = _trim_messages(messages, int(provider.max_context * 0.8))
async def stream_response() -> AsyncGenerator[str, None]:
full_content = ""
try:
if provider.supports_tools:
async for kind, payload in agent_stream(
messages,
provider,
tools,
execute=lambda name, args: _dispatch_tool(db, name, args, mcp_ctx),
describe=lambda name: _describe_tool(name, mcp_ctx),
):
if kind == "token":
full_content += payload
yield f"data: {json.dumps({'type': 'token', 'content': payload}, ensure_ascii=False)}\n\n"
elif kind == "tool":
yield f"data: {json.dumps({'type': 'tool', 'name': payload, 'status': 'start'}, ensure_ascii=False)}\n\n"
elif kind == "tool_done":
yield f"data: {json.dumps({'type': 'tool', 'name': payload, 'status': 'done'}, ensure_ascii=False)}\n\n"
else:
async for token in provider.chat_stream(messages, temperature=0.7):
full_content += token
yield f"data: {json.dumps({'type': 'token', 'content': token}, ensure_ascii=False)}\n\n"
except Exception as e:
logger.exception("助手流式生成失败(conversation=%s)", conv.id)
yield f"data: {json.dumps({'type': 'error', 'content': str(e)}, ensure_ascii=False)}\n\n"
return
parsed = parse_citations(full_content, segments)
assistant_msg = Message(
conversation_id=conv.id,
parent_message_id=user_msg.id,
role="assistant",
content=parsed["text"],
citations=json.dumps(parsed["citations"], ensure_ascii=False),
model=user_msg.model or conv.model,
)
db.add(assistant_msg)
await db.commit()
await db.refresh(assistant_msg)
yield f"data: {json.dumps({'type': 'done', 'message_id': assistant_msg.id, 'citations': parsed['citations']}, ensure_ascii=False)}\n\n"
return StreamingResponse(stream_response(), media_type="text/event-stream")
+27 -1
View File
@@ -20,13 +20,39 @@ async def init_db():
async def _migrate():
"""轻量迁移:SQLite 下给已有 segments 表补 embedding 列(create_all 不会改老表)。"""
"""轻量迁移:SQLite 下补列 / 重建表(create_all 不会改老表)。"""
async with engine.begin() as conn:
rows = (await conn.exec_driver_sql("PRAGMA table_info(segments)")).all()
cols = {row[1] for row in rows}
if rows and "embedding" not in cols:
await conn.exec_driver_sql("ALTER TABLE segments ADD COLUMN embedding TEXT")
# conversations:全局「第二大脑」对话需要 book_id 可空 + kind 列(book/assistant)。
# SQLite 无法 ALTER 去掉 NOT NULL,这里整表重建(保留数据,旧会话统一 kind='book');
# 新建库由 create_all 直接按新 ORM schema 建表,走到这里时 kind 已存在 → 跳过。
rows = (await conn.exec_driver_sql("PRAGMA table_info(conversations)")).all()
cols = {row[1] for row in rows}
if rows and "kind" not in cols:
await conn.exec_driver_sql("ALTER TABLE conversations RENAME TO conversations_old")
await conn.exec_driver_sql(
"CREATE TABLE conversations ("
" id VARCHAR(36) NOT NULL PRIMARY KEY,"
" book_id VARCHAR(36),"
" title VARCHAR(500),"
" model VARCHAR(50) NOT NULL,"
" created_at DATETIME DEFAULT CURRENT_TIMESTAMP,"
" updated_at DATETIME DEFAULT CURRENT_TIMESTAMP,"
" kind VARCHAR(20) DEFAULT 'book',"
" FOREIGN KEY (book_id) REFERENCES books (id) ON DELETE CASCADE"
")"
)
await conn.exec_driver_sql(
"INSERT INTO conversations (id, book_id, title, model, created_at, updated_at, kind)"
" SELECT id, book_id, title, model, created_at, updated_at, 'book'"
" FROM conversations_old"
)
await conn.exec_driver_sql("DROP TABLE conversations_old")
async def get_db():
async with async_session() as session:
+77 -3
View File
@@ -1,11 +1,49 @@
import json
import os
import re
from abc import ABC, abstractmethod
from dataclasses import dataclass, field
from typing import AsyncGenerator
from sqlalchemy.ext.asyncio import AsyncSession
# DeepSeek 部分模型会以「纯文本 content」输出 DSML 工具调用而不是标准 tool_calls:
# <||DSML|| calls>
# <||DSML|| invoke name="tool">
# <||DSML|| parameter name="k" string="true">v</||DSML|| parameter>
# </||DSML|| invoke>
# </||DSML|| calls>
# 这里把 DSML 块解析成 OpenAI 兼容的 tool_calls,保证工具循环不中断(模型选型不确定时也能兜底)。
DSML_CALLS_START = "<||DSML|| calls>"
DSML_CALLS_END = "</||DSML|| calls>"
_DSML_CALLS_RE = re.compile(r"<||DSML|| calls>(.*?)</||DSML|| calls>", re.DOTALL)
_DSML_INVOKE_RE = re.compile(r'<||DSML|| invoke name="([^"]+)">(.*?)</||DSML|| invoke>', re.DOTALL)
_DSML_PARAM_RE = re.compile(r'<||DSML|| parameter name="([^"]+)"(?: string="true")?>(.*?)</||DSML|| parameter>', re.DOTALL)
def _parse_dsml_calls(block: str) -> list[dict]:
"""把一段 DSML calls 块解析成 OpenAI 兼容的 tool_calls 列表(参数统一按字符串)。
id 必须生成真实值:后续第二轮会带 role="tool" 的消息(tool_call_id 指向它),
空 id 会导致模型 API 拒绝、工具循环第二轮中断。
"""
calls: list[dict] = []
m = _DSML_CALLS_RE.search(block)
if not m:
return calls
for n, inv in enumerate(_DSML_INVOKE_RE.finditer(m.group(1))):
name = inv.group(1).strip()
args: dict = {}
for pm in _DSML_PARAM_RE.finditer(inv.group(2)):
key = pm.group(1).strip()
args[key] = pm.group(2).strip()
calls.append({
"id": f"call_{name}_{n}",
"type": "function",
"function": {"name": name, "arguments": json.dumps(args, ensure_ascii=False)},
})
return calls
@dataclass
class ChatResponse:
@@ -83,6 +121,8 @@ class OpenAIProvider(LLMProvider):
json=payload,
) as resp:
tool_calls_acc: dict[int, dict] = {}
dsml_buf = ""
in_dsml = False
async for line in resp.aiter_lines():
if line.startswith("data: "):
data_str = line[6:]
@@ -97,8 +137,38 @@ class OpenAIProvider(LLMProvider):
delta = choices[0].get("delta", {})
content = delta.get("content", "")
if content:
self.last_content += content
yield content
dsml_buf += content
if in_dsml:
# 正在 DSML 块内:等结束标记后解析工具调用
end = dsml_buf.find(DSML_CALLS_END)
if end != -1:
block = dsml_buf[: end + len(DSML_CALLS_END)]
self.tool_calls.extend(_parse_dsml_calls(block))
tail = dsml_buf[end + len(DSML_CALLS_END):]
in_dsml = False
dsml_buf = tail
# 块后的残余文本(罕见)按正常内容继续
if dsml_buf:
self.last_content += dsml_buf
yield dsml_buf
dsml_buf = ""
else:
idx = dsml_buf.find(DSML_CALLS_START)
if idx != -1:
before = dsml_buf[:idx]
if before:
self.last_content += before
yield before
dsml_buf = dsml_buf[idx:]
in_dsml = True
else:
# 未发现 DSML 标记:yield 除检测窗口外的文本(跨 chunk 检测需保留尾部)
keep = len(DSML_CALLS_START) - 1
if len(dsml_buf) > keep:
head = dsml_buf[:-keep]
self.last_content += head
yield head
dsml_buf = dsml_buf[-keep:]
# 流式 tool_calls:每个 index 首个 delta 带 id/name,后续增量拼 arguments
for tc in delta.get("tool_calls") or []:
idx = tc.get("index", 0)
@@ -114,8 +184,12 @@ class OpenAIProvider(LLMProvider):
entry["function"]["arguments"] += fn["arguments"]
except json.JSONDecodeError:
continue
# 流结束:把检测窗口残留的文本冲出来,避免丢失内容
if dsml_buf:
self.last_content += dsml_buf
yield dsml_buf
if tool_calls_acc:
self.tool_calls = [tool_calls_acc[i] for i in sorted(tool_calls_acc)]
self.tool_calls.extend(tool_calls_acc[i] for i in sorted(tool_calls_acc))
class AnthropicProvider(LLMProvider):
+13 -2
View File
@@ -154,7 +154,18 @@ async def agent_stream(
- ("tool", "<服务器>/<工具>") —— 工具调用开始
- ("tool_done", "<服务器>/<工具>") —— 工具执行完成
"""
for _round in range(MAX_TOOL_ROUNDS):
# 前 MAX_TOOL_ROUNDS 轮正常工具循环;若模型持续调工具仍不收尾(DeepSeek 用 DSML
# 反复输出工具调用尤甚),中段追加「强制收尾」指令再续同样轮数,之后兜底无工具流式。
total_rounds = MAX_TOOL_ROUNDS * 2
for _round in range(total_rounds):
if _round == MAX_TOOL_ROUNDS:
messages.append({
"role": "user",
"content": (
"(工具调用已结束)请直接基于以上已有信息,用中文给出最终回答,"
"不要再调用任何工具,也不要输出任何工具调用标记。"
),
})
provider.tool_calls = []
provider.last_content = ""
async for token in provider.chat_stream(messages, temperature=0.7, tools=tools):
@@ -192,6 +203,6 @@ async def agent_stream(
"tool_call_id": call["id"],
"content": (out or "[工具无返回内容]")[:TOOL_RESULT_CHARS],
})
# 达到最大轮数仍未见收尾:用已收集上下文走无工具流式生成,保证必出最终答案
# 兜底:无工具流式生成,保证必出最终答案
async for token in provider.chat_stream(messages, temperature=0.7):
yield ("token", token)
+3 -1
View File
@@ -13,7 +13,9 @@ class Conversation(Base):
__tablename__ = "conversations"
id: Mapped[str] = mapped_column(String(36), primary_key=True, default=gen_uuid)
book_id: Mapped[str] = mapped_column(String(36), ForeignKey("books.id", ondelete="CASCADE"))
# kind="book" 绑定具体书(book_id 必填);kind="assistant" 为全局「第二大脑」对话(book_id 为 NULL)
kind: Mapped[str] = mapped_column(String(20), default="book")
book_id: Mapped[str | None] = mapped_column(String(36), ForeignKey("books.id", ondelete="CASCADE"), nullable=True)
title: Mapped[str | None] = mapped_column(String(500))
model: Mapped[str] = mapped_column(String(50), default="claude")
created_at: Mapped[datetime] = mapped_column(DateTime, server_default=func.now())
+97 -5
View File
@@ -65,15 +65,18 @@ def match_chapters(chapters: list[Chapter], query: str, tokens: list[str] | None
return hits
async def _keyword_candidates(db: AsyncSession, book_id: str, tokens: list[str], limit: int) -> list[Segment]:
"""关键词候选:SQL LIKE 召回(含整句子串),按命中词数 + 长词加权打分。"""
if not tokens:
async def _keyword_candidates(db: AsyncSession, book_ids: list[str], tokens: list[str], limit: int) -> list[Segment]:
"""关键词候选:SQL LIKE 召回(含整句子串),按命中词数 + 长词加权打分。
book_ids 传单个书名 id 的列表(单书检索)或多个 id(跨书检索)。
"""
if not tokens or not book_ids:
return []
ordered = sorted(tokens, key=len, reverse=True)[:8]
conds = [Segment.text.contains(w) for w in ordered]
q = (
select(Segment)
.where(Segment.book_id == book_id, or_(*conds))
.where(Segment.book_id.in_(book_ids), or_(*conds))
.order_by(Segment.order_idx)
.limit(limit * 4)
)
@@ -124,7 +127,7 @@ async def search_segments(
status = "failed"
# 1. 关键词候选
kw_candidates = await _keyword_candidates(db, book_id, tokens, top_k * 3)
kw_candidates = await _keyword_candidates(db, [book_id], tokens, top_k * 3)
# 2. 向量候选
vec_candidates: list[Segment] = []
@@ -163,6 +166,93 @@ async def search_segments(
return merged[:top_k]
async def _empty_fallback_all(db: AsyncSession, books: list[Book], top_k: int) -> list[Segment]:
"""跨书空检索兜底:每本书取最先的非空开头段,保证 AI 至少有所依据。"""
out: list[Segment] = []
for b in books:
result = await db.execute(
select(Segment)
.where(Segment.book_id == b.id)
.order_by(Segment.order_idx)
.limit(top_k)
)
for s in result.scalars().all():
if (s.text or "").strip():
out.append(s)
break
return out[:top_k]
async def search_all_books(
db: AsyncSession,
query: str,
top_k: int = 12,
books: list[Book] | None = None,
) -> list[Segment]:
"""跨书混合检索(全局「第二大脑」用):关键词跨书一次 SQL + 每书向量余弦 → RRF → rerank。
与 search_segments 的区别:
- 不调用 ensure_embeddings(全局检索不阻塞向量生成,缺向量的书靠关键词兜底);
- 向量部分逐书扫描后合并(query embedding 只算一次)。
"""
tokens = _tokenize(query)
if books is None:
result = await db.execute(select(Book).order_by(Book.created_at))
books = list(result.scalars().all())
book_ids = [b.id for b in books]
if not book_ids:
return []
# 1. 关键词候选(一次 SQL 跨全部书)
kw_candidates = await _keyword_candidates(db, book_ids, tokens, top_k * 4)
# 2. 向量候选(每书全量余弦,合并)
vec_candidates: list[Segment] = []
if tokens:
qvecs = await embed_texts([query])
qvec = qvecs[0] if qvecs else None
if qvec:
for bid in book_ids:
mat, vec_segments = await load_vectors(db, bid)
if mat is not None and len(mat) > 0:
idxs = vector_search(qvec, mat, top_k * 2)
vec_candidates.extend(vec_segments[i] for i in idxs)
# 3. RRF 融合
if kw_candidates and vec_candidates:
by_id = {s.id: s for s in kw_candidates + vec_candidates}
fused = rrf_fuse(
[s.id for s in kw_candidates],
[s.id for s in vec_candidates],
)
merged = [by_id[sid] for sid in fused if sid in by_id]
elif vec_candidates:
merged = vec_candidates
else:
merged = kw_candidates
# 4. rerank 精排(对融合后的 top50 段打分)
if merged and RERANK_ENABLED:
docs = [s.text[:800] for s in merged[:50]]
idxs = await rerank_documents(query, docs, top_n=top_k)
if idxs is not None:
merged = [merged[i] for i in idxs if i < len(merged)]
# 5. 兜底:完全没召回时给各书开头段
if not merged:
merged = await _empty_fallback_all(db, books, top_k)
# 去重(跨书向量合并可能撞段)
seen: set[str] = set()
out: list[Segment] = []
for s in merged:
if s.id in seen:
continue
seen.add(s.id)
out.append(s)
return out[:top_k]
def _chapter_full_text(chapter_segments: list[Segment], budget_tokens: int = 6000) -> str:
"""整章注入:按 token 预算截取命中章节的原文。"""
parts = []
@@ -274,6 +364,8 @@ def parse_citations(content: str, segments: list[Segment]) -> dict:
"segment_id": seg.id,
"cfi": seg.cfi,
"excerpt": seg.text[:200],
# 跨书检索时由调用方在段上临时标注 _book_title,供前端显示出处
"book_title": getattr(seg, "_book_title", None),
})
clean_text = CITATION_RE.sub("", content)
return {"text": clean_text, "citations": citations}