Mastering LangChain: A Comprehensive Guide.

from openai import OpenAI
 
# Using DeepSeek API (OpenAI-compatible format)
client = OpenAI(
    api_key="your-deepseek-api-key",
    base_url="https://api.deepseek.com"
)
 
response = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "What is an Agent?"}
    ]
)
print(response.choices[0].message.content)

from langchain.agents import create_agent
 
def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Search results: information about {query}..."
 
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))
 
agent = create_agent(
    model="openai:deepseek-v4-flash",
    tools=[search_web, calculate],
    system_prompt="You are a helpful assistant that can search and calculate."
)
 
result = agent.invoke({
    "messages": [{"role": "user", "content": "Calculate 3^10 for me"}]
})

def get_weather(city: str) -> str:
    """Get weather information for a given city."""
    # In production, this would call a real weather API
    return f"It's sunny in {city}, 22°C"
 
# Pass directly to create_agent
agent = create_agent(
    model="openai:deepseek-v4-flash",
    tools=[get_weather],
    system_prompt="You are a weather assistant."
)

from typing import Annotated
from typing_extensions import TypedDict
from operator import add
 
class AgentState(TypedDict):
    messages: Annotated[list, add]  # Message list; new messages are appended
    current_step: str               # Current step

from langgraph.graph import StateGraph, MessagesState, START, END
 
def chatbot(state: MessagesState):
    """A node that processes user messages."""
    return {"messages": [{"role": "ai", "content": "Hello! I'm your assistant."}]}
 
graph = StateGraph(MessagesState)
graph.add_node(chatbot)
graph.add_edge(START, "chatbot")
graph.add_edge("chatbot", END)
graph = graph.compile()
 
result = graph.invoke({"messages": [{"role": "user", "content": "Hello!"}]})

# pip install -qU langchain "langchain[openai]"
from langchain.agents import create_agent
 
# Using a DeepSeek model
agent = create_agent(
    model="openai:deepseek-v4-flash",
    system_prompt="You are a helpful assistant."
)

def lookup_database(query: str) -> str:
    """Query the database for information.
 
    Args:
        query: SQL query string
    """
    # In production, connect to a real database
    return f"Query results: 3 matching records found"
 
def send_email(to: str, subject: str, body: str) -> str:
    """Send an email.
 
    Args:
        to: Recipient email address
        subject: Email subject line
        body: Email body content
    """
    return f"Email sent to {to}"

from langchain.agents import create_agent
 
agent = create_agent(
    model="openai:deepseek-v4-flash",
    tools=[get_weather],
    system_prompt="You are an assistant.",
    # Add prebuilt or custom middleware here
)

from langgraph.graph import StateGraph, MessagesState, START, END
 
# 1. Define node functions
def greet(state: MessagesState):
    """A greeting node."""
    return {"messages": [{"role": "ai", "content": "Hello! How can I help you?"}]}
 
# 2. Build the graph
graph = StateGraph(MessagesState)
graph.add_node(greet)
graph.add_edge(START, "greet")       # From START to greet node
graph.add_edge("greet", END)         # From greet to END
 
# 3. Compile and execute
app = graph.compile()
result = app.invoke({"messages": [{"role": "user", "content": "Hello"}]})

from openai import OpenAI
 
client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.deepseek.com"
)
 
# Basic chat
response = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain quantum computing"}
    ],
    stream=False
)
print(response.choices[0].message.content)
 
# Streaming output
stream = client.chat.completions.create(
    model="deepseek-v4-flash",
    messages=[{"role": "user", "content": "Write a poem about spring"}],
    stream=True
)
for chunk in stream:
    if chunk.choices[0].delta.content:
        print(chunk.choices[0].delta.content, end="")

from langchain.agents import create_agent
 
def search_knowledge_base(query: str) -> str:
    """Search the knowledge base for relevant documents.
 
    Args:
        query: Search keywords
    """
    # In production, connect to a vector database
    return f"Found documents related to '{query}': LangGraph is a..."
 
def create_summary(text: str, max_words: int = 100) -> str:
    """Generate a summary for the given text.
 
    Args:
        text: Original text
        max_words: Maximum word count for the summary
    """
    return f"Summary ({max_words} words max): {text[:50]}..."
 
agent = create_agent(
    model="openai:deepseek-v4-flash",
    tools=[search_knowledge_base, create_summary],
    system_prompt="You are a research assistant. Search the knowledge base first, then summarize."
)
 
