LangChain: Building LLM-Powered Applications

pip install langchain langchain-community langchain-openai langchain-anthropic

from langchain_core.prompts import PromptTemplate, ChatPromptTemplate

# Basic prompt template
template = PromptTemplate.from_template(
    "You are a {role}. Answer the question: {question}"
)
prompt = template.invoke({"role": "cybersecurity expert", "question": "What is XSS?"})
print(prompt.text)

# Chat prompt template (with system message)
chat_template = ChatPromptTemplate.from_messages([
    ("system", "You are a {role} at DodaTech."),
    ("human", "{input}")
])
chat_prompt = chat_template.invoke({
    "role": "security analyst",
    "input": "Analyze this log entry: failed login from 192.168.1.100"
})
print(chat_prompt.messages)

from langchain_openai import ChatOpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Create components
llm = ChatOpenAI(model="gpt-4", temperature=0)
prompt = PromptTemplate.from_template(
    "Explain {concept} like I'm 5 years old."
)
parser = StrOutputParser()

# Build chain
chain = prompt | llm | parser

# Run
result = chain.invoke({"concept": "API"})
print(result)

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

# Chain 1: Generate a product idea
prompt1 = ChatPromptTemplate.from_template(
    "Suggest a product name for a {industry} startup. Only return the name."
)
chain1 = prompt1 | llm | StrOutputParser()

# Chain 2: Write a tagline for the product
prompt2 = ChatPromptTemplate.from_template(
    "Write a one-line tagline for a product called {product_name} in the {industry} space."
)
chain2 = prompt2 | llm | StrOutputParser()

# Compose
full_chain = chain1 | (lambda name: {"product_name": name, "industry": "cybersecurity"}) | chain2
result = full_chain.invoke({"industry": "cybersecurity"})
print(f"Result: {result}")

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser

# Define specialized chains
tech_chain = PromptTemplate.from_template(
    "Answer this tech question: {input}"
) | llm | StrOutputParser()

health_chain = PromptTemplate.from_template(
    "Answer this health question: {input}"
) | llm | StrOutputParser()

finance_chain = PromptTemplate.from_template(
    "Answer this finance question: {input}"
) | llm | StrOutputParser()

# Router (simplified with conditional)
def route_query(query):
    """Route query to appropriate chain based on topic."""
    topics = {
        "tech": tech_chain,
        "health": health_chain,
        "finance": finance_chain,
    }
    
    query_lower = query.lower()
    if any(w in query_lower for w in ["code", "software", "computer", "api"]):
        return tech_chain.invoke({"input": query})
    elif any(w in query_lower for w in ["symptom", "disease", "doctor", "medicine"]):
        return health_chain.invoke({"input": query})
    elif any(w in query_lower for w in ["money", "stock", "invest", "bank"]):
        return finance_chain.invoke({"input": query})
    else:
        return "I can only answer tech, health, and finance questions."

print(route_query("What is an API?"))
print(route_query("What is a stock?"))

from langchain_core.output_parsers import JsonOutputParser, CommaSeparatedListOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

# JSON output parser
class ThreatReport(BaseModel):
    threat_name: str = Field(description="Name of the threat")
    severity: str = Field(description="Severity: low/medium/high/critical")
    affected_systems: list = Field(description="Affected systems")

parser = JsonOutputParser(pydantic_object=ThreatReport)

prompt = PromptTemplate.from_template(
    "Analyze this threat: {log_entry}\n{format_instructions}",
    partial_variables={"format_instructions": parser.get_format_instructions()}
)

chain = prompt | llm | parser
result = chain.invoke({"log_entry": "Multiple failed SSH attempts from IP 10.0.0.5 targeting root"})
print(result)

# List output parser
list_parser = CommaSeparatedListOutputParser()
list_prompt = PromptTemplate.from_template(
    "List 3 {topic} security tools.\n{format_instructions}",
    partial_variables={"format_instructions": list_parser.get_format_instructions()}
)
list_chain = list_prompt | llm | list_parser
print(list_chain.invoke({"topic": "open-source network monitoring"}))

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

