DEV Community

Cover image for 105. LangChain: Orchestrating AI Applications
Akhilesh
Akhilesh

Posted on

105. LangChain: Orchestrating AI Applications

You have spent four posts building agents from scratch. Raw API calls. Custom tool loops. Manual memory management. Now see it in ten lines.

chain = prompt | llm | parser
Enter fullscreen mode Exit fullscreen mode

LangChain wraps the patterns you built by hand into composable, reusable components. The criticism: it abstracts too much and debugging is hard. The counter: you now know what is under the hood, so the abstractions are navigable. You built the engine. Now drive the car.


Setup

# pip install langchain langchain-core langchain-community
# pip install langchain-anthropic langchain-openai
# pip install langchain-chroma sentence-transformers faiss-cpu

import os
from langchain_anthropic import ChatAnthropic
from langchain_core.prompts import ChatPromptTemplate, PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import FAISS
from langchain_community.embeddings import HuggingFaceEmbeddings
from langchain_core.documents import Document
import warnings
warnings.filterwarnings("ignore")

llm = ChatAnthropic(
    model      = "claude-3-5-haiku-20241022",
    api_key    = os.environ.get("ANTHROPIC_API_KEY"),
    max_tokens = 500,
    temperature= 0.7,
)

parser = StrOutputParser()
response = llm.invoke("What is LangChain in one sentence?")
print(f"LLM response: {response.content}")
Enter fullscreen mode Exit fullscreen mode

LCEL: The Pipe Operator

basic_chain = (
    ChatPromptTemplate.from_template("Explain {concept} to a beginner in 2 sentences.")
    | llm
    | parser
)

result = basic_chain.invoke({"concept": "vector embeddings"})
print(f"Chain output: {result}")

from pydantic import BaseModel, Field
from langchain_core.output_parsers import PydanticOutputParser
from typing import List

class CodeReview(BaseModel):
    has_bugs:    bool       = Field(description="Whether code has bugs")
    quality:     int        = Field(description="Quality score 1-10")
    issues:      List[str]  = Field(description="Specific issues found")
    suggestions: List[str]  = Field(description="Improvement suggestions")

pydantic_parser = PydanticOutputParser(pydantic_object=CodeReview)