result = agent.invoke({
    "messages": [{"role": "user", "content": "Look up LangGraph's core features and summarize them"}]
})

from langchain.agents import create_agent
 
def retrieve_documents(query: str) -> str:
    """Retrieve relevant documents from a vector database.
 
    Args:
        query: User query
    """
    # In production, use LangChain's retriever
    return f"Retrieved 3 relevant documents: [Doc1...], [Doc2...], [Doc3...]"
 
rag_agent = create_agent(
    model="openai:deepseek-v4-flash",
    tools=[retrieve_documents],
    system_prompt="""You are a RAG assistant.
    1. First, use the retrieve_documents tool to fetch relevant documents
    2. Answer the user's question based on the retrieved content
    3. If no relevant information is found, honestly tell the user"""
)

from typing import Literal
from langgraph.graph import StateGraph, MessagesState, START, END
 
def agent_node(state: MessagesState):
    """Agent thinking node."""
    return {"messages": [{"role": "ai", "content": "I need to use a tool"}]}
 
def tool_node(state: MessagesState):
    """Tool execution node."""
    return {"messages": [{"role": "tool", "content": "Tool execution result"}]}
 
def should_continue(state: MessagesState) -> Literal["tools", "__end__"]:
    """Routing function: decide whether to continue using tools or end."""
    last_message = state["messages"][-1]
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "__end__"
 
# Build an Agent graph with a loop
builder = StateGraph(MessagesState)
builder.add_node("agent", agent_node)
builder.add_node("tools", tool_node)
 
builder.add_edge(START, "agent")
builder.add_conditional_edges("agent", should_continue)
builder.add_edge("tools", "agent")  # After tool execution, return to agent — forming a loop
 
app = builder.compile()

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph, MessagesState, START, END
from langgraph.types import Command, interrupt
 
def human_review(state: MessagesState):
    """Pause execution and wait for human approval."""
    answer = interrupt("Please review the agent's action. Continue?")
    return {"messages": [{"role": "user", "content": answer}]}
 
builder = StateGraph(MessagesState)
builder.add_node("review", human_review)
builder.add_edge(START, "review")
builder.add_edge("review", END)
 
checkpointer = InMemorySaver()
app = builder.compile(checkpointer=checkpointer)
 
config = {"configurable": {"thread_id": "review-1"}}
 
# First invocation — pauses at the interrupt
stream = app.invoke({"messages": []}, config)
 
# After human review, resume execution
result = app.invoke(Command(resume="Continue"), config)

from openai import OpenAI
 
client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.deepseek.com"
)
 
# Enable thinking mode
response = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[
        {"role": "user", "content": "Prove that the square root of 2 is irrational"}
    ],
    reasoning_effort="high",  # Control reasoning effort: high or max
    extra_body={"thinking": {"type": "enabled"}},
)
 
# Access the reasoning process and final answer
reasoning = response.choices[0].message.reasoning_content  # CoT reasoning trace
answer = response.choices[0].message.content                # Final answer
 
print("=== Reasoning Process ===")
print(reasoning)
print("=== Final Answer ===")
print(answer)

from langgraph.graph import StateGraph, MessagesState, START, END
 
def research_agent(state: MessagesState):
    """Research agent: responsible for information gathering."""
    return {"messages": [{"role": "ai", "content": "Research complete. Here's the gathered info..."}]}
 
def writer_agent(state: MessagesState):
    """Writer agent: responsible for content generation."""
    return {"messages": [{"role": "ai", "content": "Based on the research, generating the article..."}]}
 
def reviewer_agent(state: MessagesState):
    """Reviewer agent: responsible for quality checks."""
    return {"messages": [{"role": "ai", "content": "Review passed."}]}
 
# Build the collaboration pipeline
builder = StateGraph(MessagesState)
builder.add_node("researcher", research_agent)
builder.add_node("writer", writer_agent)
builder.add_node("reviewer", reviewer_agent)
 
builder.add_edge(START, "researcher")
builder.add_edge("researcher", "writer")
builder.add_edge("writer", "reviewer")
builder.add_edge("reviewer", END)
 
app = builder.compile()

from langgraph.checkpoint.memory import InMemorySaver
 
# Use an in-memory Checkpointer (for development)
checkpointer = InMemorySaver()
 
# For production, use PostgreSQL, MongoDB, or other persistent storage
app = builder.compile(checkpointer=checkpointer)
 