# Buffer memory (stores full history)
buffer_memory = ConversationBufferMemory()
conversation = ConversationChain(llm=llm, memory=buffer_memory)

print(conversation.predict(input="Hi, I'm working on a security project."))
print(conversation.predict(input="What was my project about?"))

# Summary memory (summarizes history)
summary_memory = ConversationSummaryMemory(llm=llm)
summary_conversation = ConversationChain(llm=llm, memory=summary_memory)

print(summary_conversation.predict(input="Tell me about Python generators."))
print(summary_conversation.predict(input="Give me an example."))

from langchain_community.document_loaders import TextLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

# Load a document
# loader = TextLoader("report.txt")
# documents = loader.load()

# Simulate a document
doc_text = """LangChain is a framework for LLM applications. It provides chains, agents, and memory. 
The framework supports multiple LLM providers including OpenAI and Anthropic. 
LangChain also integrates with vector stores like Pinecone and Chroma."""

from langchain_core.documents import Document
documents = [Document(page_content=doc_text)]

# Split text into chunks
splitter = RecursiveCharacterTextSplitter(
    chunk_size=100,
    chunk_overlap=20,
    separators=["\n\n", "\n", " ", ""]
)
chunks = splitter.split_documents(documents)
print(f"Split into {len(chunks)} chunks")
for i, chunk in enumerate(chunks):
    print(f"Chunk {i+1}: {chunk.page_content[:50]}...")

from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA

# Create embeddings and vector store
# embeddings = OpenAIEmbeddings()
# vectorstore = FAISS.from_documents(documents, embeddings)

# Simulated vector store
fake_docs = [
    Document(page_content="API stands for Application Programming Interface"),
    Document(page_content="LangChain supports OpenAI and Anthropic models"),
    Document(page_content="RAG stands for Retrieval Augmented Generation"),
]

# Build QA chain
# qa_chain = RetrievalQA.from_chain_type(
#     llm=llm,
#     retriever=vectorstore.as_retriever()
# )
# result = qa_chain.invoke({"query": "What is RAG?"})
# print(result)

from langchain.agents import create_react_agent, AgentExecutor
from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Results for '{query}': [simulated web search results]"

@tool
def calculate(expression: str) -> str:
    """Calculate a mathematical expression."""
    try:
        return str(eval(expression))
    except:
        return "Error in calculation"

tools = [search_web, calculate]

from langchain_core.prompts import PromptTemplate

agent_prompt = PromptTemplate.from_template(
    "You are a helpful assistant. Use tools to answer questions.\n\n"
    "Tools: {tools}\n"
    "Tool names: {tool_names}\n"
    "Input: {input}\n"
    "Agent scratchpad: {agent_scratchpad}"
)

# agent = create_react_agent(llm, tools, agent_prompt)
# agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
# result = agent_executor.invoke({"input": "What is 25 * 4 + 10?"})
# print(result)

from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel, Field

class DocAnalysis(BaseModel):
    summary: str = Field(description="One-sentence document summary")
    key_topics: list = Field(description="3-5 key topics")
    sentiment: str = Field(description="positive/negative/neutral")
    risk_level: str = Field(description="low/medium/high")

def analyze_document(text):
    """Multi-chain document analyzer."""
    # Chain 1: Summarize
    summarize = PromptTemplate.from_template(
        "Summarize this in one sentence: {text}"
    ) | llm | StrOutputParser()
    
    # Chain 2: Extract topics
    extract_topics = PromptTemplate.from_template(
        "List 3-5 key topics from: {text}"
    ) | llm | StrOutputParser()
    
    # Chain 3: Analyze risk
    analyze = PromptTemplate.from_template(
        "Analyze the sentiment and risk level (low/medium/high) of: {text}"
    ) | llm | StrOutputParser()
    
    summary = summarize.invoke({"text": text})
    topics = extract_topics.invoke({"text": text})
    risk = analyze.invoke({"text": text})
    
    return {
        "summary": summary,
        "topics": topics,
        "risk_analysis": risk
    }

doc = "Our system detected 47 failed login attempts from 3 different IPs in the last hour."
print(analyze_document(doc))