ngfchl
170 lines
6.6 KiB
Python
170 lines
6.6 KiB
Python
"""知识库问答服务 — 调用 knowledge/ 轻量检索 + Ollama 生成答案"""
|
||
import logging
|
||
|
||
import httpx
|
||
|
||
from knowledge import (
|
||
OLLAMA_MODEL,
|
||
OLLAMA_URL,
|
||
SITE_URL,
|
||
get_cached_answer,
|
||
search as kb_search,
|
||
set_cached_answer,
|
||
)
|
||
|
||
logger = logging.getLogger("spam_guard")
|
||
|
||
KNOWLEDGE_PROMPT = """你是收割机（go-harvest）文档助手。请根据知识库内容，给用户一个准确、紧凑、可执行的中文回答。
|
||
|
||
相关文档内容：
|
||
{context_text}
|
||
|
||
用户问题：{query}
|
||
|
||
硬性要求（非常重要）：
|
||
1. 只能使用上方知识库内容回答。不要编造镜像名、环境变量、端口、命令、配置文件或链接。
|
||
2. 如果知识库里有代码块/YAML/env/bash 示例，涉及配置时必须原样引用或忠实摘录，不要自行重写完整配置，不要改写变量名。
|
||
3. 如果资料没有给出完整示例，不要自己补全完整示例；应说明“文档未给出完整示例”，再列出文档中已有的片段。
|
||
4. 对 Go Harvest 安装配置，若资料中出现 `newptools/go-harvest:latest`、`EMAIL`、`TOKEN`、`GO_WEB_PORT`、`POSTGRES_DB` 等，请保持这些原名，不要改成其他名字。
|
||
5. 不要把常识当成文档内容。可以说“文档没有说明”，不能猜。
|
||
|
||
回答格式要求：
|
||
1. 尽量控制在 Telegram 单条消息内，优先 1200 字以内，最多不要超过 3000 字。
|
||
2. 使用 Telegram 兼容 Markdown 排版：标题用 *粗体*，列表用 - 或 1.，命令/变量用反引号，配置示例用代码块。
|
||
3. 如果问题是“怎么做/如何配置/怎么安装/报错怎么办”，请使用紧凑结构：
|
||
*结论*
|
||
*步骤*
|
||
*关键配置/命令*（仅限资料中出现的内容）
|
||
*注意*
|
||
*文档*
|
||
4. 不要展开无关背景，不要重复用户问题，不要写长篇解释。
|
||
5. 如果知识库信息不足，要明确说“文档里没有直接说明”，只列可确认部分。
|
||
6. 最后列出相关文档链接，最多 3 个。"""
|
||
|
||
|
||
def build_context(docs: list[dict], max_chars: int = 12000) -> str:
|
||
"""构建给 LLM 的上下文，保留来源、标题、分数。"""
|
||
parts = []
|
||
used = 0
|
||
for i, d in enumerate(docs, 1):
|
||
text = d.get("text", "").strip()
|
||
if not text:
|
||
continue
|
||
block = (
|
||
f"【资料 {i}】\n"
|
||
f"标题：{d.get('title', '')}\n"
|
||
f"来源：{d.get('source', '')}\n"
|
||
f"链接：{d.get('url', '')}\n"
|
||
f"相关度：{d.get('score', 0):.2f}\n"
|
||
f"内容：\n{text}\n"
|
||
)
|
||
if used + len(block) > max_chars and parts:
|
||
break
|
||
parts.append(block)
|
||
used += len(block)
|
||
return "\n---\n".join(parts)
|
||
|
||
|
||
def extract_verbatim_snippets(query: str, docs: list[dict], limit: int = 2, max_chars: int = 4200) -> str:
|
||
"""配置/安装类问题优先展示知识库原文代码块，避免 LLM 重写配置时幻觉。"""
|
||
q = query.lower()
|
||
if not any(k in q for k in ["compose", "docker", "postgres", "postgresql", "配置", "安装", "部署", "环境变量"]):
|
||
return ""
|
||
|
||
snippets = []
|
||
used = 0
|
||
need_postgres = any(k in q for k in ["postgres", "postgresql", "pgsql"])
|
||
|
||
for d in docs:
|
||
text = d.get("text", "")
|
||
text_lower = text.lower()
|
||
if "[代码块]" not in text:
|
||
continue
|
||
# PostgreSQL 问题优先只展示 PostgreSQL 相关代码块，避免混入 SQLite 示例
|
||
if need_postgres and "postgres" not in text_lower:
|
||
continue
|
||
# 优先 yaml/env/bash/text 配置块
|
||
if not any(mark in text_lower for mark in ["```yaml", "```env", "```bash", "```text"]):
|
||
continue
|
||
block = text.replace("[代码块]", "").strip()
|
||
rendered = (
|
||
f"📌 文档原文配置片段（{d.get('title', '')}）\n"
|
||
f"来源：{d.get('url', '')}\n\n"
|
||
f"{block}"
|
||
)
|
||
if used + len(rendered) > max_chars and snippets:
|
||
continue
|
||
snippets.append(rendered)
|
||
used += len(rendered)
|
||
if len(snippets) >= limit:
|
||
break
|
||
|
||
if not snippets:
|
||
return ""
|
||
return "\n\n".join(snippets)
|
||
|
||
|
||
def build_source_text(docs: list[dict], limit: int = 5) -> str:
|
||
seen = []
|
||
for d in docs:
|
||
url = d.get("url")
|
||
title = d.get("doc_title") or d.get("title") or url
|
||
if not url or url in [x[1] for x in seen]:
|
||
continue
|
||
seen.append((title, url))
|
||
if len(seen) >= limit:
|
||
break
|
||
return "\n".join([f"📎 {title}: {url}" for title, url in seen])
|
||
|
||
|
||
async def ask_knowledge_base(query: str) -> str:
|
||
"""知识库问答"""
|
||
cached = get_cached_answer(query)
|
||
if cached:
|
||
logger.info("📦 命中问答缓存")
|
||
return cached
|
||
|
||
# top_k 提高 + search 内部扩展相邻 chunk，让回答不再只基于碎片
|
||
relevant = kb_search(query, top_k=8, expand=True, context_window=1)
|
||
if not relevant:
|
||
return f"❌ 未找到相关内容，请换个关键词试试\n\n文档: {SITE_URL}"
|
||
|
||
context_text = build_context(relevant, max_chars=12000)
|
||
prompt = KNOWLEDGE_PROMPT.format(context_text=context_text, query=query)
|
||
|
||
try:
|
||
import time as _time
|
||
t0 = _time.time()
|
||
logger.info(f"🤖 调用 LLM 生成回答... (模型: {OLLAMA_MODEL}, 上下文: {len(context_text)}字, 文档块: {len(relevant)})")
|
||
async with httpx.AsyncClient(timeout=120, proxy=None) as client:
|
||
resp = await client.post(
|
||
f"{OLLAMA_URL}/api/chat",
|
||
json={
|
||
"model": OLLAMA_MODEL,
|
||
"messages": [{"role": "user", "content": prompt}],
|
||
"stream": False,
|
||
"options": {
|
||
"num_predict": 3000,
|
||
"temperature": 0.25,
|
||
"top_p": 0.9,
|
||
},
|
||
"think": False,
|
||
},
|
||
)
|
||
resp.raise_for_status()
|
||
answer = resp.json()["message"]["content"].strip()
|
||
elapsed = _time.time() - t0
|
||
logger.info(f"✅ LLM 回答完成 (耗时: {elapsed:.1f}s, 长度: {len(answer)}字)")
|
||
|
||
source_text = build_source_text(relevant, limit=5)
|
||
verbatim = extract_verbatim_snippets(query, relevant, limit=2)
|
||
body = f"{verbatim}\n\n---\n\n{answer}" if verbatim else answer
|
||
full_answer = f"{body}\n\n📚 相关文档:\n{source_text}" if source_text else body
|
||
|
||
set_cached_answer(query, full_answer)
|
||
return full_answer
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 知识库问答失败: {e}")
|
||
return f"❌ 回答失败: {e}"
|