LangChain and LangGraph

Overview

LangChain is the earliest and most mature LLM application development framework, while LangGraph is its extension for stateful, multi-step workflows. Together they form the most complete agent development ecosystem available today.

LangChain Core Concepts

Architecture Layers

graph TD
    subgraph LangChain Ecosystem
        A[LangChain Core] --> B[LangChain Community]
        A --> C[LangGraph]
        A --> D[LangServe]
        A --> E[LangSmith]
    end

    A --> A1[Models]
    A --> A2[Prompts]
    A --> A3[Output Parsers]
    A --> A4[Chains / LCEL]
    A --> A5[Tools]
    A --> A6[Retrievers]

Model Invocation

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# Unified model interface
openai_llm = ChatOpenAI(model="gpt-4o", temperature=0)
claude_llm = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0)

# Switching models requires only one line change
response = openai_llm.invoke("What is AI?")

Prompt Templates

from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# System + user template
prompt = ChatPromptTemplate.from_messages([
    ("system", "You are a helpful assistant specialized in {domain}."),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}"),
])

# Usage
chain = prompt | llm
response = chain.invoke({
    "domain": "machine learning",
    "history": [],
    "input": "Explain gradient descent"
})

LCEL (LangChain Expression Language)

LCEL is LangChain's declarative orchestration syntax, using the pipe operator | to compose components:

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnablePassthrough, RunnableLambda

# Simple chain
chain = prompt | llm | StrOutputParser()

# RAG chain with retrieval
rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough()
    }
    | prompt
    | llm
    | StrOutputParser()
)

# Parallel execution
from langchain_core.runnables import RunnableParallel

parallel_chain = RunnableParallel(
    summary=summary_chain,
    translation=translation_chain,
    keywords=keyword_chain,
)

Output Parsers

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field

class AnalysisResult(BaseModel):
    sentiment: str = Field(description="positive, negative, or neutral")
    confidence: float = Field(description="confidence score 0-1")
    key_topics: list[str] = Field(description="main topics discussed")

parser = JsonOutputParser(pydantic_object=AnalysisResult)

chain = prompt | llm | parser
result = chain.invoke({"text": "I love this product!"})
# result = AnalysisResult(sentiment="positive", confidence=0.95, ...)

Tool Definition

from langchain_core.tools import tool

@tool
def search_web(query: str) -> str:
    """Search the web for current information."""
    # Implement search logic
    return search_engine.search(query)

@tool
def calculate(expression: str) -> float:
    """Evaluate a mathematical expression."""
    return eval(expression)  # Use a safe math parser in production

# Bind tools to the model
llm_with_tools = llm.bind_tools([search_web, calculate])

LangGraph: Stateful Workflows

LangGraph models agent workflows as directed graphs, providing precise process control and state management.

Core Concepts

Concept	Description	Analogy
State	Global state of the workflow	Shared memory
Node	Processing node (function)	A step in the process
Edge	Connection between nodes	Transition condition
Conditional Edge	Conditional branch	if-else
Checkpoint	State snapshot	Save point

State Definition

from typing import TypedDict, Annotated, Sequence
from langchain_core.messages import BaseMessage
import operator

class AgentState(TypedDict):
    """Agent workflow state definition"""
    messages: Annotated[Sequence[BaseMessage], operator.add]
    current_step: str
    iteration_count: int
    final_answer: str | None

Building a Graph

from langgraph.graph import StateGraph, END

def call_model(state: AgentState) -> AgentState:
    """Call LLM"""
    messages = state["messages"]
    response = llm_with_tools.invoke(messages)
    return {"messages": [response]}

def call_tools(state: AgentState) -> AgentState:
    """Execute tool calls"""
    last_message = state["messages"][-1]
    tool_results = []
    for tool_call in last_message.tool_calls:
        result = tool_map[tool_call["name"]].invoke(tool_call["args"])
        tool_results.append(ToolMessage(
            content=str(result),
            tool_call_id=tool_call["id"]
        ))
    return {"messages": tool_results}

