LangChain Agent Fundamentals

import os
from dotenv import load_dotenv
load_dotenv()

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain.agents import create_agent
from langchain.messages import HumanMessage

model = ChatGoogleGenerativeAI(model='gemini-2.5-flash')

system_prompt = "You are a helpful assistant that provides concise and accurate responses."

agent = create_agent(model=model, system_prompt=system_prompt)

tech_prompt = """You are a technical expert.
- Provide detailed technical responses
- Use precise terminology
- Include code examples when relevant
"""

tech_agent = create_agent(model=model, system_prompt=tech_prompt)
response = tech_agent.invoke({'messages': [HumanMessage('I can not login into my account')]})
print(response['messages'][-1].text)

First identify the error state: 401 Unauthorized (bad credentials/expired token),
403 Forbidden (account locked), 429 Too Many Requests (rate limiting), or 500/503
(auth service down). For web logins, clear stale cookies/LocalStorage or use an
incognito window, and verify NTP clock sync for TOTP/JWT validation...

from langchain.tools import tool
from langchain.agents.middleware import wrap_tool_call
from langchain.messages import ToolMessage

@tool
def divide(a: float, b: float):
    """Divide the two numbers"""
    return a / b

@wrap_tool_call
def handle_tool_errors(request, handler):
    try:
        return handler(request)
    except Exception as e:
        return ToolMessage(
            content=f"Error: {str(e)}. Try different Input.",
            tool_call_id=request.tool_call['id']
        )

agent = create_agent(model=model,
                     tools=[base_tools.web_search, base_tools.get_weather, divide],
                     system_prompt='You are a helpful AI assistant.',
                     middleware=[handle_tool_errors])

agent.invoke({'messages': [HumanMessage('what is the current weather in Mumbai and what is 1/0?')]})

from langchain.tools import ToolRuntime
from dataclasses import dataclass

@tool
def get_message_count(runtime: ToolRuntime):
    """Get the total number of messages exchanged in the conversation."""
    messages = runtime.state['messages']
    context = runtime.context
    return f"User '{context.user_id}' with Session '{context.session_id}' has '{len(messages)}' messages."

@dataclass
class UserContext:
    user_id: str
    session_id: str

agent = create_agent(model=model,
                     tools=[base_tools.get_weather, get_message_count],
                     system_prompt='You are a helpful AI assistant.',
                     context_schema=UserContext)

user_context = UserContext(user_id='kgptalkie', session_id='session_1')
agent.invoke({'messages': [HumanMessage('weather in Mumbai then how many messages are in this conversation')]},
             context=user_context)

from langgraph.checkpoint.postgres import PostgresSaver
import psycopg

pg_conn = psycopg.connect(os.getenv("POSTGRESQL_URL"), autocommit=True)
checkpointer = PostgresSaver(pg_conn)
checkpointer.setup()

config = {"configurable": {"thread_id": "user_123"}}
agent = create_agent(model=model, system_prompt=system_prompt, checkpointer=checkpointer)

from langchain.tools import tool, ToolRuntime
from pathlib import Path

@tool
def save_conversation_summary(summary: str, runtime: ToolRuntime):
    """Save conversation summary to disk for context offloading."""
    user_id = runtime.context.user_id
    thread_id = runtime.context.thread_id

    summary_dir = Path(f"data/{user_id}/{thread_id}")
    summary_dir.mkdir(parents=True, exist_ok=True)
    summary_path = summary_dir / "summary.md"
    summary_path.write_text(summary)
    return f"Summary saved to {summary_path}"

from langchain.messages import RemoveMessage, ToolMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langgraph.types import Command

@tool
def load_conversation_summary(runtime: ToolRuntime):
    """Load previous conversation summary from disk."""
    user_id = runtime.context.user_id
    thread_id = runtime.config['configurable']['thread_id']
    summary_path = Path(f"data/{user_id}/{thread_id}/summary.md")

    if not summary_path.exists():
        return Command(update={'messages': [
            ToolMessage("No previous summary found.", tool_call_id=runtime.tool_call_id)]})

    summary_text = summary_path.read_text()
    messages = runtime.state.get('messages', [])
    last_ai_message = messages[-1] if messages else None

    new_messages = [
        RemoveMessage(id=REMOVE_ALL_MESSAGES),
        HumanMessage(f"Previous conversation summary:\n{summary_text}"),
    ]
    if last_ai_message:
        new_messages.append(last_ai_message)
    new_messages.append(ToolMessage("Successfully loaded previous summary.", tool_call_id=runtime.tool_call_id))

    return Command(update={'messages': new_messages})

from langchain_google_genai import GoogleGenerativeAIEmbeddings
from langgraph.store.postgres import PostgresStore
import psycopg

embeddings = GoogleGenerativeAIEmbeddings(model='gemini-embedding-001')

def embed(texts: list[str]):
    return embeddings.embed_documents(texts, output_dimensionality=768)

pg_conn = psycopg.connect(os.getenv('POSTGRESQL_URL'), autocommit=True)
store = PostgresStore(pg_conn, index={'embed': embed, 'dims': 768})
store.setup()

from langchain.tools import tool, ToolRuntime
from dataclasses import dataclass

@dataclass
class UserContext:
    user_id: str

@tool
def save_user_memory(category: str, information: dict, runtime: ToolRuntime):
    """Save user preference or information to long-term memory.

    Examples:
        category='food', information={'diet': 'vegetarian', 'likes': ['pasta']}
        category='work', information={'role': 'Data Scientist', 'interests': ['AI', 'ML']}
    """
    store = runtime.store
    user_id = runtime.context.user_id
    namespace = (user_id, "preferences")
    store.put(namespace=namespace, key=category, value=information)
    return f"Saved {category} preferences for {user_id}"

