134 lines
5.6 KiB
Python
134 lines
5.6 KiB
Python
import logging
|
||
import re
|
||
|
||
import httpx
|
||
|
||
from app.config import LOW_SCORE, llm_config
|
||
|
||
logger = logging.getLogger("learn_rag")
|
||
|
||
# 发给服务器的第一条消息，role = system。
|
||
# 这里只有客服身份和回答规则，不含检索结果。换模型时这段一般不用改。
|
||
SYSTEM_PROMPT = """你是「知迹」的客服。用友好、耐心、自然的中文和顾客说话，像在柜台前当面回答。
|
||
规则：
|
||
1. 只能根据本次给出的知识切片回答，不要补充切片之外的信息，不要编造。
|
||
2. 先用「您好」一类的问候开场，再回答问题，并补上顾客需要知道的条件。语气像柜台客服，礼貌、清楚，不要只报两个数字。
|
||
3. 如果切片不够回答，坦诚说明现在查不到，并建议对方到服务台再确认。
|
||
4. 不要使用 Markdown，不要写标题，不要复述全部切片。
|
||
5. 引用事实时在相关句末标注〔来源1〕这种编号。编号只能使用给定切片的编号。"""
|
||
|
||
|
||
def build_context(chunks: list[dict]) -> str:
|
||
blocks = []
|
||
for chunk in chunks:
|
||
blocks.append(
|
||
f"〔来源{chunk['n']}〕《{chunk['title']}》相关度 {chunk['score']:.0%}\n{chunk['text']}"
|
||
)
|
||
return "\n\n".join(blocks)
|
||
|
||
|
||
def extractive_answer(chunks: list[dict], low_confidence: bool) -> str:
|
||
if not chunks:
|
||
return "知识库里还没有可以检索的内容。"
|
||
lead = ""
|
||
if low_confidence:
|
||
lead = "最接近的切片和问题只是部分相关，下面按原文摘出，请以右侧切片为准。\n\n"
|
||
lines = [lead + chunks[0]["text"].strip() + f"〔来源{chunks[0]['n']}〕"]
|
||
for chunk in chunks[1:2]:
|
||
if chunks[0]["score"] - chunk["score"] <= 0.08:
|
||
lines.append("\n" + chunk["text"].strip() + f"〔来源{chunk['n']}〕")
|
||
return "\n".join(lines).strip()
|
||
|
||
|
||
def generate(question: str, chunks: list[dict]) -> tuple[str, str, str | None, bool]:
|
||
low_confidence = not chunks or chunks[0]["score"] < LOW_SCORE
|
||
if not chunks:
|
||
return "知识库里还没有可以检索的内容。", "extractive", None, True
|
||
|
||
settings = llm_config()
|
||
if settings is None:
|
||
notice = "还没有配置大模型，回答直接摘自相关度最高的切片。在项目目录的 .env 中填入 OPENAI_API_KEY 后重启，即可生成连贯回答。"
|
||
return extractive_answer(chunks, low_confidence), "extractive", notice, low_confidence
|
||
|
||
caution = ""
|
||
if low_confidence:
|
||
caution = "注意：最高相关度偏低。如果切片不能直接支持答案，请坦诚告诉顾客查不到。\n\n"
|
||
# 发给服务器的第二条消息，role = user。检索到的切片从这里进提示词。
|
||
# 拼出来的样子：
|
||
# 下面是检索到的知识切片，请据此回答顾客。
|
||
#
|
||
# 〔来源1〕《文档标题》相关度 75%
|
||
# 切片原文……
|
||
#
|
||
# 〔来源2〕《文档标题》相关度 71%
|
||
# 切片原文……
|
||
#
|
||
# 顾客的问题：柏卡会员年费多少，能借几本书？
|
||
user_prompt = (
|
||
f"{caution}下面是检索到的知识切片，请据此回答顾客。\n\n"
|
||
f"{build_context(chunks)}\n\n"
|
||
f"顾客的问题：{question}"
|
||
)
|
||
try:
|
||
answer = call_llm(settings, user_prompt)
|
||
except Exception:
|
||
logger.exception("大模型调用失败")
|
||
notice = "大模型暂时不可用，已改为直接摘录检索到的切片。"
|
||
return extractive_answer(chunks, low_confidence), "extractive", notice, low_confidence
|
||
return answer, "llm", None, low_confidence
|
||
|
||
|
||
def strip_think(text: str) -> str:
|
||
return re.sub(r"<think>.*?</think>", "", text, flags=re.DOTALL).strip()
|
||
|
||
|
||
def call_llm(settings: dict, user_prompt: str) -> str:
|
||
# 一次请求只生成一次回答。messages 就是上面两条：
|
||
# system = 客服身份；user = 知识切片 + 顾客问题。
|
||
# 换模型只改 OLLAMA_MODEL（现在是 qwen3-vl:4b），这两条提示词保持不变。
|
||
messages = [
|
||
{"role": "system", "content": SYSTEM_PROMPT},
|
||
{"role": "user", "content": user_prompt},
|
||
]
|
||
timeout = httpx.Timeout(180.0, connect=10.0)
|
||
if settings["provider"] == "ollama":
|
||
# POST {OLLAMA_BASE_URL}/api/chat
|
||
# think=false：不要把推理过程写进回答。
|
||
response = httpx.post(
|
||
f"{settings['base']}/api/chat",
|
||
json={
|
||
"model": settings["model"],
|
||
"messages": messages,
|
||
"stream": False,
|
||
"think": False,
|
||
"options": {"temperature": 0.4, "num_predict": 800},
|
||
},
|
||
timeout=timeout,
|
||
)
|
||
response.raise_for_status()
|
||
message = response.json().get("message") or {}
|
||
text = strip_think(message.get("content") or "")
|
||
else:
|
||
response = httpx.post(
|
||
f"{settings['base']}/chat/completions",
|
||
headers={"Authorization": f"Bearer {settings['key']}"},
|
||
json={
|
||
"model": settings["model"],
|
||
"temperature": 0.4,
|
||
"max_tokens": 800,
|
||
"messages": messages,
|
||
},
|
||
timeout=timeout,
|
||
)
|
||
response.raise_for_status()
|
||
message = response.json()["choices"][0]["message"]
|
||
content = message.get("content")
|
||
if isinstance(content, list):
|
||
content = "".join(
|
||
part.get("text", "") if isinstance(part, dict) else str(part) for part in content
|
||
)
|
||
text = strip_think(content or "")
|
||
if not text:
|
||
raise ValueError("模型返回了空回答")
|
||
return text
|