Langgraph

from typing import Annotated, TypedDictfrom langgraph.graph import StateGraph, START, ENDfrom langgraph.graph.message import add_messagesfrom langgraph.prebuilt import ToolNodefrom langchain_openai import ChatOpenAIfrom langchain_core.tools import tool
# 1. Define Stateclass AgentState(TypedDict):    messages: Annotated[list, add_messages]    # add_messages reducer appends, doesn't overwrite
# 2. Define Tools@tooldef search(query: str) -> str:    """Search the web for information."""    # Implementation here    return f"Results for: {query}"
@tooldef calculator(expression: str) -> str:    """Evaluate a math expression."""    return str(eval(expression))
tools = [search, calculator]
# 3. Create LLM with toolsllm = ChatOpenAI(model="gpt-4o").bind_tools(tools)
# 4. Define Nodesdef agent(state: AgentState) -> dict:    """The agent node - calls LLM."""    response = llm.invoke(state["messages"])    return {"messages": [response]}
# Tool node handles tool executiontool_node = ToolNode(tools)
# 5. Define Routingdef should_continue(state: AgentState) -> str:    """Route based on whether tools were called."""    last_message = state["messages"][-1]    if last_message.tool_calls:        return "tools"    return END
# 6. Build Graphgraph = StateGraph(AgentState)
# Add nodesgraph.add_node("agent", agent)graph.add_node("tools", tool_node)
# Add edgesgraph.add_edge(START, "agent")graph.add_conditional_edges("agent", should_continue, ["tools", END])graph.add_edge("tools", "agent")  # Loop back
# Compileapp = graph.compile()
# 7. Runresult = app.invoke({    "messages": [("user", "What is 25 * 4?")]})

from typing import Annotated, TypedDictfrom operator import addfrom langgraph.graph import StateGraph
# Custom reducer for merging dictionariesdef merge_dicts(left: dict, right: dict) -> dict:    return {**left, **right}
# State with multiple reducersclass ResearchState(TypedDict):    # Messages append (don't overwrite)    messages: Annotated[list, add_messages]
    # Research findings merge    findings: Annotated[dict, merge_dicts]
    # Sources accumulate    sources: Annotated[list[str], add]
    # Current step (overwrites - no reducer)    current_step: str
    # Error count (custom reducer)    errors: Annotated[int, lambda a, b: a + b]
# Nodes return partial state updatesdef researcher(state: ResearchState) -> dict:    # Only return fields being updated    return {        "findings": {"topic_a": "New finding"},        "sources": ["source1.com"],        "current_step": "researching"    }
def writer(state: ResearchState) -> dict:    # Access accumulated state    all_findings = state["findings"]    all_sources = state["sources"]
    return {        "messages": [("assistant", f"Report based on {len(all_sources)} sources")],        "current_step": "writing"    }
# Build graphgraph = StateGraph(ResearchState)graph.add_node("researcher", researcher)graph.add_node("writer", writer)# ... add edges

from langgraph.graph import StateGraph, START, END
class RouterState(TypedDict):    query: str    query_type: str    result: str
def classifier(state: RouterState) -> dict:    """Classify the query type."""    query = state["query"].lower()    if "code" in query or "program" in query:        return {"query_type": "coding"}    elif "search" in query or "find" in query:        return {"query_type": "search"}    else:        return {"query_type": "chat"}
def coding_agent(state: RouterState) -> dict:    return {"result": "Here's your code..."}
def search_agent(state: RouterState) -> dict:    return {"result": "Search results..."}
def chat_agent(state: RouterState) -> dict:    return {"result": "Let me help..."}
# Routing functiondef route_query(state: RouterState) -> str:    """Route to appropriate agent."""    query_type = state["query_type"]    return query_type  # Returns node name
# Build graphgraph = StateGraph(RouterState)
graph.add_node("classifier", classifier)graph.add_node("coding", coding_agent)graph.add_node("search", search_agent)graph.add_node("chat", chat_agent)
graph.add_edge(START, "classifier")
# Conditional edges from classifiergraph.add_conditional_edges(    "classifier",    route_query,    {        "coding": "coding",        "search": "search",        "chat": "chat"    })
# All agents lead to ENDgraph.add_edge("coding", END)graph.add_edge("search", END)graph.add_edge("chat", END)
app = graph.compile()