This is a production-ready agent system built on LangGraph that shows how to orchestrate complex workflows using state machines. I designed it to handle real-world scenarios like conditional branching, parallel execution, error recovery, and situations where you need humans to step in and make decisions.
I chose LangGraph because it extends LangChain with exactly what you need for real workflows:
- State Machines: Explicit state management for agent workflows
- Conditional Edges: Dynamic routing based on state
- Cycles: Support for iterative/reflective workflows
- Persistence: Save and resume workflow state
- Checkpointing: Recovery from failures
- Visualization: Built-in workflow graph visualization
Here's how I structured everything:
The workflow logic is encapsulated within the BaseWorkflow class in base_workflow.py. It manages the StateGraph compilation, node execution, and checkpointing using LangGraph's MemorySaver.
We use Pydantic models in models.py to define the WorkflowState. This provides type safety and validation as the state transitions between nodes.
Nodes are individual units of work located in the backend/nodes/ directory. Each node (e.g., llm_node.py) implements an execute method that takes the current state and returns an update.
We've implemented several common agentic patterns:
Located in approval_workflow.py. It demonstrates Human-in-the-loop (HITL) capabilities by pausing for approval before proceeding to execution.
Located in parallel_workflow.py. Shows how to branch a workflow into multiple parallel nodes and aggregate results.
Located in conditional_workflow.py. Uses conditional edges to route the workflow based on analysis of the current state.
Located in iterative_workflow.py. Implements a loop where the agent generates, reviews, and refines its output until quality criteria are met.
The system leverages LangGraph's checkpointing (in base_workflow.py) to allow workflows to be resumed and state to be persisted.
The approval_node.py demonstrates how to pause execution and wait for external input, a key requirement for production agent systems.
langgraph_workflow_agent/
├── backend/
│ ├── workflows/ # LangGraph graph definitions
│ │ ├── base_workflow.py
│ │ ├── approval_workflow.py
│ │ ├── parallel_workflow.py
│ │ ├── conditional_workflow.py
│ │ └── iterative_workflow.py
│ ├── nodes/ # Logic for individual graph nodes
│ │ ├── base_node.py
│ │ ├── llm_node.py
│ │ ├── approval_node.py
│ │ └── conditional_node.py
│ ├── agents/ # Agent wrappers for LLMs
│ │ └── base_agent.py
│ ├── core/ # Core system utilities
│ └── models.py # Pydantic & TypedDict models
├── frontend/ # UI implementations
│ ├── chainlit_app.py # Main Chat UI
│ └── gradio_app.py # Alternative UI
└── requirements.txt
├── data/ │ ├── workflows/ │ └── history/ ├── tests/ │ ├── test_workflows.py │ └── test_nodes.py ├── requirements.txt ├── README.md └── ARCHITECTURE.md
## Workflow State Model
```python
class WorkflowState(TypedDict):
"""State that flows through the workflow."""
messages: List[BaseMessage]
data: Dict[str, Any]
metadata: Dict[str, Any]
current_step: str
completed_steps: List[str]
errors: List[str]
human_input: Optional[Dict[str, Any]]
approval_status: Optional[str]
Start → Prepare → Submit → [Human Review] → Approved?
→ Yes → Execute → Notify → End
→ No → Reject → Notify → End
Start → Split → [Agent A] ┐
→ [Agent B] ├→ Merge → Synthesize → End
→ [Agent C] ┘
Start → Analyze → Condition?
→ Path A → Process A → Merge → End
→ Path B → Process B → Merge → End
→ Path C → Process C → Merge → End
Start → Generate → Review → Quality Check
→ Good? → Yes → End
→ No → Refine → [Loop back to Review]
Here's what makes this system useful:
- State Machine-Based: You always know exactly what state your workflow is in. No guessing.
- Conditional Routing: Workflows can branch dynamically based on what's happening. One input might go down path A, another down path B.
- Parallel Execution: Why wait? Run multiple agents at the same time when tasks are independent.
- Error Recovery: If something breaks, you can retry or resume from a checkpoint. No need to start over.
- Human Integration: Pause workflows to get human input when needed. Approvals, clarifications, whatever.
- Persistence: Everything is saved, so long-running workflows can survive restarts.
- Visualization: See your workflows as graphs. Makes debugging and understanding much easier.
- Monitoring: Track what's happening in real-time. Know when things are working and when they're not.
I kept the stack simple and focused:
- LangGraph: Does the heavy lifting for workflow orchestration
- LangChain: Provides the agent framework and LLM integration
- Gemini API: For LLM access (you could swap this out for other providers)
- Gradio: Modern UI framework that's easy to use and looks good
- Pydantic: For state validation and type safety
If you're running this in production, here are some things to track:
- Workflow completion rate (how many finish successfully)
- Average execution time (how long workflows take)
- Error recovery success rate (how often retries work)
- Human intervention rate (how often you need to step in)
- State transition accuracy (are workflows following the right paths)
These metrics help you understand if your workflows are working well and where you might need to improve things.