Planner-Executor Agent

A production-grade Planner-Executor agent built on LangGraph and LangChain. The system decomposes a user goal into an ordered plan, executes each task using a small set of tools, revises the plan when tasks fail, and synthesizes a final answer. It is designed as a template for enterprise AI workflows such as Deep Research, BI copilots, coding assistants, and data-analysis agents.

Features

• Planner that produces a structured, ordered list of tasks.
• Executor that selects the right tool per task (search, calculator, or "no tool" for pure synthesis) and runs it with retries.
• Replanner that revises the plan when tasks fail or information is missing, up to a configurable cap.
• Synthesizer that combines intermediate results into a coherent user-facing answer.
• Input and output guardrails that block prompt-injection attempts and redact secrets, API keys, and stack traces.
• Pluggable memory store (in-process by default; Redis/Postgres can be added behind the same protocol).
• Structured JSON logging with no print statements.
• Strict typing, pytest test suite with 90%+ coverage, and configuration that never hardcodes secrets.

Quick start

# 1. Install dependencies
pip install -r requirements.txt

# 2. Provide an LLM API key (optional - the system falls back to a
#    deterministic offline stub if no key is set)
cp .env.example .env
# Edit .env and set OPENAI_API_KEY=sk-...

# 3. Run the agent
python -m app.app "Analyze NVIDIA earnings and summarize implications."

# 4. (Optional) run the test suite
pytest

CLI

usage: python -m app.app <query>

Set APP_OUTPUT=json to print the result as a single JSON document.

When APP_OUTPUT=json, log records are routed to stderr so the JSON document on stdout is machine-readable.

Project layout

app/
  app.py                  # CLI / programmatic entry point
  config/
    settings.py           # YAML config + pydantic-settings (.env) merge
  graph/
    workflow.py           # LangGraph state machine
  agents/
    llm_factory.py        # Provider + offline stub factory
    planner.py
    executor.py
    replanner.py
    synthesizer.py
  models/
    state.py              # AgentState TypedDict
    plan_models.py        # Plan, RevisedPlan, TaskResult
  tools/
    search_tool.py        # SearchBackend protocol + stub
    calculator_tool.py    # Safe AST-based calculator
  guardrails/
    input_guardrail.py
    output_guardrail.py
  storage/
    memory_store.py       # InMemoryStore + protocol
config/
  config.yaml             # Non-secret defaults
tests/                    # pytest suite
docs/
  ARCHITECTURE.md         # Detailed architecture notes

Configuration

All non-secret knobs live in config/config.yaml. Secrets (API keys, OTLP endpoints, Redis/Postgres URLs) are read from .env via pydantic-settings. The two layers are merged at runtime.

Key options:

Section	Option	Purpose
llm	provider / model	Which chat model to use (or the stub).
planning	max_steps	Hard cap on plan length.
planning	replanning_enabled	Toggle the replanner entirely.
planning	max_replans	Maximum replanning attempts.
execution	max_retries	Per-task retry budget.
guardrails	redact_patterns	Regexes the output guardrail should strip.
storage	backend	memory today; redis / postgres planned.
tracing	enabled	Toggle LangSmith wiring.

See docs/ARCHITECTURE.md for the full design notes, including state shape, error-recovery semantics, and stretch goals.

Running the tests

ruff check .
mypy app
pytest

The test suite uses a tiny scripted fake LLM (tests.conftest._ScriptedLLM) so it runs without any real API calls. Coverage target is 80%+; current coverage is in the low 90s.

Extensibility

The system is intentionally modular. Common extensions:

• New tools: implement the SearchBackend protocol or add a function the executor can pick, and have the executor's heuristic + LLM prompt include it.
• Parallel execution: AgentState already uses an append reducer on task_results, so a Send()-based parallel branch can be added without changing the reducer contract.
• Human-in-the-loop: add a checkpoint node before the synthesizer and use langgraph.checkpoint to pause for approval.
• LangSmith tracing: set tracing.enabled: true in config.yaml and provide LANGCHAIN_API_KEY; the entry point configures the environment for you.