LangChain-LangGraph Starter

from typing import Annotated, Sequence, TypedDict
from langchain_core.messages import BaseMessage
from langgraph.graph.message import add_messages

class AgentState(TypedDict):
    """The state of the agent."""
    messages: Annotated[Sequence[BaseMessage], add_messages]
    number_of_steps: int

from langchain_core.tools import tool
from geopy.geocoders import Nominatim
from pydantic import BaseModel, Field
import requests

geolocator = Nominatim(user_agent="weather-app")

class SearchInput(BaseModel):
    location: str = Field(description="The city and state, e.g., San Francisco")
    date: str = Field(description="the forecasting date format (yyyy-mm-dd)")

@tool("get_weather_forecast", args_schema=SearchInput, return_direct=True)
def get_weather_forecast(location: str, date: str):
    """Retrieves the weather using Open-Meteo API for a given location and date."""
    location_obj = geolocator.geocode(location)
    if location_obj:
        try:
            response = requests.get(
                f"https://api.open-meteo.com/v1/forecast?"
                f"latitude={location_obj.latitude}&longitude={location_obj.longitude}&"
                f"hourly=temperature_2m&start_date={date}&end_date={date}"
            )
            data = response.json()
            return {
                time: temp 
                for time, temp in zip(
                    data["hourly"]["time"], 
                    data["hourly"]["temperature_2m"]
                )
            }
        except Exception as e:
            return {"error": str(e)}
    else:
        return {"error": "Location not found"}

tools = [get_weather_forecast]

from langchain_community.chat_models import ChatLiteLLM

# Create LLM instance
llm = ChatLiteLLM(model="nebius/Qwen/Qwen3-235B-A22B")

# Bind tools to the model
model = llm.bind_tools([get_weather_forecast])

from langchain_core.messages import ToolMessage
from langchain_core.runnables import RunnableConfig

tools_by_name = {tool.name: tool for tool in tools}

# Tool execution node
def call_tool(state: AgentState):
    outputs = []
    for tool_call in state["messages"][-1].tool_calls:
        tool_result = tools_by_name[tool_call["name"]].invoke(tool_call["args"])
        outputs.append(
            ToolMessage(
                content=tool_result,
                name=tool_call["name"],
                tool_call_id=tool_call["id"],
            )
        )
    return {"messages": outputs}

# Model call node
def call_model(state: AgentState, config: RunnableConfig):
    response = model.invoke(state["messages"], config)
    return {"messages": [response]}

# Conditional edge
def should_continue(state: AgentState):
    messages = state["messages"]
    if not messages[-1].tool_calls:
        return "end"
    return "continue"

from langgraph.graph import StateGraph, END

# Define a new graph with our state
workflow = StateGraph(AgentState)

# Add nodes
workflow.add_node("llm", call_model)
workflow.add_node("tools", call_tool)

# Set entrypoint
workflow.set_entry_point("llm")

# Add conditional edge
workflow.add_conditional_edges(
    "llm",
    should_continue,
    {
        "continue": "tools",
        "end": END,
    },
)

# Add edge from tools back to llm
workflow.add_edge("tools", "llm")

# Compile the graph
graph = workflow.compile()

from datetime import datetime

# Create initial message
inputs = {
    "messages": [("user", f"What is the weather in Berlin on {datetime.today()}?")]
}

# Stream the agent's execution
for state in graph.stream(inputs, stream_mode="values"):
    last_message = state["messages"][-1]
    last_message.pretty_print()

# Initial query
inputs = {"messages": [("user", "What is the weather in Berlin today?")]}

for state in graph.stream(inputs, stream_mode="values"):
    last_message = state["messages"][-1]
    last_message.pretty_print()

# Follow-up query (maintains conversation state)
state["messages"].append(("user", "Would it be warmer in Reykjavik?"))

for state in graph.stream(state, stream_mode="values"):
    last_message = state["messages"][-1]
    last_message.pretty_print()

@tool
def get_historical_weather(location: str, date: str):
    """Get historical weather data"""
    # Implementation
    return data

tools = [get_weather_forecast, get_historical_weather]
model = llm.bind_tools(tools)

class AgentState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    number_of_steps: int
    user_preferences: dict  # Track user preferences
    error_count: int        # Track errors

def call_model(state: AgentState, config: RunnableConfig):
    response = model.invoke(state["messages"], config)
    return {
        "messages": [response],
        "number_of_steps": state["number_of_steps"] + 1
    }

def validate_output(state: AgentState):
    """Validate the agent's output before returning"""
    last_message = state["messages"][-1]
    # Validation logic
    return state

workflow.add_node("validator", validate_output)
workflow.add_edge("llm", "validator")

from langgraph.checkpoint.memory import MemorySaver

# Add memory for conversation persistence
memory = MemorySaver()
graph = workflow.compile(checkpointer=memory)

# Use with thread_id for conversation tracking
config = {"configurable": {"thread_id": "conversation-1"}}
for state in graph.stream(inputs, config, stream_mode="values"):
    last_message = state["messages"][-1]
    last_message.pretty_print()

from IPython.display import Image, display

display(Image(graph.get_graph().draw_mermaid_png()))