@tool
def get_user_memory(category: str, runtime: ToolRuntime):
    """Retrieve user preference or information from long-term memory."""
    store = runtime.store
    user_id = runtime.context.user_id
    namespace = (user_id, 'preferences')
    item = store.get(namespace=namespace, key=category)
    if item:
        return f"{category}: {item.value}"
    return f"No '{category}' information found"

from langgraph.checkpoint.postgres import PostgresSaver

checkpointer = PostgresSaver(psycopg.connect(os.getenv("POSTGRESQL_URL"), autocommit=True))
store = PostgresStore(psycopg.connect(os.getenv('POSTGRESQL_URL'), autocommit=True),
                      index={'embed': embed, 'dims': 768})

agent = create_agent(
    model=model,
    tools=[base_tools.web_search, save_user_memory, get_user_memory],
    checkpointer=checkpointer,
    store=store,
    context_schema=UserContext,
    system_prompt="You are a helpful assistant with long-term memory."
)

from langgraph.checkpoint.sqlite import SqliteSaver
import sqlite3

conn = sqlite3.connect("db/5_streaming_agent.db", check_same_thread=False)
agent = create_agent(model=model,
                     tools=[base_tools.web_search, base_tools.get_weather],
                     checkpointer=SqliteSaver(conn))

config = {'configurable': {'thread_id': '05_session_1'}}

# updates mode — see each node fire
for chunk in agent.stream(
        {'messages': [HumanMessage('what is the weather in Mumbai?')]},
        config=config, stream_mode='updates'):
    print(chunk)

from langchain.messages import RemoveMessage
from langgraph.graph.message import REMOVE_ALL_MESSAGES
from langchain.agents.middleware import before_model
from langgraph.runtime import Runtime
from langchain.agents import AgentState

@before_model
def trim_messages(state: AgentState, runtime: Runtime):
    """Keep only the first and last message to fit the context window."""
    messages = state['messages']
    if len(messages) <= 3:
        return None
    return {'messages': [RemoveMessage(id=REMOVE_ALL_MESSAGES), messages[0], messages[-1]]}

agent = create_agent(model=model,
                     tools=[base_tools.web_search, base_tools.get_weather],
                     checkpointer=InMemorySaver(),
                     middleware=[trim_messages])

from langchain.agents.middleware import SummarizationMiddleware

agent = create_agent(
    model=model,
    tools=[base_tools.get_weather, base_tools.web_search],
    checkpointer=InMemorySaver(),
    middleware=[SummarizationMiddleware(
        model=ChatGoogleGenerativeAI(model='gemini-3-pro-preview'),
        trigger=[('messages', 15)],
        keep=("messages", 5)
    )]
)

from langchain.agents.middleware import wrap_model_call, ModelRequest

basic_model = ChatGoogleGenerativeAI(model='gemini-2.5-flash')
advanced_model = ChatGoogleGenerativeAI(model='gemini-3-pro-preview')

@wrap_model_call
def dynamic_model_selection(request: ModelRequest, handler):
    """Choose model based on conversation complexity."""
    count = len(request.state['messages'])
    model = advanced_model if count > 5 else basic_model
    return handler(request.override(model=model))

from langchain.agents.middleware import (
    ModelCallLimitMiddleware, ToolCallLimitMiddleware,
    TodoListMiddleware, ModelFallbackMiddleware
)

agent = create_agent(
    model=basic_model,
    tools=[base_tools.web_search, base_tools.get_weather],
    checkpointer=InMemorySaver(),
    middleware=[
        dynamic_model_selection,
        TodoListMiddleware(),
        ModelCallLimitMiddleware(run_limit=5, exit_behavior='end'),
        ToolCallLimitMiddleware(run_limit=5, exit_behavior='continue'),
        ModelFallbackMiddleware(first_model=ChatGoogleGenerativeAI(model='gemini-3-flash-preview')),
    ]
)

from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.tools import tool
from langgraph.types import Command

@tool
def write_file(path: str, content: str):
    """Write content to file."""
    with open(path, 'w') as f:
        f.write(content)
    return f"Successfully wrote to {path}"

@tool
def execute_sql(query: str):
    """Execute SQL query. Use this tool for any database related question."""
    return f"Would execute: {query}"

agent = create_agent(
    model=model,
    tools=[write_file, execute_sql],
    checkpointer=InMemorySaver(),
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={
            "write_file": True,                                   # approve, edit, or reject
            "execute_sql": {"allowed_decisions": ["approve", "reject"]},  # no editing
        },
        description_prefix="Tool execution pending approval"
    )]
)

constrained_prompt = """You are a medical information assistant.

What you CAN do:
- Provide general health information and explain terminology

What you CANNOT do:
- Diagnose conditions, prescribe medication, or replace professional advice

ALWAYS include disclaimer: "This is not medical advice. Consult a healthcare professional."
"""

few_shot_prompt = """You are a product description writer for an e-commerce site.

Format like these examples:

Example 1:
Product: Wireless Mouse
Description: Glide through your workday with our ergonomic wireless mouse.
- 18-month battery life | 6 programmable buttons | Up to 30ft range
Perfect for: Professionals, gamers, everyday users

Use this format for all product descriptions.
"""

format_control_prompt = """You are a stock analysis agent.

ALWAYS format your analysis exactly like this:

## [COMPANY NAME] Analysis
**Current Price:** $XXX.XX
**Change:** ±X.XX%
### Key Metrics
• Market Cap: $XXX billion
### Recommendation
BUY | HOLD | SELL

NEVER deviate from this format.
"""