LangGraph Challenge (M3W8) - Gonzalo Paz

Description

• This project implements a stateful agent using LangGraph that can reason, call tools, handle errors safely and persist conversation state across runs (threads).

The system is design to demostrate:

• Controlled agent-tool loops
• Tool based computation
• Error handling and guardrails
• Checkpoint and replay
• Observability

Setup

1. Install dependencies

   pip install -r requirements.txt

2. Configure environment

   • Create a new .env file in the project root and add your OPENAI_API_KEY

3. Run the application

   python app.py

This will start a CLI interface to interact with the agent.

CLI Commands

Command	Description
/help	Show available commands
/new <thread_id>	Switch conversation thread
/thread	Show current thread
/state	Show current persisted state
/history	Show checkpoint history
/stream on/off	Enable or disable streaming
/test_math_chain	Run math chain acceptance test
/test_invalid_input	Run invalid input test
/test_loop_cap	Run loop guard test
/test_replay	Run replay/resume test
/exit	Exit CLI

System Design

State Schema and Reducers

The agent uses a shared state (GraphState) with reducers:

• messages: add_messages → accumulates conversation
• tool_calls: operator.add → counts tool usage
• retries: operator.add → tracks retry attempts
• errors: operator.add → stores structured errors

This allows:

• Persistent memory
• Traceability
• Accumulation of execution metadata

Graph Flow

The system follows a controlled loop:

START → agent → tools → agent → ... → END

Additional routes:

• tools → handle_tool_error → validation failure
• tools → exit → loop cap reached
• agent → END → no tool call needed

Loop Guard

A safety mechanism prevents infinite loops:

• MAX_TOOL_CALLS = 5
  • Computed per turn using message inspection
  • Enforced after tool execution

If exceeded:

• The graph routes to exit
• A safe message is returned to the user

Error Handling

There are two layers of validation:

1. Tool-level validation
   • Input validation inside tools (parse_number)
2. Graph-level handling
   • ToolNode(handle_tool_errors=True)
   • Routes to handle_tool_error

• Output
• Structured error stored in state
• User-friendly message returned

Checkpointing & Replay

The system uses:

• InMemorySaver as checkpointer
• thread_id as session key

This enables:

• Persistence across runs
• Replay of conversations
• State inspection

Demonstration

Run:

/test_replay

Expected behavior:

• Message count increases across turns
• Checkpoint history grows
• State continuity is preserved

Observability

The system supports full execution tracing via:

/stream on

This shows:

• Node execution sequence
• Intermediate states
• Tool calls per step

Each run includes a summary:

• Nodes executed
• Tool usage
• Exit reason

Required Tests

The system includes four reproducible tests:

1. Math Chain

• /test_math_chain
  • Uses add and multiply
  • Verifies correct result
  • Validates tool usage

2. Invalid Input

• /test_invalid_input
  • Forces tool usage
  • Triggers validation error
  • Ensures safe error handling

3. Loop Cap

• /test_loop_cap
  • Forces repeated tool calls
  • Validates guard condition
  • Ensures safe exit

4. Replay / Resume

• /test_replay
  • Runs multiple turns in same thread
  • Verifies state persistence

Design Decision: Forced Tool Usage in Some Tests

The default agent does not force tool usage, allowing natural handling of invalid inputs.

However, for specific tests (invalid_input, loop_cap), the system uses:

• tool_choice="required"

This ensures:

• Deterministic traversal through tool logic
• Full coverage of validation and guard paths

Project Structure

project/
├── app.py                 # Entry point
├── cli.py                 # CLI interface
├── config/                # Settings
│   └── settings.py
├── graph/                 # LangGraph components
│   ├── builder.py
│   ├── nodes.py
│   ├── routes.py
│   └── state.py
├── tools/                 # Tools + validation
│   ├── math_tools.py
│   └── validators.py
├── observability/         # Streaming + summaries
│   ├── stream.py
│   └── summaries.py
├── tests_cli/             # Acceptance tests
│   ├── acceptance.py
│   └── replay.py
└── utils/                 # Helpers
    └── messages.py