190 lines
6.5 KiB
Python
190 lines
6.5 KiB
Python
"""Provide service-layer logic for context builder."""
|
||
|
||
from typing import List, Dict, Optional
|
||
from dataclasses import dataclass
|
||
from loguru import logger
|
||
|
||
from .retriever import RetrievedDocument
|
||
from app.config.settings import settings
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
|
||
|
||
|
||
@dataclass
|
||
class RAGContext:
|
||
"""Represent the R A G Context type."""
|
||
system_prompt: str
|
||
context_text: str
|
||
user_query: str
|
||
total_tokens: int
|
||
sources: List[Dict]
|
||
truncated: bool = False
|
||
|
||
|
||
class ContextBuilder:
|
||
"""Provide the Context Builder builder."""
|
||
|
||
def __init__(
|
||
self,
|
||
max_context_tokens: int = None,
|
||
include_metadata: bool = True,
|
||
citation_format: str = "【条款{clause}】"
|
||
):
|
||
"""Initialize the Context Builder instance."""
|
||
self.max_context_tokens = max_context_tokens or settings.rag_max_context_tokens
|
||
self.include_metadata = include_metadata
|
||
self.citation_format = citation_format
|
||
|
||
logger.info(f"上下文构建器初始化: max_tokens={self.max_context_tokens}")
|
||
|
||
def build(
|
||
self,
|
||
query: str,
|
||
documents: List[RetrievedDocument],
|
||
system_prompt: Optional[str] = None,
|
||
max_tokens: Optional[int] = None
|
||
) -> RAGContext:
|
||
"""Handle build for the Context Builder instance."""
|
||
max_tokens = max_tokens or self.max_context_tokens
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
context_text, sources, truncated = self._format_documents(
|
||
documents,
|
||
max_tokens
|
||
)
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
system_prompt = system_prompt or self._default_system_prompt()
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
total_tokens = self._estimate_tokens(system_prompt + context_text + query)
|
||
|
||
logger.info(f"上下文构建完成: {len(documents)}条文档, {total_tokens}tokens, truncated={truncated}")
|
||
|
||
return RAGContext(
|
||
system_prompt=system_prompt,
|
||
context_text=context_text,
|
||
user_query=query,
|
||
total_tokens=total_tokens,
|
||
sources=sources,
|
||
truncated=truncated
|
||
)
|
||
|
||
def _format_documents(
|
||
self,
|
||
documents: List[RetrievedDocument],
|
||
max_tokens: int
|
||
) -> tuple:
|
||
"""Handle format documents for this module for the Context Builder instance."""
|
||
context_parts = []
|
||
sources = []
|
||
current_tokens = 0
|
||
truncated = False
|
||
|
||
for i, doc in enumerate(documents):
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
formatted = self._format_single_doc(doc, i + 1)
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
doc_tokens = self._estimate_tokens(formatted)
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
if current_tokens + doc_tokens > max_tokens:
|
||
truncated = True
|
||
logger.warning(f"上下文截断: 已达到{max_tokens}tokens限制")
|
||
break
|
||
|
||
context_parts.append(formatted)
|
||
current_tokens += doc_tokens
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
sources.append({
|
||
"index": i + 1,
|
||
"doc_id": doc.doc_id,
|
||
"doc_name": doc.doc_name,
|
||
"section_title": doc.section_title,
|
||
"clause_number": doc.clause_number,
|
||
"page_number": doc.page_number,
|
||
"score": doc.score
|
||
})
|
||
|
||
context_text = "\n\n".join(context_parts)
|
||
return context_text, sources, truncated
|
||
|
||
def _format_single_doc(
|
||
self,
|
||
doc: RetrievedDocument,
|
||
index: int
|
||
) -> str:
|
||
"""Handle format single doc for this module for the Context Builder instance."""
|
||
parts = []
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
parts.append(f"[{index}]")
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
if self.include_metadata:
|
||
meta_parts = []
|
||
|
||
if doc.doc_name:
|
||
meta_parts.append(f"文档: {doc.doc_name}")
|
||
|
||
if doc.section_title:
|
||
meta_parts.append(f"章节: {doc.section_title}")
|
||
|
||
if doc.clause_number:
|
||
clause_text = self.citation_format.format(clause=doc.clause_number)
|
||
meta_parts.append(clause_text)
|
||
|
||
if meta_parts:
|
||
parts.append(" | ".join(meta_parts))
|
||
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
parts.append(doc.content)
|
||
|
||
return "\n".join(parts)
|
||
|
||
def _default_system_prompt(self) -> str:
|
||
"""Handle default system prompt for this module for the Context Builder instance."""
|
||
return """你是合规专家助手，基于提供的法规条款回答问题。
|
||
|
||
回答要求：
|
||
1. 直接回答问题，必须引用具体条款编号（如【条款5.2.1】）
|
||
2. 如引用的条款不完整，说明需要进一步查阅原文
|
||
3. 给出明确的合规建议和操作指导
|
||
4. 如果检索内容不足以回答问题，如实说明
|
||
|
||
回答格式：
|
||
- 先给出直接结论
|
||
- 然后引用支撑条款
|
||
- 最后给出合规建议"""
|
||
|
||
def _estimate_tokens(self, text: str) -> int:
|
||
"""Handle estimate tokens for this module for the Context Builder instance."""
|
||
# Keep service responsibilities explicit so downstream behavior stays predictable.
|
||
chinese_chars = sum(1 for c in text if '一' <= c <= '鿿')
|
||
other_chars = len(text) - chinese_chars
|
||
return int(chinese_chars * 1.5 + other_chars * 0.25)
|
||
|
||
def build_messages(
|
||
self,
|
||
context: RAGContext
|
||
) -> List[Dict[str, str]]:
|
||
"""Build messages for the Context Builder instance."""
|
||
messages = [
|
||
{"role": "system", "content": context.system_prompt},
|
||
{"role": "user", "content": f"参考以下法规条款回答问题。\n\n{context.context_text}\n\n问题：{context.user_query}"}
|
||
]
|
||
|
||
return messages
|
||
|
||
|
||
def build_rag_context(
|
||
query: str,
|
||
documents: List[RetrievedDocument],
|
||
**kwargs
|
||
) -> RAGContext:
|
||
"""Build rag context."""
|
||
builder = ContextBuilder()
|
||
return builder.build(query, documents, **kwargs)
|