10 LangChain Real-World Projects (Email, Research, Data)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import Literal

class EmailAnalysis(BaseModel):
    category: Literal["sales", "support", "billing", "internal", "spam", "other"]
    priority: Literal["high", "medium", "low"]
    sentiment: Literal["positive", "neutral", "negative"]
    requires_response: bool
    suggested_response: str = Field(description="Draft response if requires_response is True, else empty string")
    key_points: list[str] = Field(description="3-5 main points from the email")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = JsonOutputParser(pydantic_object=EmailAnalysis)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert email analyst. Analyze the email and return structured JSON. {format_instructions}"),
    ("human", "From: {sender}\nSubject: {subject}\n\nBody:\n{body}"),
])

chain = prompt | llm | parser

# Test it
result = chain.invoke({
    "sender": "john@company.com",
    "subject": "Invoice overdue - Account #12345",
    "body": "Hi, I noticed invoice #12345 from last month hasn't been paid yet. The amount is $2,400. Can you please advise on the payment status? This is now 30 days overdue.",
    "format_instructions": parser.get_format_instructions(),
})

print(f"Category: {result['category']}")
print(f"Priority: {result['priority']}")
print(f"Draft: {result['suggested_response'][:100]}...")

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import tool
import json

@tool
def synthesize_research(findings: str) -> str:
    """
    Synthesize research findings into a structured report.
    Input should be a JSON string with keys: topic, sources (list), key_findings (list).
    """
    data = json.loads(findings)
    report = f"# Research Brief: {data['topic']}\n\n"
    report += "## Key Findings\n"
    for i, finding in enumerate(data['key_findings'], 1):
        report += f"{i}. {finding}\n"
    report += "\n## Sources\n"
    for source in data['sources']:
        report += f"- {source}\n"
    return report

tools = [
    TavilySearchResults(max_results=5, search_depth="advanced"),
    synthesize_research,
]

llm = ChatOpenAI(model="gpt-4o", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a professional research analyst. When given a research question:
1. Search for information from multiple sources (at least 3 searches)
2. Look for recent data and statistics
3. Identify conflicting information if it exists
4. Synthesize findings into a clear, factual brief
Always cite your sources with URLs."""),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True, max_iterations=8)

result = executor.invoke({
    "input": "What are the latest developments in fusion energy in 2026? Include major milestones and timeline predictions."
})
print(result["output"])

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

def build_doc_qa(file_path: str = None, url: str = None):
    # Load document
    if file_path:
        loader = PyPDFLoader(file_path)
    elif url:
        loader = WebBaseLoader(url)
    else:
        raise ValueError("Provide file_path or url")
    
    docs = loader.load()
    
    # Chunk it
    splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=100)
    chunks = splitter.split_documents(docs)
    
    # Embed and store
    vectorstore = FAISS.from_documents(chunks, OpenAIEmbeddings())
    retriever = vectorstore.as_retriever(search_kwargs={"k": 4})
    
    # Build chain
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    prompt = ChatPromptTemplate.from_template(
        "Answer based only on the context below. If unsure, say so.\n\n"
        "Context:\n{context}\n\nQuestion: {question}"
    )
    
    chain = (
        {"context": retriever | (lambda docs: "\n\n".join(d.page_content for d in docs)),
         "question": RunnablePassthrough()}
        | prompt
        | llm
        | StrOutputParser()
    )
    return chain

# Use it
qa_chain = build_doc_qa(url="https://python.org/about/")
answer = qa_chain.invoke("What is Python primarily used for?")
print(answer)

from langchain_experimental.agents.agent_toolkits import create_pandas_dataframe_agent
from langchain_openai import ChatOpenAI
import pandas as pd

# Load your data
df = pd.read_csv("sales_data.csv")
print(df.head(2))
print(f"Shape: {df.shape}")

llm = ChatOpenAI(model="gpt-4o", temperature=0)

agent = create_pandas_dataframe_agent(
    llm,
    df,
    verbose=True,
    allow_dangerous_code=True,   # required flag
    agent_type="tool-calling",
    prefix=(
        "You are a data analyst with access to a pandas DataFrame. "
        "Answer questions by writing and running Python code. "
        "Always explain your findings in plain English after showing results."
    ),
)

questions = [
    "What is the total revenue by product category, sorted highest to lowest?",
    "Which month had the highest sales, and by how much did it beat the average?",
    "Are there any outliers in the order value column? What are they?",
]

for q in questions:
    print(f"\n{'='*50}")
    print(f"Question: {q}")
    result = agent.invoke({"input": q})
    print(f"Answer: {result['output']}")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class MeetingSummary(BaseModel):
    executive_summary: str = Field(description="2-3 sentence overview")
    key_decisions: list[str] = Field(description="Decisions made in the meeting")
    action_items: list[dict] = Field(description="List of {owner, task, due_date} dicts")
    follow_up_questions: list[str] = Field(description="Open questions needing resolution")
    sentiment: str = Field(description="Overall meeting tone: productive/mixed/difficult")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = JsonOutputParser(pydantic_object=MeetingSummary)

prompt = ChatPromptTemplate.from_messages([
    ("system", "You are an expert meeting facilitator. Extract structured information from meeting transcripts. {format_instructions}"),
    ("human", "Meeting transcript:\n\n{transcript}"),
])

chain = prompt | llm | parser

# Test with a sample transcript
sample_transcript = """
Sarah: Alright, let's start the Q3 planning meeting. We have 45 minutes.
John: Before we begin, I wanted to share that our NPS score dropped from 72 to 65 this quarter.
Sarah: That's significant. Do we know why?
John: Support tickets increased 30%. The main issue seems to be the new checkout flow.
Maria: I can have a UX audit done by end of next week. We should revert the checkout temporarily.
Sarah: Agreed, let's revert today. Maria, can you own the audit and give us findings by Friday?
Maria: Yes, I'll have it ready Friday.
John: We also need to discuss the Q4 roadmap. Are we still targeting the mobile app launch?
Sarah: Yes, October 15th is still the target. Dev team, are we on track?
Mike: We're 80% there. We need one more backend sprint and QA time.
Sarah: So we're good for October 15th with buffer. Mike, please confirm the exact date with engineering by Wednesday.
"""

result = chain.invoke({
    "transcript": sample_transcript,
    "format_instructions": parser.get_format_instructions(),
})

print("Executive Summary:", result["executive_summary"])
print("\nAction Items:")
for item in result["action_items"]:
    print(f"  - {item.get('owner', 'TBD')}: {item.get('task', '')} ({item.get('due_date', 'TBD')})")

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import Literal

class CodeIssue(BaseModel):
    severity: Literal["critical", "major", "minor", "suggestion"]
    category: Literal["bug", "security", "performance", "style", "maintainability"]
    line_reference: str
    description: str
    suggested_fix: str

class CodeReview(BaseModel):
    overall_quality: Literal["excellent", "good", "needs_work", "poor"]
    summary: str
    issues: list[CodeIssue]
    positive_aspects: list[str]

llm = ChatOpenAI(model="gpt-4o", temperature=0)
parser = JsonOutputParser(pydantic_object=CodeReview)

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a senior software engineer conducting a code review.
Focus on: bugs, security vulnerabilities, performance issues, and maintainability.
Be specific and constructive. Reference line numbers or function names where possible.
{format_instructions}"""),
    ("human", "Language: {language}\n\nCode to review:\n```{language}\n{code}\n```"),
])

