Building AI Applications with LangChain

# Basic LangChain setup
from langchain.llms import OpenAI
from langchain.chat_models import ChatOpenAI
from langchain.prompts import PromptTemplate

# Initialize LLM
llm = ChatOpenAI(temperature=0.7, model_name="gpt-3.5-turbo")

# Simple usage
response = llm.predict("What is AI Engineering?")
print(response)

# Using prompt templates
template = """You are a helpful AI assistant.
Answer the following question: {question}
Provide a clear, concise answer."""

prompt = PromptTemplate(
    input_variables=["question"],
    template=template
)

# Format and use prompt
formatted_prompt = prompt.format(question="What is prompt engineering?")
response = llm.predict(formatted_prompt)
print(response)

LangChain provides clean abstractions for working with LLMs. Prompt templates make it easy to create reusable prompts with variables.

from langchain.chains import LLMChain, SimpleSequentialChain
from langchain.prompts import PromptTemplate

# Simple chain
prompt = PromptTemplate(
    input_variables=["topic"],
    template="Write a brief explanation of {topic}"
)
chain = LLMChain(llm=llm, prompt=prompt)
result = chain.run("machine learning")

# Sequential chain (multiple steps)
# Step 1: Generate a blog post outline
outline_prompt = PromptTemplate(
    input_variables=["topic"],
    template="Create a detailed outline for a blog post about {topic}"
)
outline_chain = LLMChain(llm=llm, prompt=outline_prompt)

# Step 2: Write the blog post based on outline
post_prompt = PromptTemplate(
    input_variables=["outline"],
    template="Write a blog post based on this outline:
{outline}"
)
post_chain = LLMChain(llm=llm, prompt=post_prompt)

# Combine into sequential chain
full_chain = SimpleSequentialChain(
    chains=[outline_chain, post_chain],
    verbose=True
)

result = full_chain.run("AI Engineering")

# Router chain example (conditional)
from langchain.chains.router import MultiPromptChain
from langchain.chains.router.llm_router import LLMRouterChain

# Different prompts for different topics
prompt_infos = [
    {
        "name": "technical",
        "description": "Good for technical questions",
        "prompt_template": "Answer this technical question: {input}"
    },
    {
        "name": "general",
        "description": "Good for general questions",
        "prompt_template": "Answer this general question: {input}"
    }
]

Chains allow you to build complex workflows by combining multiple steps. Sequential chains are great for multi-step processes, while router chains enable conditional logic.

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# Create memory
memory = ConversationBufferMemory()

# Create conversation chain with memory
conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# First interaction
response1 = conversation.predict(input="Hi, my name is Alice")
print(response1)  # AI responds and remembers name

# Second interaction (AI remembers context)
response2 = conversation.predict(input="What's my name?")
print(response2)  # "Your name is Alice"

# Memory with window (only last N messages)
from langchain.memory import ConversationBufferWindowMemory

window_memory = ConversationBufferWindowMemory(
    k=2  # Keep last 2 exchanges
)

# Summary memory (for long conversations)
from langchain.memory import ConversationSummaryMemory

summary_memory = ConversationSummaryMemory(
    llm=llm,
    return_messages=True
)

# Use in chain
conversation_with_summary = ConversationChain(
    llm=llm,
    memory=summary_memory,
    verbose=True
)

Memory is essential for conversational AI. Choose the right memory type based on your needs - buffer for short conversations, summary for long ones.

from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain.utilities import SerpAPIWrapper, PythonREPL

# Create tools
search = SerpAPIWrapper()
python_repl = PythonREPL()

tools = [
    Tool(
        name="Search",
        func=search.run,
        description="Useful for answering questions about current events, weather, or general knowledge"
    ),
    Tool(
        name="Python REPL",
        func=python_repl.run,
        description="Useful for executing Python code and performing calculations"
    )
]

# Initialize agent
agent = initialize_agent(
    tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# Agent can now use tools
result = agent.run(
    "What's the weather in San Francisco? Then calculate what 15% of that temperature would be in Celsius."
)

# Custom tool example
from langchain.tools import BaseTool
from typing import Optional

class CustomCalculatorTool(BaseTool):
    name = "Calculator"
    description = "Performs mathematical calculations"
    
    def _run(self, query: str) -> str:
        try:
            result = eval(query)  # In production, use safer evaluation
            return str(result)
        except:
            return "Error: Invalid calculation"
    
    async def _arun(self, query: str) -> Optional[str]:
        return self._run(query)

# Use custom tool
custom_tools = [CustomCalculatorTool()]
agent = initialize_agent(
    custom_tools,
    llm,
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

Agents with tools can perform complex tasks by using external resources. This makes AI applications much more powerful and useful.

from langchain.callbacks import get_openai_callback
from langchain.cache import InMemoryCache
import langchain

# Enable caching
langchain.llm_cache = InMemoryCache()

# Track token usage
with get_openai_callback() as cb:
    result = agent.run("Your query here")
    print(f"Total Tokens: {cb.total_tokens}")
    print(f"Prompt Tokens: {cb.prompt_tokens}")
    print(f"Completion Tokens: {cb.completion_tokens}")
    print(f"Total Cost (USD): $\{cb.total_cost\}")

# Streaming responses
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

streaming_llm = ChatOpenAI(
    streaming=True,
    callbacks=[StreamingStdOutCallbackHandler()],
    temperature=0
)

# Error handling wrapper
def safe_agent_run(agent, query: str, max_retries: int = 3):
    for attempt in range(max_retries):
        try:
            with get_openai_callback() as cb:
                result = agent.run(query)
                return {
                    "success": True,
                    "result": result,
                    "tokens": cb.total_tokens,
                    "cost": cb.total_cost
                }
        except Exception as e:
            if attempt == max_retries - 1:
                return {
                    "success": False,
                    "error": str(e)
                }
            time.sleep(2 ** attempt)  # Exponential backoff

# Production configuration
production_agent = initialize_agent(
    tools,
    ChatOpenAI(
        temperature=0,  # More deterministic
        max_tokens=500,  # Limit response length
        timeout=30  # Timeout
    ),
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=False,  # Disable verbose in production
    max_iterations=5,  # Limit agent iterations
    max_execution_time=60  # Max execution time
)

Production applications need error handling, monitoring, caching, and performance optimizations. These patterns ensure reliable, cost-effective AI applications.