# Each invoke uses a different thread_id to isolate state
config = {"configurable": {"thread_id": "user-123"}}
result = app.invoke({"messages": [...]}, config)

from typing import Literal
from langgraph.graph import StateGraph, START, END
from typing_extensions import TypedDict
 
class PlanState(TypedDict):
    goal: str
    plan: list[str]
    current_step: int
    results: list[str]
 
def planner(state: PlanState):
    """Create an execution plan."""
    plan = ["Step 1: Gather information", "Step 2: Analyze data", "Step 3: Generate report"]
    return {"plan": plan, "current_step": 0, "results": []}
 
def executor(state: PlanState):
    """Execute the current step."""
    step = state["plan"][state["current_step"]]
    return {
        "results": state["results"] + [f"Completed: {step}"],
        "current_step": state["current_step"] + 1
    }
 
def should_continue(state: PlanState) -> Literal["executor", "__end__"]:
    """Are there more steps to execute?"""
    if state["current_step"] < len(state["plan"]):
        return "executor"
    return "__end__"
 
builder = StateGraph(PlanState)
builder.add_node("planner", planner)
builder.add_node("executor", executor)
builder.add_edge(START, "planner")
builder.add_edge("planner", "executor")
builder.add_conditional_edges("executor", should_continue)
 
app = builder.compile()

# Enable LangSmith tracing (just set environment variables)
import os
os.environ["LANGSMITH_TRACING"] = "true"
os.environ["LANGSMITH_API_KEY"] = "your-langsmith-key"
 
# All agent executions are now automatically logged to LangSmith

from langchain.agents import create_agent
 
def get_time() -> str:
    """Get the current time."""
    from datetime import datetime
    return datetime.now().strftime("%Y-%m-%d %H:%M:%S")
 
agent = create_agent(
    model="openai:deepseek-v4-flash",
    tools=[get_time],
    system_prompt="You are a time assistant."
)
 
result = agent.invoke({
    "messages": [{"role": "user", "content": "What time is it now?"}]
})

from typing import Literal
from langgraph.graph import StateGraph, MessagesState, START, END
 
def classifier(state: MessagesState):
    """Classify user intent."""
    return {"messages": [{"role": "ai", "content": "Classification result: weather query"}]}
 
def weather_handler(state: MessagesState):
    """Handle weather queries."""
    return {"messages": [{"role": "ai", "content": "Sunny today, 25°C"}]}
 
def general_handler(state: MessagesState):
    """Handle general questions."""
    return {"messages": [{"role": "ai", "content": "I'm your assistant. How can I help?"}]}
 
def route(state: MessagesState) -> Literal["weather", "general"]:
    last = state["messages"][-1].content if hasattr(state["messages"][-1], "content") else ""
    if "weather" in str(last).lower():
        return "weather"
    return "general"
 
builder = StateGraph(MessagesState)
builder.add_node("classifier", classifier)
builder.add_node("weather", weather_handler)
builder.add_node("general", general_handler)
 
builder.add_edge(START, "classifier")
builder.add_conditional_edges("classifier", route, {
    "weather": "weather",
    "general": "general"
})
builder.add_edge("weather", END)
builder.add_edge("general", END)
 
app = builder.compile()

from openai import OpenAI
import json
 
client = OpenAI(
    api_key="your-api-key",
    base_url="https://api.deepseek.com"
)
 
tools = [
    {
        "type": "function",
        "function": {
            "name": "get_population",
            "description": "Get population data for a city",
            "parameters": {
                "type": "object",
                "properties": {
                    "city": {"type": "string", "description": "City name"}
                },
                "required": ["city"]
            }
        }
    }
]
 
def get_population(city: str) -> str:
    populations = {"New York": "8.3M", "London": "9.0M", "Tokyo": "14.0M"}
    return populations.get(city, "Data not available")
 
# Tool calling with thinking mode
response = client.chat.completions.create(
    model="deepseek-v4-pro",
    messages=[{"role": "user", "content": "Which city has more people: New York or London?"}],
    tools=tools,
    reasoning_effort="high",
    extra_body={"thinking": {"type": "enabled"}},
)
 
# Handle tool calls
if response.choices[0].message.tool_calls:
    for tool_call in response.choices[0].message.tool_calls:
        args = json.loads(tool_call.function.arguments)
        result = get_population(**args)
        print(f"Tool call {tool_call.function.name}({args}) = {result}")