chain = prompt | llm | parser

# Example review
code_to_review = """
def get_user(user_id):
    conn = sqlite3.connect('users.db')
    query = f"SELECT * FROM users WHERE id = {user_id}"
    cursor = conn.execute(query)
    result = cursor.fetchone()
    conn.close()
    return result

def process_users(user_ids):
    results = []
    for uid in user_ids:
        user = get_user(uid)
        results.append(user)
    return results
"""

review = chain.invoke({
    "language": "python",
    "code": code_to_review,
    "format_instructions": parser.get_format_instructions(),
})

print(f"Overall: {review['overall_quality']}")
print(f"\nSummary: {review['summary']}")
print("\nCritical Issues:")
for issue in review['issues']:
    if issue['severity'] == 'critical':
        print(f"  [{issue['category']}] {issue['description']}")
        print(f"  Fix: {issue['suggested_fix']}")

from langchain_huggingface import HuggingFacePipeline, HuggingFaceEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.document_loaders import DirectoryLoader, TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser
from transformers import AutoTokenizer, AutoModelForCausalLM, pipeline
import torch

def build_local_chat(docs_dir: str):
    # Load all text files from a directory
    loader = DirectoryLoader(docs_dir, glob="**/*.txt", loader_cls=TextLoader)
    docs = loader.load()
    print(f"Loaded {len(docs)} documents")
    
    # Chunk
    splitter = RecursiveCharacterTextSplitter(chunk_size=400, chunk_overlap=40)
    chunks = splitter.split_documents(docs)
    
    # Local embeddings
    embeddings = HuggingFaceEmbeddings(
        model_name="sentence-transformers/all-MiniLM-L6-v2",
        model_kwargs={"device": "cuda" if torch.cuda.is_available() else "cpu"},
    )
    vectorstore = FAISS.from_documents(chunks, embeddings)
    retriever = vectorstore.as_retriever(search_kwargs={"k": 3})
    
    # Local LLM (Phi-3-mini for speed, fits on 8GB GPU)
    model_id = "microsoft/Phi-3-mini-4k-instruct"
    tokenizer = AutoTokenizer.from_pretrained(model_id)
    model = AutoModelForCausalLM.from_pretrained(
        model_id, torch_dtype=torch.float16,
        device_map="auto", trust_remote_code=True,
    )
    pipe = pipeline("text-generation", model=model, tokenizer=tokenizer,
                   max_new_tokens=256, return_full_text=False)
    llm = HuggingFacePipeline(pipeline=pipe)
    
    # RAG chain
    prompt = PromptTemplate.from_template(
        "<|user|>\nAnswer this question using only the context below. "
        "If the answer isn't in the context, say 'I don't know.'\n\n"
        "Context: {context}\n\nQuestion: {question}<|end|>\n<|assistant|>"
    )
    
    chain = (
        {"context": retriever | (lambda d: "\n".join(x.page_content for x in d)),
         "question": RunnablePassthrough()}
        | prompt | llm | StrOutputParser()
    )
    return chain

