Create app.py

app.py

@@ -0,0 +1,134 @@
+import os
+import sys
+import logging
+import gradio as gr
+import requests
+from pinecone import Pinecone, ServerlessSpec
+from haystack.components.embedders import OpenAIAPIDocumentEmbedder, OpenAIAPIQueryEmbedder
+from haystack.components.writers import DocumentWriter
+from haystack.document_stores.pinecone import PineconeDocumentStore
+from haystack.components.retrievers.in_memory import InMemoryEmbeddingRetriever
+from haystack.components.retrievers.pinecone import PineconeEmbeddingRetriever
+from haystack import Pipeline
+from haystack.components.routers import DocumentJoiner
+from haystack.components.generators import OpenAIGenerator
+from haystack.components.builders import PromptBuilder
+from haystack.components.converters import TextFileToDocument
+from haystack.components.preprocessors import DocumentSplitter
+from haystack.components.routers import FileTypeRouter
+from haystack.document_stores.in_memory import InMemoryDocumentStore
+
+# --- Logging ---
+logging.basicConfig(stream=sys.stdout, level=logging.INFO)
+
+# --- Environment Variables ---
+api_key = os.getenv("PINECONE_API_KEY")
+openai_api_key = os.getenv("OPENAI_API_KEY")
+
+if not api_key:
+    raise ValueError("Please set the PINECONE_API_KEY as an environment variable.")
+if not openai_api_key:
+    raise ValueError("Please set the OPENAI_API_KEY as an environment variable.")
+os.environ["OPENAI_API_KEY"] = openai_api_key
+
+# --- Pinecone Setup ---
+index_name = "quickstart"
+dimension = 1536
+pc = Pinecone(api_key=api_key)
+
+# Create index if not exists
+if index_name not in [idx['name'] for idx in pc.list_indexes()]:
+    pc.create_index(
+        name=index_name,
+        dimension=dimension,
+        metric="euclidean",
+        spec=ServerlessSpec(cloud="aws", region="us-east-1")
+    )
+    
+# --- Document Loading and Processing ---
+os.makedirs("data/paul_graham", exist_ok=True)
+file_path = "data/paul_graham/paul_graham_essay.txt"
+if not os.path.exists(file_path):
+    url = "https://raw.githubusercontent.com/run-llama/llama_index/main/docs/docs/examples/data/paul_graham/paul_graham_essay.txt"
+    r = requests.get(url)
+    with open(file_path, "w") as f:
+        f.write(r.text)
+
+# --- Haystack Pipeline for Indexing ---
+document_store = PineconeDocumentStore(api_key=api_key, index=index_name)
+
+indexing_pipeline = Pipeline()
+indexing_pipeline.add_component("converter", TextFileToDocument())
+indexing_pipeline.add_component("splitter", DocumentSplitter(split_by="word", split_length=100))
+indexing_pipeline.add_component("embedder", OpenAIAPIDocumentEmbedder())
+indexing_pipeline.add_component("writer", DocumentWriter(document_store))
+
+indexing_pipeline.connect("converter.documents", "splitter.documents")
+indexing_pipeline.connect("splitter.documents", "embedder.documents")
+indexing_pipeline.connect("embedder.documents", "writer.documents")
+
+if document_store.count_documents() == 0:
+    logging.info("Indexing the document...")
+    indexing_pipeline.run({"converter": {"sources": [file_path]}})
+    logging.info("Indexing complete.")
+
+# --- Haystack Query Pipeline ---
+template = """
+Given the following context, answer the user's question.
+If the context isn't sufficient, say that you don't have enough information.
+
+Context:
+{% for doc in documents %}
+    {{ doc.content }}
+{% endfor %}
+
+Question: {{ query }}
+"""
+query_pipeline = Pipeline()
+query_pipeline.add_component("embedder", OpenAIAPIQueryEmbedder())
+query_pipeline.add_component("retriever", PineconeEmbeddingRetriever(document_store=document_store))
+query_pipeline.add_component("prompt_builder", PromptBuilder(template=template))
+query_pipeline.add_component("llm", OpenAIGenerator(api_key=openai_api_key))
+
+query_pipeline.connect("embedder.query", "retriever.query")
+query_pipeline.connect("retriever.documents", "prompt_builder.documents")
+query_pipeline.connect("prompt_builder", "llm")
+
+# --- Query Function ---
+def ask_question(prompt):
+    try:
+        results = query_pipeline.run({"embedder": {"text": prompt}, "prompt_builder": {"query": prompt}})
+        response = results["llm"]["replies"][0]
+        return str(response)
+    except Exception as e:
+        return f"❌ Error: {str(e)}"
+
+# --- Gradio UI ---
+with gr.Blocks(css="""body {    background-color: #f5f5dc;    font-family: 'Georgia', 'Merriweather', serif;}h1, h2, h3 {    color: #4e342e;}.gr-box, .gr-column, .gr-group {    border-radius: 15px;    padding: 20px;    background-color: #fffaf0;    box-shadow: 2px 4px 14px rgba(0, 0, 0, 0.1);    margin-top: 10px;}textarea, input[type="text"] {    background-color: #fffaf0;    border: 1px solid #d2b48c;    color: #4e342e;    border-radius: 8px;}button {    background-color: #a1887f;    color: white;    font-weight: bold;    border-radius: 8px;    transition: background-color 0.3s ease;}button:hover {    background-color: #8d6e63;}.gr-button {    border-radius: 8px !important;}""") as demo:
+    with gr.Column():
+        gr.Markdown("""
+        <div style='text-align: center;'>
+            <h1>🧠 Paul Graham Essay Q&A</h1>
+            <div style='font-size: 1.1em; color: #6d4c41; margin-bottom: 1em;'>
+                Explore insights from Paul Graham's essay using semantic search powered by <strong>Haystack</strong> + <strong>Pinecone</strong>.
+            </div>
+        </div>
+        """)
+        with gr.Accordion("ℹ️ What is Pinecone Vector Indexing?", open=False):
+            gr.Markdown("""**Pinecone** is a vector database that stores document embeddings (numeric representations of meaning). When you ask a question, it's converted into a vector and compared against stored vectors to find the most relevant answers — even if they don't match word-for-word.""")
+        gr.Markdown("### 📖 Ask your question below:")
+        with gr.Group():
+            with gr.Row():
+                user_input = gr.Textbox(
+                    placeholder="E.g., What does Paul Graham say about startups?",
+                    label="Your Question",
+                    lines=2
+                )
+            with gr.Row():
+                output = gr.Textbox(label="Answer", lines=6)
+            with gr.Row():
+                submit_btn = gr.Button("🔍 Search Essay")
+                clear_btn = gr.Button("🧹 Clear")
+            submit_btn.click(fn=ask_question, inputs=user_input, outputs=output)
+            clear_btn.click(fn=lambda: ("", ""), inputs=None, outputs=[user_input, output])
+demo.launch()