LangChain and AI Application Frameworks

ai_frameworks:
  langchain:
    focus: Comprehensive LLM application framework
    strengths: Extensive integrations, large community, many examples
    concerns: Complexity, abstraction overhead, rapid changes

  llamaindex:
    focus: Data ingestion and retrieval for LLMs
    strengths: Strong RAG support, structured data handling
    concerns: Narrower scope than LangChain

  semantic_kernel:
    focus: Microsoft's orchestration framework
    strengths: Enterprise integration, .NET/Python support
    concerns: Smaller community, Microsoft ecosystem focus

  haystack:
    focus: NLP pipelines and search
    strengths: Production-focused, good search integration
    concerns: Learning curve, less LLM-focused historically

  roll_your_own:
    focus: Custom implementation
    strengths: Full control, no abstraction overhead
    concerns: More code to maintain, reinventing wheels

from langchain.chat_models import ChatOpenAI
from langchain.prompts import ChatPromptTemplate
from langchain.schema.output_parser import StrOutputParser

# LLM wrapper
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# Prompt template
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant that explains technical concepts."),
    ("human", "Explain {concept} in simple terms.")
])

# Chain: prompt -> LLM -> output parser
chain = prompt | llm | StrOutputParser()

# Invoke
result = chain.invoke({"concept": "kubernetes"})

from langchain.chains import LLMChain, SequentialChain

# Simple chain
summarize_chain = LLMChain(
    llm=llm,
    prompt=ChatPromptTemplate.from_template(
        "Summarize this text in 3 bullet points:\n{text}"
    ),
    output_key="summary"
)

# Sequential chain: multiple steps
analyze_chain = LLMChain(
    llm=llm,
    prompt=ChatPromptTemplate.from_template(
        "Based on this summary, what are the key implications?\n{summary}"
    ),
    output_key="analysis"
)

overall_chain = SequentialChain(
    chains=[summarize_chain, analyze_chain],
    input_variables=["text"],
    output_variables=["summary", "analysis"]
)

result = overall_chain({"text": long_document})
# result["summary"], result["analysis"]

from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Pinecone
from langchain.chains import RetrievalQA
from langchain.text_splitter import RecursiveCharacterTextSplitter

# Document processing
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
docs = text_splitter.split_documents(documents)

# Vector store
embeddings = OpenAIEmbeddings()
vectorstore = Pinecone.from_documents(docs, embeddings, index_name="my-index")

# RAG chain
qa_chain = RetrievalQA.from_chain_type(
    llm=llm,
    chain_type="stuff",  # or "map_reduce" for large contexts
    retriever=vectorstore.as_retriever(search_kwargs={"k": 5}),
    return_source_documents=True
)

result = qa_chain({"query": "How does authentication work?"})
print(result["result"])
print(result["source_documents"])

from langchain.agents import create_openai_functions_agent, AgentExecutor
from langchain.tools import Tool
from langchain import hub

# Define tools
def search_database(query: str) -> str:
    """Search the product database."""
    # Implementation
    return f"Found products matching: {query}"

def get_order_status(order_id: str) -> str:
    """Get status of an order."""
    # Implementation
    return f"Order {order_id} is shipped"

tools = [
    Tool(
        name="search_database",
        func=search_database,
        description="Search for products in the database"
    ),
    Tool(
        name="get_order_status",
        func=get_order_status,
        description="Get the status of an order by order ID"
    )
]

# Create agent
prompt = hub.pull("hwchase17/openai-functions-agent")
agent = create_openai_functions_agent(llm, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# Run agent
result = agent_executor.invoke({
    "input": "What's the status of order 12345 and do you have any blue widgets?"
})

use_framework_when:
  rapid_prototyping:
    - Testing ideas quickly
    - Building demos
    - Exploring LLM capabilities

  standard_patterns:
    - RAG applications
    - Chatbots with memory
    - Multi-step workflows

  integration_heavy:
    - Many data sources
    - Multiple LLM providers
    - Various vector stores

  learning:
    - Understanding patterns
    - Exploring possibilities
    - Following tutorials

avoid_framework_when:
  simple_use_case:
    - Single API call
    - Basic prompt template
    - Adds unnecessary complexity

  performance_critical:
    - Framework overhead matters
    - Need fine-grained control
    - Optimizing latency

  production_stability:
    - Framework changes frequently
    - Abstractions hide issues
    - Debugging is harder

  specific_requirements:
    - Framework doesn't fit well
    - Custom patterns needed
    - Better to own the code

framework_pitfalls:
  abstraction_hiding:
    problem: Don't understand what's happening
    symptom: Debug issues are mysterious
    solution: Read framework source, understand layers

  version_churn:
    problem: Rapid breaking changes
    symptom: Code breaks on updates
    solution: Pin versions, test updates carefully

  over_engineering:
    problem: Using framework for simple tasks
    symptom: 100 lines for what could be 10
    solution: Right-size tool to task

  cargo_culting:
    problem: Copying patterns without understanding
    symptom: Code works but you don't know why
    solution: Understand before implementing

# Sometimes a simple function is better than a framework

# Instead of LangChain chain for simple task:
def summarize(text: str) -> str:
    response = openai.ChatCompletion.create(
        model="gpt-3.5-turbo",
        messages=[
            {"role": "system", "content": "Summarize in 3 bullet points."},
            {"role": "user", "content": text}
        ]
    )
    return response.choices[0].message.content

# vs LangChain equivalent (more setup, same result)

from langchain.callbacks import get_openai_callback
from langchain.callbacks.base import BaseCallbackHandler

# Track costs and tokens
with get_openai_callback() as cb:
    result = chain.invoke({"question": "What is RAG?"})
    print(f"Total tokens: {cb.total_tokens}")
    print(f"Total cost: ${cb.total_cost:.4f}")

# Custom callback for logging
class LoggingCallback(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        logger.info(f"LLM start: {prompts}")

    def on_llm_end(self, response, **kwargs):
        logger.info(f"LLM response: {response}")

    def on_chain_error(self, error, **kwargs):
        logger.error(f"Chain error: {error}")

# Test chains with mocked LLMs
from langchain.llms.fake import FakeListLLM

def test_summarize_chain():
    fake_llm = FakeListLLM(responses=["• Point 1\n• Point 2\n• Point 3"])

    chain = create_summarize_chain(llm=fake_llm)
    result = chain.invoke({"text": "Some long text..."})

    assert "Point 1" in result