File size: 5,046 Bytes
4abd84c
 
 
 
 
35913c9
 
 
 
 
5bc045f
 
35913c9
 
4abd84c
e76dfcc
 
4abd84c
e76dfcc
 
 
5bc045f
 
 
35913c9
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
5bc045f
 
 
35913c9
5bc045f
 
35913c9
 
 
 
 
 
5bc045f
35913c9
5bc045f
35913c9
 
 
 
 
5bc045f
 
 
35913c9
5bc045f
35913c9
 
 
 
5bc045f
4abd84c
e76dfcc
 
 
4abd84c
5bc045f
e76dfcc
 
4abd84c
 
 
 
 
 
 
 
 
 
e76dfcc
 
 
 
 
 
 
 
 
 
 
 
 
4abd84c
e76dfcc
 
 
4abd84c
e76dfcc
 
 
 
 
 
 
 
4abd84c
e76dfcc
 
 
 
 
 
 
 
 
 
 
 
4abd84c
 
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
127
128
129
130
131
132
import gradio as gr
from typing import List, Tuple
from src.config import Settings
from src.retrieval.search import retrieve, format_citations
from src.llm.answer import compose_answer
from src.retrieval.query_handler import (
    classify_query, 
    QueryType, 
    keyword_search_documents,
    generate_greeting_response,
    generate_semantic_analysis,
    detect_result_count
)
from src.utils import load_pickle

def _ask(query: str, history: List, settings, embedder, vstore) -> tuple[List, List[dict]]:
    """Process a chat message and return updated history + citations."""
    if not query or not query.strip():
        history.append((query, "Please enter a question about the documents."))
        return history, []
    
    # Detect how many results the user wants
    result_count = detect_result_count(query)
    
    # Classify the query type
    query_type = classify_query(query)
    
    # Handle greetings
    if query_type == QueryType.GREETING:
        answer = generate_greeting_response()
        history.append((query, answer))
        return history, []
    
    # Handle keyword search
    if query_type == QueryType.KEYWORD_SEARCH:
        # Load all documents for keyword search
        all_docs = load_pickle(settings.docs_path)
        hits = keyword_search_documents(query, all_docs)
        
        if not hits:
            history.append((query, "No documents found containing those keywords."))
            return history, []
        
        # Limit results to detected count
        hits = hits[:result_count]
        
        # Format keyword search results
        answer = f"**🔍 Keyword Search Results** (Showing {len(hits)} results)\n\n"
        for i, (doc, score) in enumerate(hits, 1):
            source = "unknown"
            body = doc
            if "[Source:" in doc:
                parts = doc.rsplit("[Source:", 1)
                body = parts[0].strip()
                source = "[Source:" + parts[1]
            answer += f"**{i}.** {body}\n\n*{source}*\n*keyword match score: {score:.3f}*\n\n---\n\n"
        
        citations = format_citations(hits, max_items=result_count)
        history.append((query, answer))
        return history, citations
    
    # Handle semantic analysis
    if query_type == QueryType.SEMANTIC_ANALYSIS:
        # For semantic analysis, limit to 3 max for readability
        analysis_count = min(result_count, 3)
        hits = retrieve(query, vstore=vstore, embedder=embedder, k=analysis_count)
        answer = generate_semantic_analysis(query, hits)
        citations = format_citations(hits, max_items=analysis_count)
        history.append((query, answer))
        return history, citations
    
    # Handle regular document questions (default)
    hits: List[Tuple[str, float]] = retrieve(query, vstore=vstore, embedder=embedder, k=result_count)
    if not hits:
        history.append((query, "No relevant passages found. Try adjusting your query."))
        return history, []
    
    answer = compose_answer(query, hits, settings)
    citations = format_citations(hits, max_items=result_count)
    history.append((query, answer))
    return history, citations

def build_app(settings: Settings, embedder, vstore):
    with gr.Blocks(title=settings.title) as demo:
        gr.Markdown(f"# {settings.title}\n{settings.description}")
        gr.Markdown(
            f"**Mode:** `{settings.mode}` "
            + ("— no LLM used, showing excerpts only. LLM will be added later for summarization" if settings.mode == "retrieval"
               else "— retrieval + summarizer enabled.")
        )

        # Chat interface
        chatbot = gr.Chatbot(label="Conversation", height=400)
        with gr.Row():
            query = gr.Textbox(
                label="",
                placeholder="Ask a question about the documents...",
                scale=4,
                container=False
            )
            submit_btn = gr.Button("Send", variant="primary", scale=1)
        
        # Citations accordion below chat
        with gr.Accordion("Citations (top matches from last query)", open=False):
            citations = gr.JSON(label="Source & similarity")
        
        # Clear button
        clear_btn = gr.Button("Clear Chat", size="sm")

        def on_submit(message, history):
            """Handle message submission."""
            new_history, new_citations = _ask(message, history or [], settings, embedder, vstore)
            return new_history, new_citations, ""  # Return empty string to clear input
        
        def clear_chat():
            """Clear the chat history."""
            return [], []

        # Wire up the interactions
        submit_btn.click(
            on_submit,
            inputs=[query, chatbot],
            outputs=[chatbot, citations, query]
        )
        query.submit(
            on_submit,
            inputs=[query, chatbot],
            outputs=[chatbot, citations, query]
        )
        clear_btn.click(clear_chat, outputs=[chatbot, citations])
    
    return demo