LangGraph & LangChain: Building Agentic AI

# Classic LangChain chain
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

prompt = PromptTemplate(template="Summarize: {text}")
chain = LLMChain(llm=model, prompt=prompt)
result = chain.run("Your text here")

# LangGraph workflow with memory
from langgraph.graph import StateGraph
from langgraph.checkpoint.memory import MemorySaver

workflow = StateGraph(YourState)
workflow.add_node("step1", your_function)
workflow.add_edge("step1", "step2")

memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI

def get_weather(city: str) -> str:
    """Get weather for a given city."""
    return f"It's sunny in {city}!"

def create_ticket(issue: str) -> str:
    """Create support ticket."""
    return f"Ticket created: {issue}"

agent = create_react_agent(
    model=ChatOpenAI(model="gpt-4"),
    tools=[get_weather, create_ticket],
    prompt="You are a helpful assistant"
)

# One agent, multiple capabilities!
response = agent.invoke({
    "messages": [{"role": "user", "content": "Weather in NYC and create ticket for broken printer"}]
})

from langgraph.graph import StateGraph, START, END
from typing import TypedDict

class ContentState(TypedDict):
    topic: str
    research: str
    draft: str
    final: str

def research_agent(state: ContentState):
    return {"research": f"Research data for {state['topic']}"}

def writer_agent(state: ContentState):
    return {"draft": f"Draft based on: {state['research']}"}

def editor_agent(state: ContentState):
    return {"final": f"Polished version of: {state['draft']}"}

# Build the assembly line
workflow = StateGraph(ContentState)
workflow.add_node("research", research_agent)
workflow.add_node("write", writer_agent)
workflow.add_node("edit", editor_agent)

# Define the flow
workflow.add_edge(START, "research")
workflow.add_edge("research", "write")
workflow.add_edge("write", "edit")
workflow.add_edge("edit", END)

app = workflow.compile()

from langgraph.constants import Send

class DocState(TypedDict):
    documents: list
    results: list

def route_docs(state: DocState):
    # Fan out to parallel workers
    return [Send("process", {"doc": doc}) for doc in state["documents"]]

def process_doc(state: dict):
    doc = state["doc"]
    processed = f"Processed: {doc['name']}"
    return {"results": [processed]}

def combine_results(state: DocState):
    summary = f"Processed {len(state['results'])} documents"
    return {"summary": summary}

workflow = StateGraph(DocState)
workflow.add_node("route", route_docs)
workflow.add_node("process", process_doc)
workflow.add_node("combine", combine_results)

workflow.add_conditional_edges("route", lambda x: "process")
workflow.add_edge("process", "combine")

from langgraph.checkpoint.memory import MemorySaver

class ApprovalState(TypedDict):
    request: str
    analysis: str
    approved: bool
    response: str

def analyze_request(state: ApprovalState):
    return {"analysis": f"AI analysis of: {state['request']}"}

def need_approval(state: ApprovalState):
    return "wait_approval" if not state.get("approved") else "finalize"

def wait_for_human(state: ApprovalState):
    # 🛑 Workflow pauses here for human input
    return {"response": "Awaiting human approval..."}

def finalize_response(state: ApprovalState):
    return {"response": f"Approved response: {state['analysis']}"}

workflow = StateGraph(ApprovalState)
workflow.add_node("analyze", analyze_request)
workflow.add_node("wait_approval", wait_for_human)
workflow.add_node("finalize", finalize_response)

workflow.add_conditional_edges("analyze", need_approval)
workflow.add_edge("wait_approval", "finalize")

# 💾 Persistent memory for long-running workflows
memory = MemorySaver()
app = workflow.compile(checkpointer=memory)

from langchain.tools.retriever import create_retriever_tool
from langgraph.prebuilt import tools_condition, ToolNode

# Create smart retriever
retriever_tool = create_retriever_tool(
    retriever=your_vectorstore.as_retriever(),
    name="search_docs",
    description="Search company knowledge base"
)

def decide_action(state):
    """🤔 LLM decides whether to search or respond directly"""
    last_message = state["messages"][-1]["content"]

    # Bind tools to model
    model_with_tools = model.bind_tools([retriever_tool])
    response = model_with_tools.invoke(state["messages"])
    return {"messages": [response]}

workflow = StateGraph(MessagesState)
workflow.add_node("agent", decide_action)
workflow.add_node("tools", ToolNode([retriever_tool]))

workflow.add_conditional_edges("agent", tools_condition)
workflow.add_edge("tools", "agent")
workflow.add_edge(START, "agent")

rag_agent = workflow.compile()

# 🏭 Production configuration
from langgraph.checkpoint.postgres import PostgresSaver

# Use database for persistence (not memory!)
checkpointer = PostgresSaver.from_conn_string(
    "postgresql://user:pass@host/db"
)

production_app = workflow.compile(
    checkpointer=checkpointer,
    debug=False  # Turn off debug logs
)

# 📊 Deploy with monitoring
from langsmith import traceable

@traceable
def monitored_call(input_data):
    return production_app.invoke(input_data)