pip install langchain langchain-openai langchain-community faiss-cpu

from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage

# Initialize model — similar to OpenAI client but with LangChain interface
llm = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0,
    max_tokens=1024,
    # api_key read from OPENAI_API_KEY env var
)

# Direct invocation
response = llm.invoke([
    SystemMessage(content="You are a concise Python expert."),
    HumanMessage(content="What is the difference between a list and a tuple?")
])
print(response.content)
print(f"Tokens used: {response.usage_metadata}")

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# Reusable prompt template
code_review_prompt = ChatPromptTemplate.from_messages([
    ("system", """You are an expert Python code reviewer.
Review the provided code for:
- Correctness and logic errors
- Performance issues
- Security vulnerabilities
- PEP 8 compliance
Be specific and actionable. Format as a numbered list."""),
    ("human", "Review this code:\n\n```python\n{code}\n```")
])

# Template with conversation history placeholder
chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful {role}. Be concise and accurate."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate

# Basic LCEL chain: prompt | model | parser
code_chain = (
    code_review_prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

# Invoke
review = code_chain.invoke({
    "code": """
def find_user(users, name):
    for i in range(len(users)):
        if users[i]['name'] == name:
            return users[i]
"""
})
print(review)

# Stream
for chunk in code_chain.stream({"code": "x = [i for i in range(1000000)]"}):
    print(chunk, end="", flush=True)

# Batch — process multiple items efficiently
reviews = code_chain.batch([
    {"code": "import *"},
    {"code": "password = 'admin123'"},
])

from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field
from typing import Optional

class CodeReview(BaseModel):
    """Structured code review output."""
    severity: str = Field(description="Overall severity: critical/major/minor/pass")
    issues: list[str] = Field(description="List of specific issues found")
    suggestions: list[str] = Field(description="Actionable improvement suggestions")
    estimated_fix_time: Optional[str] = Field(description="Rough time to fix all issues")

# Structured output via .with_structured_output()
structured_llm = ChatOpenAI(model="gpt-4o-mini").with_structured_output(CodeReview)

structured_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a code reviewer. Analyze the code and provide structured feedback."),
    ("human", "Review:\n```python\n{code}\n```")
])

structured_chain = structured_prompt | structured_llm

result = structured_chain.invoke({"code": "def calc(x): return x*x*x"})
print(f"Severity: {result.severity}")
print(f"Issues: {result.issues}")
print(f"Suggestions: {result.suggestions}")

from langchain_core.chat_history import BaseChatMessageHistory
from langchain_core.messages import AIMessage, HumanMessage
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_openai import ChatOpenAI

# In-memory chat history store (replace with Redis/database for production)
store: dict = {}

def get_session_history(session_id: str) -> BaseChatMessageHistory:
    from langchain_community.chat_message_histories import ChatMessageHistory
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

# Base chain
base_prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful Python tutor. Be concise."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

base_chain = base_prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

# Wrap with memory
chain_with_memory = RunnableWithMessageHistory(
    base_chain,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history"
)

# Use session_id to maintain separate conversations per user
config = {"configurable": {"session_id": "user-123"}}

print(chain_with_memory.invoke({"input": "What is list comprehension?"}, config=config))
print(chain_with_memory.invoke({"input": "Can you show an example with filtering?"}, config=config))
print(chain_with_memory.invoke({"input": "How does this differ from filter()?"}, config=config))

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.document_loaders import PyPDFLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from pathlib import Path
import os

class DocumentQA:
    """RAG-based document Q&A system built with LangChain + OpenAI."""

    def __init__(self, model: str = "gpt-4o-mini"):
        self.embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
        self.llm = ChatOpenAI(model=model, temperature=0)
        self.vectorstore = None
        self.retriever = None

        self.prompt = ChatPromptTemplate.from_messages([
            ("system", """You are a precise assistant that answers questions based only on the provided context.
If the answer is not in the context, say "I cannot find this in the provided documents."
Do not make up information or use external knowledge.

Context:
{context}"""),
            ("human", "{question}")
        ])

    def load_documents(self, file_paths: list[str]) -> int:
        """Load and index documents from file paths."""
        documents = []

        for path in file_paths:
            if path.endswith(".pdf"):
                loader = PyPDFLoader(path)
            else:
                loader = TextLoader(path)
            documents.extend(loader.load())

        # Split into chunks
        splitter = RecursiveCharacterTextSplitter(
            chunk_size=1000,
            chunk_overlap=200,
            separators=["\n\n", "\n", ". ", " "]
        )
        chunks = splitter.split_documents(documents)

        # Create vector store
        self.vectorstore = FAISS.from_documents(chunks, self.embeddings)
        self.retriever = self.vectorstore.as_retriever(
            search_kwargs={"k": 4}  # Return top 4 chunks
        )

        print(f"Indexed {len(chunks)} chunks from {len(documents)} documents")
        return len(chunks)

    def _format_docs(self, docs) -> str:
        """Format retrieved documents into a single context string."""
        formatted = []
        for i, doc in enumerate(docs, 1):
            source = doc.metadata.get("source", "Unknown")
            page = doc.metadata.get("page", "")
            header = f"[Document {i} — {Path(source).name}{f', page {page}' if page else ''}]"
            formatted.append(f"{header}\n{doc.page_content}")
        return "\n\n---\n\n".join(formatted)

    def build_chain(self):
        """Build the RAG chain using LCEL."""
        self.chain = (
            {
                "context": self.retriever | self._format_docs,
                "question": RunnablePassthrough()
            }
            | self.prompt
            | self.llm
            | StrOutputParser()
        )

    def ask(self, question: str) -> str:
        """Answer a question using retrieved context."""
        if not self.chain:
            self.build_chain()
        return self.chain.invoke(question)

    def ask_with_sources(self, question: str) -> dict:
        """Answer a question and return sources."""
        # Get retrieved documents separately
        docs = self.retriever.invoke(question)
        sources = [
            {
                "source": doc.metadata.get("source", "Unknown"),
                "page": doc.metadata.get("page", "N/A"),
                "preview": doc.page_content[:200] + "..."
            }
            for doc in docs
        ]

        answer = self.ask(question)
        return {"answer": answer, "sources": sources}


# Usage
qa = DocumentQA(model="gpt-4o-mini")

# Load your documents
chunk_count = qa.load_documents([
    "./engineering_spec.pdf",
    "./api_documentation.txt"
])

qa.build_chain()

# Simple Q&A
answer = qa.ask("What authentication method does the API use?")
print(answer)

# Q&A with sources
result = qa.ask_with_sources("What are the rate limits?")
print(f"Answer: {result['answer']}")
print(f"Sources: {[s['source'] for s in result['sources']]}")