def should_continue(state: AgentState) -> str:
    """Decide whether to continue"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return "end"

# Build graph
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)

# Add edges
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {"tools": "tools", "end": END}
)
workflow.add_edge("tools", "agent")

# Compile
app = workflow.compile()

Visualization

graph TD
    Start((Start)) --> Agent[Agent Node<br/>Call LLM]
    Agent -->|Has tool calls| Tools[Tools Node<br/>Execute Tools]
    Agent -->|No tool calls| End((End))
    Tools --> Agent

Checkpointing (State Persistence)

from langgraph.checkpoint.sqlite import SqliteSaver

# Use SQLite for state persistence
memory = SqliteSaver.from_conn_string(":memory:")
app = workflow.compile(checkpointer=memory)

# State is automatically saved on each invocation
config = {"configurable": {"thread_id": "user_123"}}
result = app.invoke(
    {"messages": [HumanMessage(content="Hello")]},
    config=config
)

# Can restore to any historical state
states = list(app.get_state_history(config))

Human-in-the-Loop

from langgraph.graph import StateGraph, END

def human_review(state: AgentState) -> AgentState:
    """Human review node"""
    # In practice, this would pause and wait for human input
    pass

workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)
workflow.add_node("human_review", human_review)

# High-risk operations require human review
workflow.add_conditional_edges(
    "agent",
    route_by_risk,
    {
        "low_risk": "tools",
        "high_risk": "human_review",
        "done": END
    }
)

# Compile with interrupts enabled
app = workflow.compile(
    checkpointer=memory,
    interrupt_before=["human_review"]  # Interrupt before human_review
)

Subgraphs

# Define subgraph: research workflow
research_workflow = StateGraph(ResearchState)
research_workflow.add_node("search", search_node)
research_workflow.add_node("analyze", analyze_node)
research_workflow.add_node("summarize", summarize_node)
# ... configure edges

# Use subgraph in main graph
main_workflow = StateGraph(MainState)
main_workflow.add_node("research", research_workflow.compile())
main_workflow.add_node("write", write_node)
main_workflow.add_node("review", review_node)

Practical Example: Research Assistant

from langgraph.graph import StateGraph, END
from langchain_core.messages import HumanMessage, AIMessage

class ResearchState(TypedDict):
    messages: Annotated[list, operator.add]
    research_topic: str
    search_results: list[str]
    draft: str
    feedback: str
    revision_count: int

def plan_research(state):
    """Plan research directions"""
    prompt = f"Plan a research outline for: {state['research_topic']}"
    response = llm.invoke(prompt)
    return {"messages": [AIMessage(content=response.content)]}

def search_sources(state):
    """Search for relevant sources"""
    results = web_search.invoke(state["research_topic"])
    return {"search_results": results}

def write_draft(state):
    """Write initial draft"""
    context = "\n".join(state["search_results"])
    prompt = f"Write a report on {state['research_topic']}.\nSources:\n{context}"
    draft = llm.invoke(prompt)
    return {"draft": draft.content}

def review_draft(state):
    """Review the draft"""
    prompt = f"Review this draft and provide feedback:\n{state['draft']}"
    feedback = llm.invoke(prompt)
    return {"feedback": feedback.content, "revision_count": state["revision_count"] + 1}

def should_revise(state):
    """Whether revision is needed"""
    if state["revision_count"] >= 3:
        return "finalize"
    if "APPROVED" in state["feedback"]:
        return "finalize"
    return "revise"

# Build graph
graph = StateGraph(ResearchState)
graph.add_node("plan", plan_research)
graph.add_node("search", search_sources)
graph.add_node("write", write_draft)
graph.add_node("review", review_draft)

graph.set_entry_point("plan")
graph.add_edge("plan", "search")
graph.add_edge("search", "write")
graph.add_edge("write", "review")
graph.add_conditional_edges("review", should_revise, {
    "revise": "write",
    "finalize": END
})

app = graph.compile()

LangSmith: Observability

LangSmith provides tracing and monitoring for LangChain applications:

• Trace visualization: View inputs, outputs, and latency at each step
• Evaluation: Automated testing and evaluation
• Dataset management: Build and manage test datasets
• Feedback collection: Collect user feedback

Best Practices

1. Model Selection

# Use different models for different tasks
cheap_llm = ChatOpenAI(model="gpt-4o-mini")    # Simple tasks
powerful_llm = ChatOpenAI(model="gpt-4o")       # Complex reasoning
fast_llm = ChatOpenAI(model="gpt-4o-mini")      # Low-latency scenarios

2. Error Handling

from langchain_core.runnables import RunnableConfig
from tenacity import retry, stop_after_attempt

@retry(stop=stop_after_attempt(3))
def robust_chain_invoke(chain, input_data):
    try:
        return chain.invoke(input_data)
    except Exception as e:
        logger.error(f"Chain failed: {e}")
        raise

3. Streaming Output

# Streaming output
async for chunk in app.astream(
    {"messages": [HumanMessage(content="Tell me about AI")]},
    config=config
):
    for node_name, output in chunk.items():
        print(f"[{node_name}]: {output}")

Summary

Choice	Scenario
LangChain Core	Simple LLM applications, RAG pipelines
LangGraph	Stateful workflows, complex agents, human-in-the-loop
LangSmith	Production environments needing observability and evaluation
Raw API	Minimal scenarios or when maximum control is needed

The greatest advantage of the LangChain ecosystem is completeness---from prototype to production, from single agent to multi-agent, from development to monitoring, it provides a full-stack solution.

---

评论