review_chain = (
    ChatPromptTemplate.from_messages([
        ("system", f"Review Python code. {pydantic_parser.get_format_instructions()}"),
        ("human",  "Review:\n```

python\n{code}\n

```")
    ])
    | llm
    | pydantic_parser
)

buggy = "def divide(a, b):\n    return a / b\n\nprint(divide(10, 0))"
review = review_chain.invoke({"code": buggy})
print(f"\nCode Review:")
print(f"  Has bugs: {review.has_bugs}")
print(f"  Quality:  {review.quality}/10")
print(f"  Issues:   {review.issues}")
Enter fullscreen mode Exit fullscreen mode

Memory and History

from langchain_core.chat_history import InMemoryChatMessageHistory
from langchain_core.runnables.history import RunnableWithMessageHistory

store = {}

def get_session_history(session_id: str) -> InMemoryChatMessageHistory:
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

chat_prompt = ChatPromptTemplate.from_messages([
    ("system",      "You are a helpful tutor."),
    ("placeholder", "{chat_history}"),
    ("human",       "{input}"),
])

chain_with_history = RunnableWithMessageHistory(
    chat_prompt | llm | parser,
    get_session_history,
    input_messages_key   = "input",
    history_messages_key = "chat_history",
)

config = {"configurable": {"session_id": "student_001"}}

turns = [
    "My name is Priya. I am learning about neural networks.",
    "Can you explain what backpropagation does?",
    "What is my name and what am I studying?",
]

print("\nMulti-turn conversation with memory:")
for turn in turns:
    response = chain_with_history.invoke({"input": turn}, config=config)
    print(f"  User:  {turn}")
    print(f"  Agent: {response[:100]}...")
    print()

print(f"History: {len(store['student_001'].messages)} messages stored")
Enter fullscreen mode Exit fullscreen mode

RAG with LangChain

documents = [
    Document(page_content="The transformer uses self-attention mechanisms.",
             metadata={"source": "transformer_paper"}),
    Document(page_content="BERT is pretrained using masked language modeling.",
             metadata={"source": "bert_paper"}),
    Document(page_content="GPT uses autoregressive next-token prediction.",
             metadata={"source": "gpt_paper"}),
    Document(page_content="LoRA adds low-rank matrices to frozen pretrained weights.",
             metadata={"source": "lora_paper"}),
    Document(page_content="Fine-tuning adapts pretrained models to downstream tasks.",
             metadata={"source": "fine_tuning_guide"}),
]

from langchain.text_splitter import RecursiveCharacterTextSplitter

splitter    = RecursiveCharacterTextSplitter(chunk_size=200, chunk_overlap=30)
split_docs  = splitter.split_documents(documents)

embeddings  = HuggingFaceEmbeddings(model_name="sentence-transformers/all-MiniLM-L6-v2")
vectorstore = FAISS.from_documents(split_docs, embeddings)
retriever   = vectorstore.as_retriever(search_kwargs={"k": 3})

def format_docs(docs):
    return "\n\n".join([
        f"[{i+1}] Source: {d.metadata['source']}\n{d.page_content}"
        for i, d in enumerate(docs)
    ])

rag_prompt = ChatPromptTemplate.from_template("""
Answer ONLY from the context below. Cite sources as [1], [2].
If not in context: "I cannot find this in the documents."

Context:
{context}

Question: {question}
""")

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt
    | llm
    | parser
)

queries = [
    "How does BERT get pretrained?",
    "What is LoRA?",
    "Who is the president of France?",
]

print("\nRAG Chain Demo:")
for query in queries:
    answer = rag_chain.invoke(query)
    print(f"  Q: {query}")
    print(f"  A: {answer[:120]}...")
    print()
Enter fullscreen mode Exit fullscreen mode

Agents in LangChain

from langchain_core.tools import tool
from langchain.agents import create_tool_calling_agent, AgentExecutor
import math

@tool
def calculator(expression: str) -> str:
    """Evaluate a math expression. Supports +,-,*,/,sqrt,pi,e."""
    try:
        result = eval(expression, {"__builtins__": {}},
                      {"sqrt": math.sqrt, "pi": math.pi, "e": math.e})
        return str(round(float(result), 6))
    except Exception as e:
        return f"Error: {e}"

@tool
def word_count(text: str) -> str:
    """Count words and characters in text."""
    return f"Words: {len(text.split())}, Characters: {len(text)}"

@tool
def reverse_string(text: str) -> str:
    """Reverse a string."""
    return text[::-1]

agent_prompt = ChatPromptTemplate.from_messages([
    ("system",      "You are a helpful assistant. Use tools when needed."),
    ("placeholder", "{chat_history}"),
    ("human",       "{input}"),
    ("placeholder", "{agent_scratchpad}"),
])

agent    = create_tool_calling_agent(llm, [calculator, word_count, reverse_string], agent_prompt)
executor = AgentExecutor(agent=agent, tools=[calculator, word_count, reverse_string],
                          verbose=False, max_iterations=5)

test_queries = [
    "What is the square root of 1764?",
    "Count the words in 'To be or not to be that is the question'",
    "Reverse the string 'LangChain'",
]

print("\nAgent with Tools:")
for query in test_queries:
    result = executor.invoke({"input": query, "chat_history": []})
    print(f"  Q: {query}")
    print(f"  A: {result['output']}")
    print()
Enter fullscreen mode Exit fullscreen mode

When to Use LangChain vs From Scratch

comparison = {
    "Simple single LLM call":          ("From scratch", "Overkill to add LangChain"),
    "RAG with multiple retrievers":     ("LangChain",    "Pre-built components save time"),
    "Multi-provider LLM switching":     ("LangChain",    "Unified interface is valuable"),
    "Custom agent with full control":   ("From scratch", "Full visibility, easier debug"),
    "Rapid prototyping":                ("LangChain",    "Much faster to build"),
    "Production with LangSmith tracing":("LangChain",    "Observability built in"),
}

print("LangChain vs From Scratch:")
print(f"{'Scenario':<42} {'Recommendation':<15} {'Reason'}")
print("=" * 85)
for scenario, (rec, reason) in comparison.items():
    print(f"{scenario:<42} {rec:<15} {reason}")
Enter fullscreen mode Exit fullscreen mode

LangSmith: Debugging LangChain

print("\nLangSmith: Essential for Production LangChain")
print()
print("Setup (3 lines):")
print("  export LANGSMITH_API_KEY='your_key'")
print("  export LANGSMITH_TRACING='true'")
print("  export LANGCHAIN_PROJECT='my_project'")
print()
print("That is it. Every run is traced automatically.")
print()
print("What you see in LangSmith:")
features = [
    "Full trace of every LLM call with tokens and cost",
    "Timing breakdown per chain step",
    "Side-by-side diff of prompt changes",
    "User feedback collection",
    "Dataset creation from production traces",
    "A/B testing of different prompts",
]
for f in features:
    print(f"{f}")
print()
print("Without LangSmith, debugging LangChain in production is painful.")
print("It is not optional for serious applications.")
Enter fullscreen mode Exit fullscreen mode

Reference Links

print("\nLangChain Reference Links:")
print()

refs = {
    "Official Docs": [
        ("LangChain Python docs",           "python.langchain.com"),
        ("LCEL reference",                  "python.langchain.com/docs/expression_language"),
        ("LangGraph (stateful agents)",     "langchain-ai.github.io/langgraph"),
        ("LangSmith docs",                  "docs.smith.langchain.com"),
        ("All integrations",                "python.langchain.com/docs/integrations"),
    ],
    "Cheat Sheets": [
        ("LCEL cheatsheet",                 "python.langchain.com/docs/expression_language/interface"),
        ("Output parsers guide",            "python.langchain.com/docs/modules/model_io/output_parsers"),
        ("v0.3 migration guide",            "python.langchain.com/docs/versions/migrating_chains"),
        ("Memory types reference",          "python.langchain.com/docs/modules/memory"),
    ],
    "Tutorials": [
        ("DeepLearning.AI LangChain course", "learn.deeplearning.ai/langchain"),
        ("LangChain RAG tutorial",           "python.langchain.com/docs/use_cases/question_answering"),
        ("Build a chatbot tutorial",         "python.langchain.com/docs/use_cases/chatbots"),
        ("LangChain cookbook",               "github.com/langchain-ai/langchain/tree/master/cookbook"),
        ("LangGraph agent tutorial",         "langchain-ai.github.io/langgraph/tutorials/introduction"),
    ],
    "Community": [
        ("LangChain GitHub (star it)",      "github.com/langchain-ai/langchain"),
        ("Discord",                         "discord.gg/langchain"),
        ("Awesome LangChain resources",     "github.com/kyrolabs/awesome-langchain"),
    ],
}

for category, links in refs.items():
    print(f"  {category}:")
    for name, url in links:
        print(f"{name:<45} {url}")
    print()
Enter fullscreen mode Exit fullscreen mode

Try This

Create langchain_practice.py.

Part 1: build three LCEL chains using the pipe operator: a translation chain, a code review chain that outputs structured JSON, and a topic classification chain.

Part 2: build a RAG chain with FAISS and all-MiniLM-L6-v2. Load 20 documents. Test 10 queries. Compare retrieval quality versus the manual cosine similarity approach from post 85.

Part 3: build a LangChain agent with four tools. Give it three multi-step tasks. Enable verbose=True and compare the reasoning trace to the raw agent you built in post 101.

Part 4: add conversation history with RunnableWithMessageHistory. Build a 5-turn conversation. Inspect the stored history. Start a new session and confirm it has no memory of the previous one.


What's Next

LangChain provides the scaffolding. LangGraph adds state machines: model agent behavior as a directed graph where nodes are actions and edges are conditional transitions. Complex agents become debuggable, resumable, and production-ready. Next post.

Top comments (0)