Model Integration

from langchain_core.tools import tool
from langchain_openai import ChatOpenAI

# Define tools
@tool
def web_search(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

@tool
def calculator(expression: str) -> str:
    """Evaluate a mathematical expression."""
    return str(eval(expression))

tools = [web_search, calculator]

# Bind tools to the model
llm = ChatOpenAI(model="gpt-4o", temperature=0.0)
llm_with_tools = llm.bind_tools(tools)   # ← Binding happens here

def agent_node(state: AgentState):
    # The model now knows about the tools
    response = llm_with_tools.invoke(state["messages"])
    
    # Response may contain tool_calls
    if response.tool_calls:
        return {"messages": [response]}   # Let ToolNode handle execution
    else:
        return {"messages": [response]}   # Normal response

• The model receives tool schemas in the system prompt
• When appropriate, it outputs a tool_calls list
• LangGraph’s ToolNode executes those calls

from langgraph.prebuilt import ToolNode

# Create ToolNode
tool_node = ToolNode(tools)

# Full agent loop
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

graph.add_edge(START, "agent")
graph.add_conditional_edges(
    "agent",
    lambda state: "tools" if state["messages"][-1].tool_calls else "END"
)
graph.add_edge("tools", "agent")   # Loop back after tool use

def agent_node(state: AgentState):
    response = llm_with_tools.invoke(state["messages"])
    
    if response.tool_calls:
        return {"messages": [response]}   # LLM wants to call tools
    
    return {"messages": [response]}       # Normal final answer


def tools_node(state: AgentState):
    # ToolNode automatically executes tool_calls and returns ToolMessages
    tool_results = tool_node.invoke(state)
    return tool_results   # Contains ToolMessage objects

# 1. LLM decides to call tool
AIMessage(
    content="",
    tool_calls=[{
        "name": "web_search",
        "args": {"query": "LangGraph tutorial"},
        "id": "call_abc123"
    }]
)

# 2. Tool executes and returns
ToolMessage(
    content="LangGraph is a library for building stateful agents...",
    tool_call_id="call_abc123"
)

llm_with_tools = llm.bind_tools(
    tools,
    tool_choice="auto",           # Let model decide
    # tool_choice="any"           # Force tool use
)

• Tool binding transforms a regular LLM into a reasoning agent
• Use llm.bind_tools(tools) to enable function calling
• Combine with ToolNode for seamless execution
• Always include good tool descriptions — they heavily influence model performance

Prompt + State Injection

from langchain_core.prompts import ChatPromptTemplate

def agent_node(state: AgentState):
    prompt = ChatPromptTemplate.from_messages([
        ("system", """You are a helpful LangGraph expert.
        Use the following context when relevant: {context}"""),
        
        MessagesPlaceholder(variable_name="messages"),   # Injects full history
    ])
    
    chain = prompt | llm
    
    response = chain.invoke({
        "messages": state["messages"],
        "context": state.get("documents", [])   # Injected additional context
    })
    
    return {"messages": [response]}

Context Window Limit Handling

from langchain_core.messages import trim_messages

def context_aware_agent_node(state: AgentState):
    # Trim history while keeping system message and recent turns
    trimmed_messages = trim_messages(
        state["messages"],
        max_tokens=12000,           # Safe limit for gpt-4o
        strategy="last",            # Keep recent messages
        token_counter=llm,          # Uses model's tokenizer
        include_system=True,        # Always keep SystemMessage
        allow_partial=True
    )
    
    # Add retrieved documents as context
    context = "\n\n".join([doc['content'] for doc in state.get("documents", [])])
    
    response = llm.invoke(trimmed_messages + [
        HumanMessage(content=f"Additional context:\n{context}")
    ])
    
    return {"messages": [response]}

Conversation History Management

class AgentState(MessagesState):
    messages: Annotated[list, add_messages]   # Automatic appending

def smart_history_node(state: AgentState):
    if len(state["messages"]) > 20:
        # Summarize older messages
        summary_prompt = ChatPromptTemplate.from_template(
            "Summarize the conversation so far in 4-5 sentences:\n\n{history}"
        )
        summary = llm.invoke(summary_prompt.format(
            history="\n".join([m.content for m in state["messages"][:-10]])
        ))
        
        # Replace old history with summary
        return {
            "messages": [
                SystemMessage(content=f"Previous conversation summary: {summary.content}"),
                *state["messages"][-10:]   # Keep recent messages
            ]
        }
    return {}

• Always be conscious of token limits
• Prefer relevant context over dumping everything
• Use summarization for long histories
• Keep SystemMessage persistent
• Design state to support smart context injection

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_groq import ChatGroq

# Model hierarchy
models = {
    "primary": ChatOpenAI(model="gpt-4o", temperature=0.0),
    "backup_fast": ChatOpenAI(model="gpt-4o-mini", temperature=0.7),
    "backup_strong": ChatAnthropic(model="claude-3-5-sonnet-20240620"),
    "emergency": ChatGroq(model="llama3-70b-8192")
}

class FallbackAgent:
    def __init__(self):
        self.models = {
            "high": ChatOpenAI(model="gpt-4o"),
            "medium": ChatOpenAI(model="gpt-4o-mini"),
            "low": ChatGroq(model="llama3-70b-8192")
        }
    
    async def invoke(self, messages, complexity: str = "medium"):
        # Try high capability first
        try:
            return await self.models["high"].ainvoke(messages)
        except Exception:
            print("High-end model unavailable. Falling back...")
            
            # Try medium
            try:
                return await self.models["medium"].ainvoke(messages)
            except Exception:
                print("Medium model also failed. Using low-cost model.")
                # Final fallback
                return await self.models["low"].ainvoke(messages)

# Usage
fallback_agent = FallbackAgent()

async def agent_node(state: AgentState):
    response = await fallback_agent.invoke(state["messages"])
    return {"messages": [response]}

1. Always define at least 2–3 fallback models
2. Track failure counts per model in state
3. Use different providers for true redundancy (OpenAI + Anthropic + Groq)
4. Implement exponential backoff between retries
5. Log fallback events for monitoring
6. Gracefully degrade features when using weaker models

class ResilientLLM:
    def __init__(self):
        self.models = [
            ChatOpenAI(model="gpt-4o"),
            ChatAnthropic(model="claude-3-5-sonnet-20240620"),
            ChatOpenAI(model="gpt-4o-mini")
        ]
    
    async def invoke_with_fallback(self, messages, max_attempts=3):
        for i, llm in enumerate(self.models[:max_attempts]):
            try:
                return await llm.ainvoke(messages)
            except Exception as e:
                print(f"Model {i+1} failed: {type(e).__name__}")
                continue
        raise Exception("All models failed")

Model Evaluation & Debugging

from langchain_core.messages import HumanMessage
import pytest

def test_agent_response_quality():
    test_cases = [
        ("What is LangGraph?", "should mention stateful graphs"),
        ("How do I create a cycle?", "should mention conditional edges"),
    ]
    
    for query, expected_keyword in test_cases:
        state = {"messages": [HumanMessage(content=query)]}
        result = agent_node(state)
        
        response = result["messages"][-1].content.lower()
        assert any(word in response for word in expected_keyword.split()), \
            f"Response missing key concept: {expected_keyword}"

1. Implement automated tests for critical agent behaviors
2. Use LLM-as-Judge for scalable evaluation
3. Add hallucination detection in validation nodes
4. Log raw model inputs/outputs during development
5. Track consistency metrics over multi-turn conversations
6. Build debug nodes that can be toggled on/off
7. Monitor confidence scores and retry when low

Testing Model Outputs

import pytest
from langchain_core.messages import HumanMessage

def test_agent_knowledge():
    test_cases = [
        ("What is LangGraph?", ["state graph", "cycles", "nodes"]),
        ("How do you create a cycle?", ["conditional edges", "loop"]),
    ]
    
    for query, expected_keywords in test_cases:
        state = {"messages": [HumanMessage(content=query)]}
        result = agent_node(state)  # Your agent node
        
        response_text = result["messages"][-1].content.lower()
        
        for keyword in expected_keywords:
            assert keyword in response_text, \
                f"Expected '{keyword}' in response to: {query}"

def debug_model_node(state: AgentState):
    print("\n=== MODEL DEBUG ===")
    print(f"Input tokens: {count_tokens(state['messages'])}")
    print(f"Last message: {state['messages'][-1].content[:300]}...")
    
    response = llm.invoke(state["messages"])
    
    print(f"Output: {response.content[:400]}...")
    if hasattr(response, 'tool_calls') and response.tool_calls:
        print(f"Tool Calls: {len(response.tool_calls)}")
    
    return {"messages": [response]}

from langchain_core.callbacks import BaseCallbackHandler

class ModelDebugHandler(BaseCallbackHandler):
    def on_llm_start(self, serialized, prompts, **kwargs):
        print(f"\n[LLM Start] Model: {serialized.get('name', 'unknown')}")
    
    def on_llm_new_token(self, token: str, **kwargs):
        print(token, end="", flush=True)
    
    def on_llm_end(self, response, **kwargs):
        print("\n[LLM End]")

# Attach to model
llm = ChatOpenAI(model="gpt-4o", callbacks=[ModelDebugHandler()])

• Implement automated tests for critical paths
• Use LLM-as-Judge for scalable evaluation
• Add hallucination detection in validation nodes
• Log inputs/outputs during development
• Monitor consistency across conversations
• Use breakpoints + state inspection heavily during development

• High cost
• Slower response times
• Rate limit issues
• No specialization

• Using Claude-style verbose prompts with GPT-4o-mini
• Not adjusting temperature per model
• Ignoring model-specific formatting (e.g., Gemini needs special system message handling)

# Same prompt for all models
prompt = ChatPromptTemplate.from_template("You are a helpful assistant. {input}")

def get_prompt_for_model(model_name: str, task: str):
    if "claude" in model_name.lower():
        return ChatPromptTemplate.from_template(
            "You are an expert AI assistant. Think step by step.\n\n{input}"
        )
    else:
        return ChatPromptTemplate.from_template(
            "You are a helpful assistant.\n\n{input}"
        )

3. Ignoring Token Limits

def agent_node(state):
    # No trimming - can easily exceed context window
    response = llm.invoke(state["messages"])  
    return {"messages": [response]}

from langchain_core.messages import trim_messages

def token_aware_agent_node(state: AgentState):
    trimmed = trim_messages(
        state["messages"],
        max_tokens=12000,
        strategy="last",
        token_counter=llm,
        include_system=True
    )
    
    response = llm.invoke(trimmed)
    return {"messages": [response]}

4. Not Handling Failures

def agent_node(state):
    response = llm.invoke(state["messages"])   # No error handling!
    return {"messages": [response]}

async def resilient_agent_node(state: AgentState, attempt: int = 0):
    try:
        response = await llm.ainvoke(state["messages"])
        return {"messages": [response]}
    except Exception as e:
        if attempt < 2:
            print(f"Model call failed (attempt {attempt+1}). Retrying...")
            return await resilient_agent_node(state, attempt + 1)
        else:
            # Graceful fallback
            return {
                "messages": [AIMessage(
                    content="I'm having trouble connecting right now. "
                           "Could you please rephrase your request?"
                )]
            }

Best Practices for Model Integration

Use Multiple Models Strategically

# Define specialized models
models = {
    "fast": ChatOpenAI(model="gpt-4o-mini", temperature=0.7),      # Routing, simple tasks
    "smart": ChatOpenAI(model="gpt-4o", temperature=0.0),          # Complex reasoning
    "creative": ChatAnthropic(model="claude-3-5-sonnet-20240620"), # Writing & ideation
    "code": ChatGroq(model="llama3-70b-8192")                      # Code generation
}

def route_to_model(task_type: str):
    if task_type == "simple" or task_type == "routing":
        return models["fast"]
    elif task_type == "reasoning":
        return models["smart"]
    elif task_type == "creative":
        return models["creative"]
    else:
        return models["code"]

Separate Planning vs Execution Models

async def planning_node(state: AgentState):
    # Use powerful model for high-quality planning
    planner = ChatAnthropic(model="claude-3-5-sonnet-20240620", temperature=0.0)
    plan = await planner.ainvoke(state["messages"])
    return {"plan": plan.content, "messages": [plan]}

async def execution_node(state: AgentState):
    # Use fast model to execute the plan
    executor = ChatOpenAI(model="gpt-4o-mini", temperature=0.3)
    prompt = f"Follow this plan:\n{state['plan']}\n\nExecute it now."
    result = await executor.ainvoke(prompt)
    return {"messages": [result]}

Optimize for Cost and Latency

class ModelSelector:
    def __init__(self):
        self.models = {
            "fast": ChatOpenAI(model="gpt-4o-mini"),
            "balanced": ChatOpenAI(model="gpt-4o"),
            "premium": ChatAnthropic(model="claude-3-5-sonnet-20240620")
        }
    
    def select(self, complexity: str, budget_mode: str = "balanced"):
        if budget_mode == "cheap" or complexity == "low":
            return self.models["fast"]
        elif complexity == "high":
            return self.models["premium"]
        else:
            return self.models["balanced"]

# Usage
selector = ModelSelector()

def agent_node(state: AgentState):
    complexity = analyze_query_complexity(state["messages"][-1].content)
    llm = selector.select(complexity, budget_mode="balanced")
    
    response = llm.invoke(state["messages"])
    return {"messages": [response], "model_used": llm.model_name}

Always Add Fallbacks

async def resilient_invoke(messages, max_attempts=3):
    model_list = [
        ChatOpenAI(model="gpt-4o"),
        ChatAnthropic(model="claude-3-5-sonnet-20240620"),
        ChatOpenAI(model="gpt-4o-mini")
    ]
    
    for i, llm in enumerate(model_list[:max_attempts]):
        try:
            return await llm.ainvoke(messages)
        except Exception as e:
            print(f"Model {i+1} failed: {type(e).__name__}. Trying next...")
            continue
    
    raise Exception("All fallback models failed.")

async def agent_node(state: AgentState):
    try:
        response = await resilient_invoke(state["messages"])
        return {"messages": [response]}
    except Exception:
        return {"messages": [AIMessage(content="I'm currently experiencing technical difficulties. Please try again later.")]}

Combine with State + Memory Properly

class AgentState(MessagesState):
    messages: Annotated[list, add_messages]
    documents: list[dict] = Field(default_factory=list)
    plan: str | None = None
    model_used: str | None = None          # Track which model was used
    confidence: float = 0.0

def agent_node(state: AgentState):
    llm = select_model_based_on_state(state)
    
    response = llm.invoke(state["messages"])
    
    return {
        "messages": [response],
        "model_used": llm.model_name,
        "confidence": estimate_confidence(response)
    }