production-ready framework for orchestrating complex, stateful multi-agent workflows using LangGraph. This project demonstrates how to transition from simple linear LLM chains to sophisticated agentic state machines capable of conditional routing, parallel execution, iterative refinement, and human-in-the-loop interactions.
Built with reliability and scalability in mind, this agent system provides a blueprint for building advanced AI applications. It leverages LangGraph's state management to handle complex logic that traditional scripts struggle with:
- Stateful Orchestration: Robust state management across multi-step processes.
- Agentic Patterns: Implementation of common patterns like Parallelism, Loops, and Routing.
- Human-in-the-Loop: Seamless integration of human approval and feedback cycles.
- Resilience: Built-in checkpointing and error recovery mechanisms.
- Approval Workflow: Implements a review-and-approve cycle, perfect for content moderation or deployment gates.
- Parallel Processing: Scales tasks by executing multiple agents concurrently and synthesizing results.
- Iterative Refinement: A self-improving loop where agents generate, review, and refine output until quality thresholds are met.
- Conditional Routing: Dynamic path selection based on real-time analysis of the workflow state.
- Full Test Suite: Comprehensive unit and integration tests with LLM mocking.
- Multiple UIs: Interactive Chat UI via Chainlit and alternative Gradio interface.
- Detailed Architecture: Documented system design for easy extensibility.
- Type Safety: Built with Pydantic for rigorous data validation.
Recommended GitHub Topics: langchain, langgraph, ai-agents, workflow-automation, state-machine, python, llmops
- Clone the repository:
git clone <repository-url>
cd langgraph_workflow_agent- Install dependencies:
pip install -r requirements.txt- Set up environment variables:
cp .env.example .env
# Edit .env and add your API keyschainlit run frontend/chainlit_app.pyThe UI will open in your browser at http://localhost:8000
If port 8000 is already in use, you can specify a different port:
chainlit run frontend/chainlit_app.py --port 8001Chainlit provides a chat-based interface perfect for interacting with workflows:
- Type
workflowsto see all available workflows - Type
graph <workflow_name>to see a workflow's Mermaid diagram - Type
execute <workflow_name> <input>to run a workflow - Type
helpfor more information
The interface supports markdown and Mermaid diagrams, so you can see workflow graphs rendered beautifully.
from backend.workflows.approval_workflow import ApprovalWorkflow
# Initialize workflow
workflow = ApprovalWorkflow()
workflow.compile()
# Prepare input
initial_state = {
"messages": [],
"data": {"content": "Your content here"},
"metadata": {}
}
# Execute workflow
config = {"thread_id": "my-thread", "configurable": {"thread_id": "my-thread"}}
result = workflow.run(initial_state, config)
print(result)langgraph_workflow_agent/
βββ backend/
β βββ workflows/ # Workflow definitions
β β βββ base_workflow.py
β β βββ approval_workflow.py
β β βββ parallel_workflow.py
β β βββ iterative_workflow.py
β β βββ conditional_workflow.py
β βββ nodes/ # Workflow nodes
β β βββ base_node.py
β β βββ llm_node.py
β β βββ approval_node.py
β β βββ conditional_node.py
β βββ agents/ # Agent implementations
β β βββ base_agent.py
β βββ models.py # Data models
βββ frontend/
β βββ chainlit_app.py # Chainlit UI (main)
β βββ gradio_app.py # Gradio UI (legacy)
βββ .chainlit/
β βββ config.toml # Chainlit configuration
βββ data/
β βββ workflows/ # Workflow definitions
β βββ history/ # Execution history
βββ requirements.txt
I've included four different workflow types to show different patterns:
This one's great when you need a human in the loop. Maybe you're reviewing content before it goes live, or you want someone to approve code deployments. The workflow pauses, waits for approval, then continues based on the decision.
When you have multiple independent tasks, why wait? This workflow splits work across multiple agents that run simultaneously, then merges everything back together. Perfect for analyzing different data sources or processing multiple files at once.
Sometimes the first try isn't good enough. This workflow generates something, reviews it, and if it's not up to snuff, it refines and tries again. Great for content generation or code refinement where quality matters.
Not everything should follow the same path. This workflow analyzes the input, then routes to different processors based on what it finds. Think of it like a smart router that knows which agent should handle which type of task.
I chose LangGraph because it solves real problems you'll hit when building agent systems. It gives you:
- State Machines: You know exactly what state your workflow is in at any moment
- Cycles: Build workflows that can loop and refine themselves
- Conditional Edges: Route to different paths based on what's happening
- Checkpointing: Save progress and pick up where you left off if something breaks
- Visualization: See your workflow as a graph, which is super helpful for debugging
This makes it much easier to build workflows that actually work in production, not just demos.
- LangGraph: Workflow orchestration and state machines
- LangChain: Agent framework
- Gemini API: LLM access
- Chainlit: Chat-based UI framework with Mermaid support
- Pydantic: State validation
Let me break down how this all works:
Think of state as a backpack that travels through your workflow. It carries everything you need:
- Messages (the conversation history so far)
- Data (whatever your workflow is working with)
- Metadata (execution info, timestamps, etc.)
- Current step (where you are in the workflow)
- Errors (if something went wrong)
- Human input (when you're waiting on someone)
These are the actual work units. Each node does something specific - maybe it calls an LLM, maybe it runs a tool, maybe it just processes data. The key is that nodes update the state as they go, so the next node knows what happened before.
These connect your nodes. You can have simple sequential edges (A then B then C), conditional edges (if X then go to Y, else go to Z), or even parallel execution where multiple nodes run at the same time.
There's always more to build. Some things I'm thinking about:
- More workflow types for different use cases
- Better error recovery (smarter retries, fallback strategies)
- Workflow templates so you can quickly spin up common patterns
- Workflow versioning (track changes over time)
- Performance optimization (caching, batching, etc.)
- Cost tracking (know how much each workflow costs to run)
- A more advanced monitoring dashboard
- LangGraph: Workflow orchestration and state machines
- LangChain: Agent framework
- Chainlit: Chat-based UI with Mermaid support
- Pydantic: State validation
- Python 3.8+: Core language
- Workflow Automation: Complex multi-step workflows with conditional routing
- Approval Processes: Human-in-the-loop approval workflows
- Parallel Processing: Execute multiple tasks simultaneously
- Iterative Refinement: Self-improving workflows with loops
- State Management: Explicit workflow state tracking
If you build something cool with this or find a bug, I'd love to see it. Pull requests are always welcome!
MIT License - use it however you want.