File size: 4,238 Bytes
cebfa40
 
 
 
eb31fdf
cebfa40
 
 
 
 
 
 
1eecc1c
 
cebfa40
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1eecc1c
cebfa40
1eecc1c
 
4076e48
1eecc1c
 
 
4076e48
 
 
 
 
 
 
1eecc1c
4076e48
1eecc1c
 
4076e48
 
 
 
 
 
 
 
 
1eecc1c
 
 
4076e48
 
 
 
 
 
 
 
 
 
 
1eecc1c
 
4076e48
 
 
 
 
1eecc1c
4076e48
1eecc1c
 
 
 
 
4076e48
1eecc1c
4076e48
1eecc1c
 
 
cebfa40
 
 
 
 
 
 
 
 
 
 
 
 
 
 
0fdec96
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
import re
import logging
from typing import List

logger = logging.getLogger("axiom.postprocessor")


class PostProcessor:
    """
    Cleans generated text and injects source citations.
    """

    def __init__(self, groq_gen=None):
        self.groq_gen = groq_gen
        logger.info("Ready.")

    def _remove_repetition(self, text: str) -> str:
        """Removes HTML artifacts and duplicate sentences."""

        # Strip any HTML tags that leaked through
        text = re.sub(r'<[^>]+>', ' ', text)
        text = re.sub(r'\s+', ' ', text).strip()

        sentences = text.split(". ")
        seen      = set()
        cleaned   = []

        for sentence in sentences:
            normalized = sentence.strip().lower()
            if normalized and normalized not in seen:
                seen.add(normalized)
                cleaned.append(sentence.strip())

        return ". ".join(cleaned)

    def _inject_citations(self, text: str, sources: List[str]) -> str:
        """Appends source list at the bottom of the generated text."""
        if not sources:
            return text

        citation_block = "\n\nSources:\n"
        for i, source in enumerate(sources):
            citation_block += f"  [{i+1}] {source}\n"

        return text + citation_block

    def _check_faithfulness(self, generated: str, context: str) -> dict:
        """
        LLM-based faithfulness check.
        """
        if not self.groq_gen:
            return {
                "faithfulness_score": 0.5,
                "supported_sentences": 0,
                "unsupported_sentences": 0
            }

        # Truncate context to avoid token limits causing failures
        truncated_context = context[:3000] if len(context) > 3000 else context
        truncated_generated = generated[:1000] if len(generated) > 1000 else generated

        prompt = f"""You are a strict fact-checker. Evaluate if the Generated Text is faithful to the Context.

Context:
{truncated_context}

Generated Text:
{truncated_generated}

Rules:
- faithfulness_score: 1.0 means every claim is supported by context
- faithfulness_score: 0.0 means major claims are not in context
- Count each sentence as supported or unsupported

Respond ONLY with valid JSON, no markdown, no explanation:
{{"faithfulness_score": 0.95, "supported_sentences": 4, "unsupported_sentences": 0}}"""

        try:
            res = self.groq_gen.generate(query=prompt, context="")
            # More aggressive cleaning
            clean_res = res.strip()
            clean_res = re.sub(r'```json\s*', '', clean_res)
            clean_res = re.sub(r'```\s*', '', clean_res)
            clean_res = clean_res.strip()

            # Find JSON object in response even if there's extra text
            json_match = re.search(r'\{[^}]+\}', clean_res)
            if json_match:
                clean_res = json_match.group()

            import json
            data = json.loads(clean_res)

            score = float(data.get("faithfulness_score", 0.5))
            # Clamp between 0 and 1
            score = max(0.0, min(1.0, score))

            return {
                "faithfulness_score": score,
                "supported_sentences": int(data.get("supported_sentences", 0)),
                "unsupported_sentences": int(data.get("unsupported_sentences", 0))
            }
        except Exception as e:
            logger.error(f"Faithfulness check failed: {e}")
            # Return 0.5 instead of 0.0 so badge still shows
            return {
                "faithfulness_score": 0.5,
                "supported_sentences": 0,
                "unsupported_sentences": 0
            }

    def process(self, generated: str, sources: List[str], context: str) -> dict:
        """
        Full post-processing pipeline.
        Returns cleaned text + citations + faithfulness check.
        """
        cleaned        = self._remove_repetition(generated)
        with_citations = self._inject_citations(cleaned, sources)
        faithfulness   = self._check_faithfulness(cleaned, context)

        return {
            "final_output": with_citations,
            "clean_output": cleaned,
            "sources":      sources,
            "faithfulness": faithfulness
        }