100d1a06eb
- 后端 API（Flask + Gunicorn） - RAG 引擎（混合检索 + 云端 Reranker + 引用溯源） - 文档解析（MinerU + 多格式支持） - Docker 生产部署配置 - 排除前端项目、敏感配置、模型文件
215 lines
8.3 KiB
Python
215 lines
8.3 KiB
Python
"""
|
||
Agentic RAG - 答案生成 Mixin
|
||
|
||
包含答案生成、上下文构建、融合回答等方法
|
||
"""
|
||
|
||
import logging
|
||
|
||
from .agentic_base import logger, MODEL, SOURCE_KB, SOURCE_WEB
|
||
from core.llm_utils import call_llm
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class AnswerMixin:
|
||
"""答案生成方法"""
|
||
|
||
def _generate_fused_answer(self, query: str, contexts: list, allowed_levels: list = None) -> str:
|
||
"""生成融合答案 - 智能处理多源信息"""
|
||
# 分离不同来源
|
||
kb_contexts = [c for c in contexts if c.get('source_type') == self.SOURCE_KB]
|
||
web_contexts = [c for c in contexts if c.get('source_type') == self.SOURCE_WEB]
|
||
|
||
# 如果没有任何上下文，检测是否因权限限制
|
||
if not contexts:
|
||
return self._generate_no_context_answer(query, allowed_levels)
|
||
|
||
# 正常生成答案
|
||
context_str = self._build_context_string(kb_contexts, web_contexts)
|
||
prompt = self._build_normal_answer_prompt(query, context_str, kb_contexts, web_contexts)
|
||
|
||
result = call_llm(
|
||
self.client, prompt, MODEL,
|
||
temperature=0.7,
|
||
max_tokens=2000
|
||
)
|
||
return result or f"生成答案失败"
|
||
|
||
def _build_context_string(self, kb_contexts, web_contexts):
|
||
"""构建上下文字符串 - FAQ 优先策略，按分数排序"""
|
||
# 分离 FAQ 和普通知识库内容
|
||
faq_contexts = [c for c in kb_contexts if c.get('meta', {}).get('chunk_type') == 'faq']
|
||
regular_contexts = [c for c in kb_contexts if c.get('meta', {}).get('chunk_type') != 'faq']
|
||
|
||
# 按分数降序排列，确保最相关的内容优先展示
|
||
regular_contexts.sort(key=lambda c: c.get('score', 0), reverse=True)
|
||
|
||
# FAQ 部分（优先展示）
|
||
faq_parts = []
|
||
for i, c in enumerate(faq_contexts[:3], 1):
|
||
meta = c['meta']
|
||
answer = meta.get('faq_answer', c['doc'])
|
||
faq_parts.append(f"[FAQ-{i}] 常见问题\n问题：{c['doc']}\n标准答案：{answer}")
|
||
|
||
# 普通知识库部分（用 12 条，提升覆盖率）
|
||
kb_parts = []
|
||
for i, c in enumerate(regular_contexts[:12], 1):
|
||
meta = c['meta']
|
||
source_str = meta.get('source', '未知')
|
||
section = meta.get('section', '')
|
||
source_info = f"{source_str}"
|
||
if section:
|
||
source_info += f" > {section[:60]}"
|
||
kb_parts.append(f"[知识库-{i}] {source_info}\n{c['doc']}")
|
||
|
||
web_parts = []
|
||
for i, c in enumerate(web_contexts[:5], 1):
|
||
meta = c['meta']
|
||
web_parts.append(f"[网络-{i}] {meta.get('title', '')}\n来源:{meta.get('source', '')}\n{c['doc']}")
|
||
|
||
return "\n\n".join(faq_parts + kb_parts + web_parts)
|
||
|
||
def _build_normal_answer_prompt(self, query, context_str, kb_contexts, web_contexts):
|
||
"""构建正常回答的提示词（与生产路径 generate_answer_stream 对齐）"""
|
||
# 检测是否有图片上下文
|
||
has_images = any(c.get('meta', {}).get('chunk_type') in ('image', 'chart', 'table')
|
||
for c in kb_contexts)
|
||
|
||
image_instruction = ""
|
||
if has_images:
|
||
image_instruction = "\n5. 如果参考资料中包含【图片N】信息，请在回答中简要介绍每张图片的内容和用途"
|
||
|
||
return f"""你是一个严谨的知识库问答助手。你必须且只能根据用户提供的【参考资料】回答问题。
|
||
|
||
【参考资料】
|
||
{context_str}
|
||
|
||
【用户问题】
|
||
{query}
|
||
|
||
【回答要求】
|
||
1. 如果参考资料中有答案，必须引用对应内容回答，并在回答末尾标注引用编号（如[1]、[2]）
|
||
2. 如果参考资料中确实没有相关信息，简短说明"参考资料中没有相关信息"即可，不要编造或补充资料外的内容
|
||
3. 禁止使用参考资料以外的知识进行补充或推测
|
||
4. 分点列举，条理清晰，语言简洁{image_instruction}
|
||
|
||
请仔细阅读以上全部参考资料后回答："""
|
||
|
||
def _build_answer_prompt_with_permission(self, query, context_str, levels_str, sources_str, kb_contexts, web_contexts):
|
||
"""构建带权限提示的回答提示词"""
|
||
return f"""你是一个严谨的智能助手。
|
||
|
||
【用户问题】
|
||
{query}
|
||
|
||
【重要提示】
|
||
检测到与用户问题更相关的信息可能存在于「{levels_str}」级别的文档中，但用户当前的权限级别无法访问。
|
||
|
||
【可访问的信息来源】
|
||
{context_str}
|
||
|
||
【回答要求】
|
||
1. 首先明确告知用户：当前回答基于您有权限访问的文档，可能不完整
|
||
2. 基于现有信息如实回答
|
||
3. 建议用户如需完整信息，请联系管理员申请相应权限
|
||
|
||
请回答："""
|
||
|
||
def _generate_no_context_answer(self, query: str, allowed_levels: list = None) -> str:
|
||
"""无上下文时的回答 — 诚实告知，不编造"""
|
||
return "参考资料中没有找到与该问题相关的信息，无法根据现有知识库内容回答您的问题。"
|
||
|
||
def _verify_and_refine_answer(self, query: str, answer: str, contexts: list) -> str:
|
||
"""验证并精炼答案 - 防止幻觉
|
||
|
||
返回值始终是干净的答案文本，不包含验证推理过程。
|
||
"""
|
||
prompt = f"""请检查以下回答是否存在"幻觉"（与参考信息不符的内容）。
|
||
|
||
【用户问题】
|
||
{query}
|
||
|
||
【参考信息】
|
||
{chr(10).join([f"[{i+1}] {c['doc'][:200]}" for i, c in enumerate(contexts[:8])])}
|
||
|
||
【AI回答】
|
||
{answer}
|
||
|
||
【检查规则】
|
||
1. 逐条核对回答中的事实是否能在参考信息中找到依据
|
||
2. 如果没有幻觉，只回复一个英文单词：PASS
|
||
3. 如果有幻觉，只输出修正后的完整回答（不要输出检查过程、不要加标题、不要加"检查结果"等前缀）
|
||
|
||
修正后的回答："""
|
||
|
||
try:
|
||
result = call_llm(
|
||
self.client, prompt, MODEL,
|
||
temperature=0.1,
|
||
max_tokens=2000
|
||
)
|
||
if not result:
|
||
return answer
|
||
# 如果返回 PASS 或很短的确认，说明无幻觉
|
||
cleaned = result.strip()
|
||
if cleaned.upper() == "PASS" or len(cleaned) < 10:
|
||
return answer
|
||
# 有幻觉时，返回修正后的干净答案（去掉可能的前缀）
|
||
for prefix in ["修正后的回答：", "修正后回答：", "修正回答：", "修正后："]:
|
||
if cleaned.startswith(prefix):
|
||
cleaned = cleaned[len(prefix):].strip()
|
||
return cleaned
|
||
except Exception as e:
|
||
logger.warning(f"答案验证失败: {e}")
|
||
return answer
|
||
|
||
def _generate_uncertain_answer(self, query: str, contexts: list) -> str:
|
||
"""生成不确定性回答"""
|
||
context_str = "\n".join([c['doc'][:200] for c in contexts[:3]])
|
||
|
||
prompt = f"""用户问题：{query}
|
||
|
||
找到的信息可能不够完整或相关性不高：
|
||
{context_str}
|
||
|
||
请基于这些信息给出一个谨慎的回答，明确说明哪些部分是有依据的，哪些部分可能需要更多验证。
|
||
|
||
回答："""
|
||
|
||
try:
|
||
result = call_llm(
|
||
self.client, prompt, MODEL,
|
||
temperature=0.7,
|
||
max_tokens=1000
|
||
)
|
||
return result or "根据现有信息无法确定答案。"
|
||
except Exception as e:
|
||
logger.error(f"生成不确定性回答失败: {e}")
|
||
return "根据现有信息无法确定答案。"
|
||
|
||
def _direct_answer(self, query: str, history: list = None) -> str:
|
||
"""直接使用 LLM 回答（无知识库检索）"""
|
||
messages = [
|
||
{"role": "system", "content": "你是一个专业的助手，请用中文回答用户的问题。"}
|
||
]
|
||
|
||
if history:
|
||
for h in history[-4:]:
|
||
if h.get("role") in ["user", "assistant"]:
|
||
messages.append({"role": h["role"], "content": h.get("content", "")})
|
||
|
||
messages.append({"role": "user", "content": query})
|
||
|
||
try:
|
||
result = call_llm(
|
||
self.client, "", MODEL,
|
||
temperature=0.7,
|
||
max_tokens=1500,
|
||
messages=messages
|
||
)
|
||
return result or "抱歉，我无法回答这个问题。"
|
||
except Exception as e:
|
||
logger.error(f"直接回答失败: {e}")
|
||
return f"回答生成失败：{str(e)}"
|