Build a RAG System with Pinecone and LangGraph (Step-by-Step Guide)

Step 6 – Create Code to Connect LLM, Database, and User Input Using LangGraph

File: lgdemo.py

👉 This command:

• Prompts you to ask a question
• Converts your question into an embedding
• Searches Pinecone for relevant data
• Sends that data to the AI
• Returns a final answer

Think of this as:
“Ask questions using my data”

<?php

# ============================================
# MAIN RAG PIPELINE (ASK QUESTIONS)
# ============================================

import os
import logging
from typing import TypedDict, List
from dotenv import load_dotenv
from openai import OpenAI
from pinecone import Pinecone, ServerlessSpec
from langgraph.graph import StateGraph, END

# Load API keys
load_dotenv()

# Enable logging
logging.basicConfig(level=logging.INFO)

# ---------------------------
# DEFINE DATA STRUCTURE
# ---------------------------
class GraphState(TypedDict):
    query: str        # user question
    embedding: List[float]  # vector form of question
    results: list     # search results
    answer: str       # final AI answer

# ---------------------------
# EMBEDDING SERVICE
# ---------------------------
class EmbeddingService:
    def __init__(self, client):
        self.client = client

    def embed(self, text):
        # Convert text into vector
        res = self.client.embeddings.create(
            model="text-embedding-3-small",
            input=text
        )
        return res.data[0].embedding

# ---------------------------
# PINECONE SERVICE
# ---------------------------
class PineconeService:
    def __init__(self):
        self.pc = Pinecone(api_key=os.getenv("PINECONE_API_KEY"))
        self.index_name = os.getenv("PINECONE_INDEX_NAME")

        # Check if index exists
        existing = [i.name for i in self.pc.list_indexes()]

        if self.index_name not in existing:
            logging.info("Creating Pinecone index...")

            self.pc.create_index(
                name=self.index_name,
                dimension=1536,
                metric="cosine",
                spec=ServerlessSpec(
                    cloud="aws",
                    region=os.getenv("PINECONE_ENV")
                )
            )

        # Connect to index
        self.index = self.pc.Index(self.index_name)

    def query(self, vector):
        # Search Pinecone using vector
        return self.index.query(
            vector=vector,
            top_k=3,
            include_metadata=True
        ).matches

# ---------------------------
# LLM SERVICE
# ---------------------------
class LLMService:
    def __init__(self):
        self.client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

    def generate(self, query, context):
        # Build prompt with context
        prompt = f"""
Use the context below to answer the question.

Context:
{context}

Question:
{query}
"""

        # Ask AI for answer
        res = self.client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": prompt}]
        )

        return res.choices[0].message.content

# ---------------------------
# LANGGRAPH PIPELINE
# ---------------------------
class LangGraphService:
    def __init__(self):
        client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

        # Initialize services
        self.embedder = EmbeddingService(client)
        self.vector = PineconeService()
        self.llm = LLMService()

        # Create graph
        self.graph = StateGraph(GraphState)
        self._build()

        # Compile graph
        self.app = self.graph.compile()

    # Step 1: get question from user
    def get_query(self, state):
        return {"query": input("Enter your question: ")}

    # Step 2: convert question to vector
    def embed_query(self, state):
        return {"embedding": self.embedder.embed(state["query"])}

    # Step 3: search Pinecone
    def vector_search(self, state):
        return {"results": self.vector.query(state["embedding"])}

    # Step 4: generate answer
    def generate_answer(self, state):
        # Combine retrieved text
        context = "\n".join([r.metadata["text"] for r in state["results"]])

        # Ask AI
        answer = self.llm.generate(state["query"], context)

        return {"answer": answer}

    # Build pipeline flow
    def _build(self):
        self.graph.add_node("query", self.get_query)
        self.graph.add_node("embed", self.embed_query)
        self.graph.add_node("search", self.vector_search)
        self.graph.add_node("answer", self.generate_answer)

        # Define flow order
        self.graph.set_entry_point("query")
        self.graph.add_edge("query", "embed")
        self.graph.add_edge("embed", "search")
        self.graph.add_edge("search", "answer")
        self.graph.add_edge("answer", END)

    # Run the app
    def run(self):
        result = self.app.invoke({})

        print("\nAnswer:\n")
        print(result["answer"])

# ---------------------------
# START PROGRAM
# ---------------------------
if __name__ == "__main__":
    service = LangGraphService()
    service.run()