ReAct Agent Pattern

from typing import Annotated, Sequence
from langchain_core.messages import BaseMessage, SystemMessage
from langgraph.graph import StateGraph, START, END, add_messages
from typing_extensions import TypedDict

class ReactState(TypedDict):
    """State for ReAct agent."""
    messages: Annotated[Sequence[BaseMessage], add_messages]
    iterations: int

from langchain_core.tools import tool

@tool
def search_wikipedia(query: str) -> str:
    """Search Wikipedia for information.
    
    Args:
        query: The search query
        
    Returns:
        Summary from Wikipedia
    """
    # Mock implementation - replace with real Wikipedia API
    wiki_data = {
        "python": "Python is a high-level programming language known for its simplicity...",
        "langgraph": "LangGraph is a framework for building stateful, multi-actor applications...",
        "ai": "Artificial Intelligence (AI) is the simulation of human intelligence..."
    }
    query_lower = query.lower()
    for key, value in wiki_data.items():
        if key in query_lower:
            return value
    return f"No Wikipedia article found for '{query}'"

@tool
def calculate_math(expression: str) -> str:
    """Evaluate a mathematical expression.
    
    Args:
        expression: Math expression to evaluate
        
    Returns:
        The calculated result
    """
    try:
        result = eval(expression)
        return f"Result: {result}"
    except Exception as e:
        return f"Error calculating: {str(e)}"

@tool
def analyze_text(text: str) -> str:
    """Analyze text and provide statistics.
    
    Args:
        text: Text to analyze
        
    Returns:
        Text statistics
    """
    word_count = len(text.split())
    char_count = len(text)
    sentence_count = text.count('.') + text.count('!') + text.count('?')
    
    return f"""Text Analysis:
    - Words: {word_count}
    - Characters: {char_count}
    - Sentences: {sentence_count}
    """

tools = [search_wikipedia, calculate_math, analyze_text]

from langchain_openai import ChatOpenAI

# Initialize model with tools
model = ChatOpenAI(model="gpt-4", temperature=0)
model_with_tools = model.bind_tools(tools)

# ReAct system prompt
REACT_PROMPT = """You are a ReAct (Reasoning and Acting) agent. 

For each user query, follow this pattern:
1. THOUGHT: Reason about what you need to do
2. ACTION: Use a tool if needed
3. OBSERVATION: Analyze the result
4. Repeat until you can provide a final answer

Be explicit about your reasoning process. Think step-by-step.
"""

def agent_node(state: ReactState) -> dict:
    """ReAct agent that reasons before acting."""
    messages = state["messages"]
    iterations = state.get("iterations", 0)
    
    # Add system prompt for ReAct pattern
    full_messages = [
        SystemMessage(content=REACT_PROMPT),
        *messages
    ]
    
    # Call model
    response = model_with_tools.invoke(full_messages)
    
    return {
        "messages": [response],
        "iterations": iterations + 1
    }

from langgraph.prebuilt import ToolNode

tool_node = ToolNode(tools)

MAX_ITERATIONS = 10

def should_continue(state: ReactState) -> str:
    """Determine next step in ReAct loop."""
    messages = state["messages"]
    last_message = messages[-1]
    iterations = state.get("iterations", 0)
    
    # Check iteration limit
    if iterations >= MAX_ITERATIONS:
        return "end"
    
    # If model wants to use tools, execute them
    if last_message.tool_calls:
        return "continue"
    
    # Otherwise, we're done
    return "end"

# Initialize graph
graph = StateGraph(ReactState)

# Add nodes
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)

# Set entry point
graph.add_edge(START, "agent")

# Add conditional routing from agent
graph.add_conditional_edges(
    "agent",
    should_continue,
    {
        "continue": "tools",
        "end": END
    }
)

# After tools, return to agent for reasoning
graph.add_edge("tools", "agent")

# Compile
app = graph.compile()

from langchain_core.messages import HumanMessage

# Simple query
result = app.invoke({
    "messages": [HumanMessage(content="What is LangGraph and calculate 10 * 15")],
    "iterations": 0
})
print(result["messages"][-1].content)
# Agent will:
# 1. Think: Need to search for LangGraph AND do a calculation
# 2. Act: Use search_wikipedia tool
# 3. Observe: Read Wikipedia result
# 4. Act: Use calculate_math tool
# 5. Observe: Get calculation result
# 6. Respond: Provide comprehensive answer

# Complex multi-step query
result = app.invoke({
    "messages": [HumanMessage(
        content="""Find information about Python, then analyze 
        the text you found, and finally calculate how many 
        words per sentence on average."""
    )],
    "iterations": 0
})
print(f"Iterations used: {result['iterations']}")
print(result["messages"][-1].content)
# Agent will reason through multiple steps:
# 1. Search Wikipedia for Python
# 2. Analyze the returned text
# 3. Calculate words per sentence
# 4. Provide final answer

from typing import Annotated, Sequence
from langchain_core.messages import BaseMessage, HumanMessage, SystemMessage
from langchain_core.tools import tool
from langchain_openai import ChatOpenAI
from langgraph.graph import StateGraph, START, END, add_messages
from langgraph.prebuilt import ToolNode
from typing_extensions import TypedDict

# State
class ReactState(TypedDict):
    messages: Annotated[Sequence[BaseMessage], add_messages]
    iterations: int

# Tools
@tool
def search_wikipedia(query: str) -> str:
    """Search Wikipedia for information."""
    return f"Python is a high-level programming language known for simplicity."

@tool
def calculate_math(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        return f"Result: {eval(expression)}"
    except Exception as e:
        return f"Error: {str(e)}"

tools = [search_wikipedia, calculate_math]

# Model
model = ChatOpenAI(model="gpt-4", temperature=0)
model_with_tools = model.bind_tools(tools)

REACT_PROMPT = """You are a ReAct agent. Think step-by-step:
1. THOUGHT: Reason about the task
2. ACTION: Use tools if needed
3. OBSERVATION: Analyze results
Repeat until you can answer."""

# Nodes
def agent_node(state: ReactState) -> dict:
    messages = [SystemMessage(content=REACT_PROMPT), *state["messages"]]
    response = model_with_tools.invoke(messages)
    return {"messages": [response], "iterations": state.get("iterations", 0) + 1}

tool_node = ToolNode(tools)

# Router
MAX_ITERATIONS = 10

def should_continue(state: ReactState) -> str:
    if state.get("iterations", 0) >= MAX_ITERATIONS:
        return "end"
    if state["messages"][-1].tool_calls:
        return "continue"
    return "end"

# Graph
graph = StateGraph(ReactState)
graph.add_node("agent", agent_node)
graph.add_node("tools", tool_node)
graph.add_edge(START, "agent")
graph.add_conditional_edges("agent", should_continue, {"continue": "tools", "end": END})
graph.add_edge("tools", "agent")
app = graph.compile()

# Run with logging
print("ReAct Agent Starting...\n")
result = app.invoke({
    "messages": [HumanMessage(content="What is Python and what is 25 * 4?")],
    "iterations": 0
})
print(f"\nCompleted in {result['iterations']} iterations")
print(f"\nFinal Answer: {result['messages'][-1].content}")

python react_agent.py