llms-generative-ai

What this skill does

# LLMs & Generative AI

Production-grade LLM applications with prompt engineering, RAG systems, and modern AI development patterns.

## Quick Start

```python
# Production RAG System with LangChain (2024-2025)
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# Initialize components
llm = ChatOpenAI(model="gpt-4-turbo-preview", temperature=0)
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")

# Document processing
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", ". ", " ", ""]
)

documents = text_splitter.split_documents(raw_documents)

# Vector store
vectorstore = Chroma.from_documents(
    documents=documents,
    embedding=embeddings,
    persist_directory="./chroma_db"
)
retriever = vectorstore.as_retriever(
    search_type="mmr",  # Maximum Marginal Relevance
    search_kwargs={"k": 5, "fetch_k": 10}
)

# RAG chain
template = """Answer the question based only on the following context:

Context: {context}

Question: {question}

Answer thoughtfully and cite specific parts of the context."""

prompt = ChatPromptTemplate.from_template(template)

rag_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | StrOutputParser()
)

# Query
response = rag_chain.invoke("What are the key features?")
print(response)
```

## Core Concepts

### 1. Prompt Engineering Patterns

```python
from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate

# System prompt design
system_prompt = """You are an expert data analyst assistant.

CAPABILITIES:
- Analyze data patterns and trends
- Generate SQL queries
- Explain statistical concepts

CONSTRAINTS:
- Only use information provided in the context
- Acknowledge uncertainty when relevant
- Format outputs in clear, structured way

OUTPUT FORMAT:
- Start with a brief summary
- Use bullet points for key findings
- Include confidence level (high/medium/low)
"""

# Few-shot prompting
examples = [
    {"input": "What's the average order value?",
     "output": "```sql\nSELECT AVG(total_amount) as avg_order_value\nFROM orders\nWHERE status = 'completed';\n```"},
    {"input": "Show top customers by revenue",
     "output": "```sql\nSELECT customer_id, SUM(total_amount) as revenue\nFROM orders\nGROUP BY customer_id\nORDER BY revenue DESC\nLIMIT 10;\n```"}
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}")
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples
)

# Chain of Thought prompting
cot_prompt = """Let's solve this step by step:

Question: {question}

Step 1: Identify the key components
Step 2: Break down the problem
Step 3: Apply relevant knowledge
Step 4: Synthesize the answer

Reasoning:"""

# Self-consistency (multiple reasoning paths)
async def self_consistent_answer(question: str, n_samples: int = 5) -> str:
    responses = await asyncio.gather(*[
        llm.ainvoke(question) for _ in range(n_samples)
    ])
    # Majority voting or aggregation
    return aggregate_responses(responses)
```

### 2. Advanced RAG Patterns

```python
from langchain.retrievers import ContextualCompressionRetriever
from langchain.retrievers.document_compressors import LLMChainExtractor
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever

# Hybrid search (dense + sparse)
bm25_retriever = BM25Retriever.from_documents(documents)
bm25_retriever.k = 5

chroma_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, chroma_retriever],
    weights=[0.4, 0.6]
)

# Contextual compression
compressor = LLMChainExtractor.from_llm(llm)
compression_retriever = ContextualCompressionRetriever(
    base_compressor=compressor,
    base_retriever=ensemble_retriever
)

# Parent document retriever (for better context)
from langchain.retrievers import ParentDocumentRetriever
from langchain.storage import InMemoryStore

parent_splitter = RecursiveCharacterTextSplitter(chunk_size=2000)
child_splitter = RecursiveCharacterTextSplitter(chunk_size=400)

store = InMemoryStore()
parent_retriever = ParentDocumentRetriever(
    vectorstore=vectorstore,
    docstore=store,
    child_splitter=child_splitter,
    parent_splitter=parent_splitter
)

# Self-querying retriever
from langchain.retrievers.self_query.base import SelfQueryRetriever
from langchain.chains.query_constructor.base import AttributeInfo

metadata_field_info = [
    AttributeInfo(name="source", description="Document source", type="string"),
    AttributeInfo(name="date", description="Creation date", type="date"),
    AttributeInfo(name="category", description="Document category", type="string"),
]

self_query_retriever = SelfQueryRetriever.from_llm(
    llm=llm,
    vectorstore=vectorstore,
    document_contents="Technical documentation",
    metadata_field_info=metadata_field_info
)
```

### 3. Agents and Tool Use

```python
from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool, StructuredTool
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import Optional

# Define tools with Pydantic schemas
class SQLQueryInput(BaseModel):
    query: str = Field(description="SQL query to execute")
    limit: Optional[int] = Field(default=100, description="Max rows to return")

def execute_sql(query: str, limit: int = 100) -> str:
    """Execute SQL query against the database."""
    # Validate query (prevent injection)
    if any(kw in query.upper() for kw in ["DROP", "DELETE", "UPDATE", "INSERT"]):
        return "Error: Only SELECT queries allowed"

    result = db.execute(f"{query} LIMIT {limit}")
    return result.to_markdown()

sql_tool = StructuredTool.from_function(
    func=execute_sql,
    name="sql_executor",
    description="Execute SQL queries against the data warehouse",
    args_schema=SQLQueryInput
)

# Calculator tool
def calculate(expression: str) -> str:
    """Evaluate mathematical expression."""
    try:
        # Safe eval with limited scope
        allowed_names = {"abs": abs, "round": round, "sum": sum}
        return str(eval(expression, {"__builtins__": {}}, allowed_names))
    except Exception as e:
        return f"Error: {e}"

calc_tool = Tool.from_function(
    func=calculate,
    name="calculator",
    description="Evaluate mathematical expressions"
)

# Create agent
tools = [sql_tool, calc_tool]
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    max_iterations=5,
    early_stopping_method="generate"
)

result = agent_executor.invoke({"input": "What's the total revenue for Q4 2024?"})
```

### 4. Structured Output

```python
from langchain_core.pydantic_v1 import BaseModel, Field
from typing import List, Optional
from langchain.output_parsers import PydanticOutputParser

# Define output schema
class DataInsight(BaseModel):
    title: str = Field(description="Brief title of the insight")
    description: str = Field(description="Detailed explanation")
    confidence: float = Field(description="Confidence score 0-1")
    data_points: List[str] = Field(description="Supporting data points")
    recommendations: Optional[List[str]] = Field(description="Action items")

class AnalysisReport(BaseModel):
    summary: str = Field(description="Executive summary")
    insights: List[DataInsight] = Field(description="Key insights found")
    methodology: str = Field(description="Analysis approach used")

# Parser
parser = PydanticOutputParser(pydan