Coverage for /home/admin/Documents/AI/applications/lexigram-dev/lexigram/experimental/ai/lexigram-ai-rag/src/lexigram/ai/rag/synthesis/types.py: 67%
110 statements
« prev ^ index » next coverage.py v7.15.4, created at 2026-08-25 07:19 +0800
1"""Core types for response synthesis.
3This module defines the fundamental types used throughout the synthesis system,
4including synthesis strategies, results, context chunks, and quality metrics.
5"""
7from __future__ import annotations
9from dataclasses import dataclass, field
10from datetime import UTC, datetime
11from enum import StrEnum
12from typing import Any
14from lexigram.contracts.ai.chunks import ContextChunk as ContextChunkBase
17class SynthesisStrategy(StrEnum):
18 """Available synthesis strategies."""
20 DIRECT = "direct" # Simple concatenation
21 EXTRACTIVE = "extractive" # Extract relevant sentences
22 ABSTRACTIVE = "abstractive" # LLM-generated response
23 HYBRID = "hybrid" # Combined extractive + abstractive
26class OutputFormat(StrEnum):
27 """Available output formats."""
29 PLAIN_TEXT = "plain_text"
30 MARKDOWN = "markdown"
31 JSON = "json"
32 HTML = "html"
35@dataclass(frozen=True)
36class ContextChunk(ContextChunkBase):
37 """A chunk of context retrieved for synthesis.
39 Attributes:
40 text: The chunk text content
41 source: Source identifier (document ID, URL, etc.)
42 score: Relevance score (0-1)
43 metadata: Additional metadata (page number, section, etc.)
44 rank: Rank among all retrieved chunks (0-based)
45 """
47 text: str
48 source: str = "unknown"
49 score: float | None = 0.0
50 metadata: dict[str, Any] = field(default_factory=dict)
51 rank: int = 0
53 def __post_init__(self) -> None:
54 """Validate chunk data."""
55 if not self.text:
56 msg = "ContextChunk.text cannot be empty"
57 raise ValueError(msg)
58 if self.score is not None and (self.score < 0 or self.score > 1):
59 msg = "score must be between 0 and 1"
60 raise ValueError(msg)
63@dataclass
64class QualityMetrics:
65 """Quality metrics for a synthesized response.
67 Attributes:
68 faithfulness: How well grounded in context (0-1)
69 relevance: How well it answers the query (0-1)
70 coherence: How well structured and readable (0-1)
71 confidence: Overall confidence in response (0-1)
72 has_hallucinations: Whether potential hallucinations detected
73 hallucination_count: Number of potential hallucinations
74 """
76 faithfulness: float = 0.0
77 relevance: float = 0.0
78 coherence: float = 0.0
79 confidence: float = 0.0
80 has_hallucinations: bool = False
81 hallucination_count: int = 0
83 def __post_init__(self) -> None:
84 """Validate metrics."""
85 for name, value in [
86 ("faithfulness", self.faithfulness),
87 ("relevance", self.relevance),
88 ("coherence", self.coherence),
89 ("confidence", self.confidence),
90 ]:
91 if value < 0 or value > 1:
92 msg = f"{name} must be between 0 and 1"
93 raise ValueError(msg)
95 @property
96 def is_high_quality(self) -> bool:
97 """Check if response meets high quality thresholds."""
98 return (
99 self.faithfulness >= 0.7
100 and self.relevance >= 0.7
101 and self.coherence >= 0.7
102 and not self.has_hallucinations
103 )
105 @property
106 def average_score(self) -> float:
107 """Calculate average of all metric scores."""
108 return (
109 self.faithfulness + self.relevance + self.coherence + self.confidence
110 ) / 4
112 def to_dict(self) -> dict[str, Any]:
113 """Convert to dictionary."""
114 return {
115 "faithfulness": self.faithfulness,
116 "relevance": self.relevance,
117 "coherence": self.coherence,
118 "confidence": self.confidence,
119 "has_hallucinations": self.has_hallucinations,
120 "hallucination_count": self.hallucination_count,
121 "is_high_quality": self.is_high_quality,
122 "average_score": self.average_score,
123 }
126@dataclass
127class SynthesisResult:
128 """Result of response synthesis.
130 Attributes:
131 query: The original query
132 response: The synthesized response text
133 strategy: Strategy used for synthesis
134 context_chunks: Chunks used in synthesis
135 quality_metrics: Quality assessment metrics
136 sources: List of sources used
137 citations: Citation information
138 metadata: Additional metadata
139 created_at: Timestamp of synthesis
140 """
142 query: str
143 response: str
144 strategy: SynthesisStrategy
145 context_chunks: list[ContextChunk] = field(default_factory=list)
146 quality_metrics: QualityMetrics | None = None
147 sources: list[str] = field(default_factory=list)
148 citations: list[dict[str, Any]] = field(default_factory=list)
149 metadata: dict[str, Any] = field(default_factory=dict)
150 created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
152 def __post_init__(self) -> None:
153 """Validate and extract sources."""
154 if not self.query:
155 msg = "Query cannot be empty"
156 raise ValueError(msg)
157 if not self.response:
158 msg = "Response cannot be empty"
159 raise ValueError(msg)
161 # Extract unique sources from chunks if not provided
162 if not self.sources and self.context_chunks:
163 self.sources = list({chunk.source for chunk in self.context_chunks})
165 @property
166 def num_chunks_used(self) -> int:
167 """Number of context chunks used."""
168 return len(self.context_chunks)
170 @property
171 def is_high_confidence(self) -> bool:
172 """Check if response has high confidence."""
173 if not self.quality_metrics:
174 return False
175 return self.quality_metrics.confidence >= 0.7
177 @property
178 def is_faithful(self) -> bool:
179 """Check if response is faithful to context."""
180 if not self.quality_metrics:
181 return False
182 return self.quality_metrics.faithfulness >= 0.7
184 def to_dict(self) -> dict[str, Any]:
185 """Convert to dictionary."""
186 return {
187 "query": self.query,
188 "response": self.response,
189 "strategy": self.strategy.value,
190 "num_chunks_used": self.num_chunks_used,
191 "sources": self.sources,
192 "citations": self.citations,
193 "quality_metrics": (
194 self.quality_metrics.to_dict() if self.quality_metrics else None
195 ),
196 "metadata": self.metadata,
197 "created_at": self.created_at.isoformat(),
198 }
201@dataclass
202class SynthesisConfig:
203 """Configuration for response synthesis.
205 Attributes:
206 strategy: Synthesis strategy to use
207 max_context_length: Maximum context length in tokens
208 max_response_length: Maximum response length in tokens
209 include_citations: Whether to include citations
210 output_format: Desired output format
211 quality_check: Whether to run quality checks
212 min_confidence: Minimum confidence threshold
213 metadata: Additional configuration metadata
214 """
216 enabled: bool = True
217 strategy: SynthesisStrategy = SynthesisStrategy.HYBRID
218 model: str | None = None
219 max_context_length: int = 4000
220 max_response_length: int = 500
221 include_citations: bool = True
222 output_format: OutputFormat = OutputFormat.MARKDOWN
223 quality_check: bool = True
224 min_confidence: float = 0.5
225 error_strategy: str = "graceful"
226 fallback_strategy: SynthesisStrategy = SynthesisStrategy.DIRECT
227 metadata: dict[str, Any] = field(default_factory=dict)
229 def __post_init__(self) -> None:
230 """Validate configuration."""
231 if self.max_context_length <= 0:
232 msg = "max_context_length must be positive"
233 raise ValueError(msg)
234 if self.max_response_length <= 0:
235 msg = "max_response_length must be positive"
236 raise ValueError(msg)
237 if self.min_confidence < 0 or self.min_confidence > 1:
238 msg = "min_confidence must be between 0 and 1"
239 raise ValueError(msg)