chat = build_local_chat("./my_documents")
answer = chat.invoke("What is our company's vacation policy?")
print(answer)

from langchain_openai import ChatOpenAI
from langchain_community.tools.tavily_search import TavilySearchResults
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import tool
from datetime import datetime, timedelta
import schedule
import time
import json

# Topics to monitor
WATCH_TOPICS = [
    "OpenAI GPT updates",
    "LangChain new features",
    "AI regulation news",
]

IMPORTANCE_THRESHOLD = 7  # 1-10 scale

@tool
def assess_story_importance(story_json: str) -> str:
    """
    Assess the importance of a news story on a scale of 1-10.
    Input is JSON with keys: title, summary, source.
    Returns a score and reasoning.
    """
    story = json.loads(story_json)
    # In practice this would use a separate LLM call or heuristics
    return json.dumps({
        "score": 8,
        "reasoning": f"Story about {story.get('title', 'unknown')} appears significant"
    })

@tool  
def format_alert(alert_data: str) -> str:
    """
    Format news findings into a readable alert message.
    Input is JSON with keys: topic, stories (list), date.
    """
    data = json.loads(alert_data)
    msg = f"NEWS ALERT: {data['topic']}\n"
    msg += f"Date: {data['date']}\n\n"
    for story in data.get('stories', []):
        msg += f"• {story}\n"
    return msg

def run_news_agent(topic: str) -> str:
    llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
    tools = [TavilySearchResults(max_results=5), assess_story_importance, format_alert]
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", f"""You are a news monitoring agent. Your job:
1. Search for recent news about the given topic (last 24 hours if possible)
2. Assess each story for importance (1-10)
3. Format an alert only if important stories (score >= {IMPORTANCE_THRESHOLD}) are found
4. If no important stories found, say "No significant updates found."

Today is {datetime.now().strftime('%Y-%m-%d')}."""),
        ("human", "Monitor news about: {input}"),
        MessagesPlaceholder("agent_scratchpad"),
    ])
    
    agent = create_tool_calling_agent(llm, tools, prompt)
    executor = AgentExecutor(agent=agent, tools=tools, max_iterations=5)
    result = executor.invoke({"input": topic})
    return result["output"]

def scheduled_check():
    print(f"\n{'='*60}")
    print(f"News check at {datetime.now().strftime('%Y-%m-%d %H:%M')}")
    for topic in WATCH_TOPICS:
        print(f"\nChecking: {topic}")
        alert = run_news_agent(topic)
        print(alert)

# Run immediately and then every 4 hours
scheduled_check()
schedule.every(4).hours.do(scheduled_check)

# In production, use a proper scheduler like APScheduler or Celery beat
# while True:
#     schedule.run_pending()
#     time.sleep(60)

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_community.vectorstores import FAISS
from langchain.memory import ConversationSummaryBufferMemory
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain.tools import tool
from langchain_community.tools.tavily_search import TavilySearchResults

# Build knowledge base from your policies
policies = [
    "Refund policy: Full refunds within 30 days of purchase. Partial refund (50%) for 31-60 days.",
    "Shipping: Standard 5-7 business days free. Express 2-day shipping for $12.99.",
    "Warranty: All products carry a 1-year manufacturer warranty.",
    "Account issues: Password resets via email. Account deletion requires 48-hour notice.",
]

embeddings = OpenAIEmbeddings()
kb = FAISS.from_texts(policies, embeddings)
kb_retriever = kb.as_retriever(search_kwargs={"k": 2})

@tool
def search_knowledge_base(query: str) -> str:
    """
    Search the company knowledge base for policies, procedures, and FAQs.
    Use this for questions about refunds, shipping, warranties, and account issues.
    """
    docs = kb_retriever.invoke(query)
    return "\n".join(d.page_content for d in docs)

@tool
def escalate_to_human(reason: str) -> str:
    """
    Escalate the conversation to a human agent.
    Use when the customer is upset, the issue is complex, or you cannot resolve it.
    Input should describe why escalation is needed.
    """
    # In production: create a ticket in Zendesk, send Slack notification, etc.
    ticket_id = f"TKT-{hash(reason) % 100000:05d}"
    return f"I've escalated your case to our team (Ticket #{ticket_id}). A human agent will contact you within 2 business hours."

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0.2)
memory = ConversationSummaryBufferMemory(
    llm=llm, max_token_limit=500, return_messages=True, memory_key="chat_history"
)

tools = [search_knowledge_base, escalate_to_human]

prompt = ChatPromptTemplate.from_messages([
    ("system", """You are a helpful customer support agent for TechStore.
