« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Extractive synthesizer implementation.
3This module implements an extractive synthesizer that selects and ranks
4relevant sentences from context chunks to build a response.
5"""
7from __future__ import annotations
9import re
11from lexigram.ai.rag.synthesis.synthesizers.base import AbstractSynthesizer
12from lexigram.ai.rag.synthesis.types import (
13 ContextChunk,
14 SynthesisResult,
15 SynthesisStrategy,
16)
19class ExtractiveSynthesizer(AbstractSynthesizer):
20 """Extractive sentence-based synthesizer.
22 This synthesizer extracts the most relevant sentences from context chunks
23 using scoring methods like TF-IDF, keyword matching, and position.
25 Attributes:
26 max_sentences: Maximum number of sentences to extract
27 min_sentence_length: Minimum sentence length in characters
28 use_query_keywords: Whether to boost sentences with query keywords
29 reorder_sentences: Whether to reorder for coherence
30 """
32 def __init__(
33 self,
34 max_sentences: int = 5,
35 min_sentence_length: int = 20,
36 use_query_keywords: bool = True,
37 reorder_sentences: bool = True,
38 ):
39 """Initialize the extractive synthesizer.
41 Args:
42 max_sentences: Maximum number of sentences to extract
43 min_sentence_length: Minimum sentence length
44 use_query_keywords: Whether to use query keyword matching
45 reorder_sentences: Whether to reorder for coherence
46 """
47 self.max_sentences = max_sentences
48 self.min_sentence_length = min_sentence_length
49 self.use_query_keywords = use_query_keywords
50 self.reorder_sentences = reorder_sentences
52 def _extract_sentences(self, text: str) -> list[str]:
53 """Extract sentences from text.
55 Args:
56 text: Input text
58 Returns:
59 List of sentence strings
60 """
61 # Simple sentence splitting (can be improved with spacy/nltk)
62 sentences = re.split(r"[.!?]+\s+", text)
63 return [
64 s.strip()
65 for s in filter(
66 lambda s: len(s.strip()) >= self.min_sentence_length,
67 sentences,
68 )
69 ]
71 def _extract_keywords(self, text: str) -> set[str]:
72 """Extract keywords from text.
74 Args:
75 text: Input text
77 Returns:
78 Set of keywords (lowercased, filtered)
79 """
80 # Simple keyword extraction (can be improved with NLP)
81 words = re.findall(r"\b\w+\b", text.lower())
83 # Filter stop words
84 stop_words = {
85 "the",
86 "a",
87 "an",
88 "and",
89 "or",
90 "but",
91 "in",
92 "on",
93 "at",
94 "to",
95 "for",
96 "of",
97 "with",
98 "by",
99 "from",
100 "as",
101 "is",
102 "was",
103 "are",
104 "been",
105 "be",
106 "have",
107 "has",
108 "had",
109 "do",
110 "does",
111 "did",
112 "will",
113 "would",
114 "should",
115 "could",
116 "may",
117 "might",
118 "must",
119 "can",
120 "this",
121 "that",
122 "these",
123 "those",
124 "i",
125 "you",
126 "he",
127 "she",
128 "it",
129 "we",
130 "they",
131 "what",
132 "which",
133 "who",
134 "when",
135 "where",
136 "why",
137 "how",
138 }
140 return {w for w in words if w not in stop_words and len(w) > 2}
142 def _score_sentence(
143 self,
144 sentence: str,
145 query_keywords: set[str],
146 position: int,
147 total_sentences: int,
148 ) -> float:
149 """Score a sentence for relevance.
151 Args:
152 sentence: The sentence to score
153 query_keywords: Keywords from the query
154 position: Position in original text (0-based)
155 total_sentences: Total number of sentences
157 Returns:
158 Relevance score (higher is better)
159 """
160 score = 0.0
162 # Length score (prefer moderate length)
163 length = len(sentence)
164 if 50 <= length <= 200:
165 score += 1.0
166 elif 20 <= length < 50 or 200 < length <= 300:
167 score += 0.5
169 # Keyword matching score
170 if self.use_query_keywords and query_keywords:
171 sentence_words = set(re.findall(r"\b\w+\b", sentence.lower()))
172 keyword_overlap = len(sentence_words & query_keywords)
173 score += keyword_overlap * 2.0
175 # Position score (early sentences often more relevant)
176 position_score = 1.0 - (position / max(total_sentences, 1))
177 score += position_score * 0.5
179 return score
181 async def _synthesize_internal(
182 self,
183 query: str,
184 context_chunks: list[ContextChunk],
185 **kwargs,
186 ) -> SynthesisResult:
187 """Synthesize response by extracting relevant sentences.
189 Args:
190 query: The user query
191 context_chunks: Retrieved context chunks
192 **kwargs: Additional parameters
194 Returns:
195 SynthesisResult with extracted sentences
197 Raises:
198 ValueError: If query is empty or no context chunks provided
199 """
200 if not query:
201 msg = "Query cannot be empty"
202 raise ValueError(msg)
203 if not context_chunks:
204 msg = "No context chunks provided"
205 raise ValueError(msg)
207 # Extract query keywords
208 query_keywords = self._extract_keywords(query)
210 # Extract and score sentences from all chunks
211 scored_sentences: list[tuple[str, float, ContextChunk, int]] = []
213 for chunk in context_chunks:
214 sentences = self._extract_sentences(chunk.text)
216 for pos, sentence in enumerate(sentences):
217 score = self._score_sentence(
218 sentence,
219 query_keywords,
220 pos,
221 len(sentences),
222 )
223 # Boost by chunk relevance score
224 final_score = score * (chunk.score if chunk.score else 1.0)
225 scored_sentences.append((sentence, final_score, chunk, pos))
227 # Sort by score and select top sentences
228 scored_sentences.sort(key=lambda x: x[1], reverse=True)
229 top_sentences = scored_sentences[: self.max_sentences]
231 # Reorder for coherence if requested
232 if self.reorder_sentences:
233 # Group by chunk and sort by position within chunk
234 top_sentences.sort(key=lambda x: (x[2].rank, x[3]))
236 # Build response
237 response_sentences = [s[0] for s in top_sentences]
238 response = " ".join(response_sentences)
240 # Track which chunks were used (deduplicate by source)
241 chunks_dict = {}
242 for s in top_sentences:
243 chunk = s[2]
244 if chunk.source not in chunks_dict:
245 chunks_dict[chunk.source] = chunk
246 chunks_used = list(chunks_dict.values())
248 # Build citations
249 citations = [
250 {
251 "source": chunk.source,
252 "score": chunk.score,
253 "sentences_extracted": sum(1 for s in top_sentences if s[2] == chunk),
254 }
255 for chunk in chunks_used
256 ]
258 return SynthesisResult(
259 query=query,
260 response=response,
261 strategy=SynthesisStrategy.EXTRACTIVE,
262 context_chunks=chunks_used,
263 citations=citations,
264 metadata={
265 "num_sentences": len(response_sentences),
266 "total_candidates": len(scored_sentences),
267 "query_keywords": list(query_keywords),
268 "reordered": self.reorder_sentences,
269 },
270 )