- Always be polite and empathetic
- Use the knowledge base for policy questions
- Escalate if the customer is frustrated or the issue needs account access
- Keep responses concise (2-3 sentences max unless detail is needed)"""),
    MessagesPlaceholder("chat_history"),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, memory=memory, verbose=False)

# Simulate a conversation
turns = [
    "Hi, I bought a laptop 3 weeks ago and want to return it",
    "The screen has a dead pixel, this is really frustrating",
    "Can I get a full refund or just store credit?",
]

for message in turns:
    print(f"Customer: {message}")
    result = executor.invoke({"input": message})
    print(f"Agent: {result['output']}\n")

from langchain_openai import ChatOpenAI
from langchain_community.utilities import SQLDatabase
from langchain_community.agent_toolkits.sql.base import create_sql_agent
from langchain_community.agent_toolkits.sql.toolkit import SQLDatabaseToolkit
from langchain.agents import AgentType
import sqlite3

# Create sample database for testing
def create_sample_db():
    conn = sqlite3.connect("analytics.db")
    c = conn.cursor()
    c.execute("""CREATE TABLE IF NOT EXISTS orders (
        id INTEGER PRIMARY KEY, customer TEXT, product TEXT,
        amount REAL, date TEXT, region TEXT
    )""")
    sample_data = [
        ("Alice Chen", "Laptop Pro", 1299.99, "2026-01-15", "West"),
        ("Bob Smith", "Wireless Mouse", 49.99, "2026-01-16", "East"),
        ("Carol White", "Monitor 4K", 599.99, "2026-02-01", "West"),
        ("David Lee", "Keyboard", 129.99, "2026-02-14", "North"),
        ("Eve Martin", "Laptop Pro", 1299.99, "2026-03-01", "East"),
        ("Frank Brown", "Webcam HD", 89.99, "2026-03-15", "South"),
        ("Grace Kim", "Monitor 4K", 599.99, "2026-04-01", "West"),
        ("Henry Wang", "Wireless Mouse", 49.99, "2026-04-20", "North"),
    ]
    c.executemany(
        "INSERT INTO orders (customer, product, amount, date, region) VALUES (?,?,?,?,?)",
        sample_data
    )
    conn.commit()
    conn.close()

create_sample_db()

# Connect LangChain to the DB
db = SQLDatabase.from_uri("sqlite:///analytics.db")
llm = ChatOpenAI(model="gpt-4o", temperature=0)

toolkit = SQLDatabaseToolkit(db=db, llm=llm)

agent = create_sql_agent(
    llm=llm,
    toolkit=toolkit,
    agent_type=AgentType.OPENAI_FUNCTIONS,
    verbose=True,
    prefix="""You are a SQL expert analyst. When asked questions about data:
1. First check the table schema
2. Write an accurate SQL query
3. Run it and verify the results make sense
4. Explain findings in plain English with context
Never make up data - only report what the query returns.""",
)

# Natural language queries
queries = [
    "What is our total revenue by region?",
    "Which product generated the most revenue overall?",
    "Show me the month-over-month revenue trend for Q1 2026",
    "Which customers have spent more than $500 total?",
]

for query in queries:
    print(f"\nQuestion: {query}")
    print("-" * 40)
    result = agent.invoke({"input": query})
    print(result["output"])

# Caching to reduce API costs and latency
from langchain.globals import set_llm_cache
from langchain_community.cache import InMemoryCache, SQLiteCache

# Use SQLite cache for persistence across restarts
set_llm_cache(SQLiteCache(database_path=".langchain_cache.db"))

# Rate limiting to avoid hitting API limits
import time
from functools import wraps

def rate_limit(calls_per_minute: int):
    min_interval = 60.0 / calls_per_minute
    last_called = [0.0]
    
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            elapsed = time.time() - last_called[0]
            wait_time = min_interval - elapsed
            if wait_time > 0:
                time.sleep(wait_time)
            result = func(*args, **kwargs)
            last_called[0] = time.time()
            return result
        return wrapper
    return decorator