---
title: Multi-Agent Autonomous Procurement System
emoji: 🏢
colorFrom: blue
colorTo: purple
sdk: docker
sdk_version: "3.11"
python_version: "3.11"
app_file: app.py
pinned: false
---

# Multi-Agent Autonomous Procurement System

A production-ready implementation of a multi-agent procurement orchestration system using **LangGraph**, **LangChain**, and **LangSmith**. This system demonstrates enterprise-grade agent design patterns including human-in-the-loop interruptions, stateful graph execution, and comprehensive observability.

## 📋 Overview

This POC implements a complete procurement workflow where autonomous agents collaborate to:

1. **Discover vendors** through intelligent search
2. **Analyze pricing** against budget constraints
3. **Pause for human approval** with persistent state
4. **Generate legal contracts** post-approval

The system showcases modern LLM orchestration patterns suitable for production deployment.

```
START → ResearchAgent → AnalysisAgent → [Budget Check]
                                            ├─ OK: HumanApproval (INTERRUPTS) → LegalAgent → END
                                            └─ Exceeded: END (Reject)
```

## 🎯 Key Features

### Core Architecture
- **Multi-Agent Orchestration**: 3 specialized agents with clear responsibilities
- **Stateful Execution**: Persistent state management via `MemorySaver` (upgradable to PostgreSQL)
- **Human-in-the-Loop**: Graph pauses at interruption points for external decision-making
- **Production Patterns**: Uses modern LangGraph Command-based routing

### Observability
- **Full LangSmith Tracing**: Every LLM call, tool invocation, and state transition logged
- **Audit Trail**: Complete decision history in graph state
- **Performance Metrics**: Latency, token usage, and cost tracking

### Enterprise-Ready
- **Error Handling**: Graceful failure modes and retry logic
- **Type Safety**: Strict TypedDict schema for all state
- **Modular Design**: Agents, tools, and graph cleanly separated
- **Async Support**: Non-blocking workflow execution

## 📁 Project Structure

```
.
├── procurement_system.py           # Main implementation (production-grade)
├── advanced_examples.py            # Additional patterns and integrations
├── setup_and_deployment.md         # Detailed setup & production guide
├── EXECUTION_WALKTHROUGH.md        # Step-by-step execution trace
├── requirements.txt                # Python dependencies
└── README.md                       # This file
```

## 🚀 Quick Start

### 1. Setup Environment

```bash
# Clone or download project
cd /path/to/procurement_system

# Create virtual environment
python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate

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

### 2. Configure API Keys

Create a `.env` file in the project root:

```bash
# Required: Anthropic API Key
ANTHROPIC_API_KEY="sk-ant-xxx"

# Optional but recommended: LangSmith for observability
LANGSMITH_API_KEY="lsk_xxx"
LANGSMITH_PROJECT="procurement_system"
```

### 3. Run the System

```bash
# Execute complete workflow with human approval
python procurement_system.py

# Output will show:
# - Stage 1: Vendor research and analysis → Graph pauses
# - Stage 2: Human reviews state at interruption
# - Stage 3: State updated with approval → Graph resumes
# - Stage 4: Contract generated and printed
```

## 📖 Usage Examples

### Basic Workflow Execution

```python
from langchain_anthropic import ChatAnthropic
from procurement_system import build_procurement_graph, ProcurementWorkflowExecutor

# Initialize
llm = ChatAnthropic(model="claude-3-5-sonnet-20241022")
graph, memory, agents = build_procurement_graph(llm)
executor = ProcurementWorkflowExecutor(graph, memory)

# Start workflow (executes until interruption point)
state = executor.start_workflow(
    procurement_request="Enterprise cloud infrastructure with auto-scaling",
    budget_limit=5500.0
)

# Inspect state for human review
current_state = executor.inspect_state()

# Human approves vendor
final_state = executor.approve_vendor(approval=True)

# Access generated contract
print(final_state["contract_draft"])
```

### Programmatic State Inspection

```python
from advanced_examples import WorkflowStateInspector

inspector = WorkflowStateInspector(graph, thread_id="procurement_20250628_145230")

# Get current state
state = inspector.get_current_state()

# Analyze decision trail
decisions = inspector.get_decision_trail()

# Generate audit report
report = inspector.generate_audit_report()
print(report)
```

### Vendor Comparison

```python
from advanced_examples import VendorComparison

# Get vendor comparison matrix
comparison = VendorComparison.build_comparison_matrix(state, llm)
print(comparison)
```

### Metrics & Analytics

```python
from advanced_examples import ProcurementMetrics

metrics = ProcurementMetrics()

# Calculate KPIs
cycle_time = metrics.calculate_procurement_cycle_time(state)
budget_utilization = metrics.calculate_budget_utilization(state)
vendor_variance = metrics.calculate_vendor_variance(state)

print(f"Cycle Time: {cycle_time}s")
print(f"Budget Utilization: {budget_utilization}%")
print(f"Vendor Variance: {vendor_variance['variance_percentage']:.2f}%")
```

## 🔍 System Components

### State Schema

```python
class ProcurementState(TypedDict):
    procurement_request: str          # What's being procured
    vendor_options: list[dict]        # Candidate vendors from research
    selected_vendor: dict             # Chosen vendor after analysis
    budget_limit: float               # Maximum spending limit
    analysis_approved: bool           # Budget validation result
    human_approved: bool              # Human decision (set at interruption)
    contract_draft: str               # Generated legal document
    logs: list[str]                   # Audit trail
```

### Agent Nodes

#### ResearchNode (Agent 1)
**Responsibility**: Vendor discovery
- **Input**: Procurement request
- **Tool**: `mock_vendor_search` (mock for POC, integrate real APIs in production)
- **Output**: `vendor_options` list with pricing and capabilities
- **Logic**: Searches for vendors matching requirement

#### AnalysisNode (Agent 2)
**Responsibility**: Financial validation and recommendation
- **Input**: Vendor list, budget limit
- **LLM Call**: Claude analyzes vendors for best fit
- **Tools**: 
  - `validate_budget`: Checks if vendor price ≤ budget
- **Routing**:
  - Budget OK → Send to approval gate
  - Budget exceeded → Terminate (rejection)
- **Output**: `selected_vendor`, `analysis_approved` flag

#### ApprovalGate (Human-in-the-Loop)
**Responsibility**: Interruption point for external approval
- **Behavior**: Graph pauses before legal_node
- **State Persisted**: Via `MemorySaver` checkpoint
- **External Update**: System updates `human_approved` flag
- **Resumption**: Graph continues with updated state
- **Use Case**: Slack notification, email approval, web dashboard

#### LegalNode (Agent 3)
**Responsibility**: Contract generation
- **Input**: Approved vendor details, procurement request
- **LLM Call**: Claude generates markdown purchase agreement
- **Output**: `contract_draft` with complete legal terms
- **Template**: Includes parties, services, pricing, terms, warranties

### Tools

#### `mock_vendor_search(query: str) → list[dict]`
Mock vendor discovery returning 3-5 vendors with details.

In production, replace with:
```python
@tool
def real_vendor_search(query: str) -> list[dict]:
    """Query vendor APIs or internal procurement catalog."""
    response = requests.get("https://vendor-api.com/search", params={"q": query})
    return response.json()
```

#### `validate_budget(vendor_price: float, budget_limit: float) → dict`
Budget validation returning:
```python
{
    "within_budget": True/False,
    "variance": 500.00,
    "variance_percentage": 9.09,
    "status": "APPROVED" or "REJECTED"
}
```

## 📊 LangSmith Integration

### Setup

1. **Create Account**: https://smith.langchain.com
2. **Get API Key**: Settings → API Keys
3. **Set Environment**:
   ```bash
   export LANGSMITH_API_KEY="lsk_xxx"
   export LANGSMITH_PROJECT="procurement_system"
   ```
4. **Run System**: All LLM calls automatically traced

### Observability Features

**Trace Tree View**:
```
Procurement Workflow Run #1
├─ research_node (5ms)
│  └─ Tool: mock_vendor_search
├─ analysis_node (10ms)
│  ├─ LLM: ChatAnthropic (1.23s, 342 tokens in, 198 out)
│  └─ Tool: validate_budget
├─ approval_gate (1ms) [INTERRUPTED]
│  └─ *** PAUSED FOR HUMAN APPROVAL ***
└─ legal_node (8ms)
   └─ LLM: ChatAnthropic (1.89s, 287 tokens in, 856 out)
```

**Analytics Dashboard**:
- Total cost tracking
- LLM model performance
- Tool invocation success rates
- Latency by agent
- Token usage trends

## 🔄 Workflow Execution Timeline

### Complete Execution Flow

```
T+0ms:   Graph starts with initial_state
T+5ms:   ResearchNode executes → vendor_options populated
T+10ms:  AnalysisNode executes → LLM recommendation + budget check
         If within budget: goto approval_gate
         If exceeded: goto END
T+15ms:  ApprovalGate processes → Graph PAUSES (checkpoint created)

*** EXTERNAL SYSTEM ***
Human reviews state at interruption
Updates state: human_approved = True

T+60s:   Graph resumes from checkpoint
T+61ms:  ApprovalDecision routes → goto legal_node
T+70ms:  LegalNode executes → contract_draft generated
T+75ms:  Graph reaches END

Total:   ~75ms execution + human approval time
```

## 🛠️ Production Deployment

### Checkpointer Upgrade

**Development (current)**:
```python
from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()  # In-memory only
```

**Production**:
```python
from langgraph.checkpoint.postgres import PostgresSaver
checkpointer = PostgresSaver(
    conn_string="postgresql://user:pwd@localhost/langgraph"
)
# Provides: Persistence, recovery, multi-worker support
```

### Real Tool Integration

```python
# Replace mock tools with production implementations
@tool
def real_vendor_search(query: str) -> list[dict]:
    """Search live vendor database."""
    db = VendorDatabase()
    return db.search(query)

@tool
def real_budget_validator(vendor_id: str, budget: float) -> dict:
    """Query actual ERP system."""
    erp = ERPClient()
    return erp.check_budget_allocation(vendor_id, budget)
```

### Error Handling

```python
def research_node(state: ProcurementState) -> Command:
    try:
        vendors = self.vendor_search_tool(state["procurement_request"])
        if not vendors:
            raise ValueError("No vendors found")
    except Exception as e:
        logger.error(f"Research failed: {str(e)}")
        return Command(update={"logs": [f"ERROR: {str(e)}"]}, goto=END)
    
    return Command(update={"vendor_options": vendors, ...}, goto="analysis_node")
```

### Async Execution

```python
# For non-blocking workflows in web applications
async def run_procurement_async():
    async for event in graph.astream(initial_state, config):
        logger.info(f"Event: {event}")
```

## 🧪 Testing

### Unit Tests

```bash
# Install test dependencies
pip install pytest pytest-asyncio

# Run tests
pytest tests/ -v

# Example test:
def test_analysis_node_budget_exceeded():
    state = ProcurementState(
        procurement_request="",
        vendor_options=[{"price_per_month": 10000}],
        budget_limit=1000,
        ...
    )
    result = agents.analysis_node(state)
    assert result.goto == END
```

### Integration Tests

```python
# Test complete workflow
def test_complete_workflow():
    executor = ProcurementWorkflowExecutor(graph, memory)
    state = executor.start_workflow(
        procurement_request="Cloud infrastructure",
        budget_limit=5000
    )
    assert state["selected_vendor"]["name"] is not None
    assert state["analysis_approved"] == True
```

## 📈 Monitoring & Metrics

### Key Metrics to Track

```python
from advanced_examples import ProcurementMetrics

metrics = ProcurementMetrics()

# Cycle time: Time from request to contract generation
cycle_time = metrics.calculate_procurement_cycle_time(state)

# Budget utilization: Percentage of budget used
utilization = metrics.calculate_budget_utilization(state)

# Vendor variance: Price spread across options
variance = metrics.calculate_vendor_variance(state)

# Dashboard queries (with PostgreSQL)
# - Weekly procurement throughput
# - Avg approval time
# - Vendor selection patterns
# - Budget compliance rate
```

## 🔐 Security Considerations

### Secrets Management

```bash
# Never commit API keys
echo ".env" >> .gitignore

# Use environment variables or secret manager
export ANTHROPIC_API_KEY=$(aws secretsmanager get-secret-value --secret-id anthropic-key | jq -r .SecretString)
```

### State Security

- Sensitive data (contracts, pricing) stored in checkpointer
- **Production**: Use encrypted PostgreSQL with TLS
- **Access Control**: RBAC for state updates
- **Audit Logging**: All approvals timestamped and logged

### LLM Safety

```python
llm = ChatAnthropic(
    model="claude-3-5-sonnet-20241022",
    temperature=0.1,  # Lower for consistent decisions
    max_tokens=4096,
    timeout=30,
    max_retries=3
)
```

## 🎓 Learning Resources

### Included Documentation
- **EXECUTION_WALKTHROUGH.md**: Detailed trace of every step
- **setup_and_deployment.md**: Production deployment guide
- **advanced_examples.py**: Patterns for escalation, async, metrics

### External Resources
- [LangGraph Documentation](https://langchain-ai.github.io/langgraph/)
- [LangSmith Tracing Guide](https://docs.smith.langchain.com/)
- [Claude API Documentation](https://docs.anthropic.com/)
- [LangChain Tools Reference](https://python.langchain.com/docs/modules/tools/)

## 🤝 Contributing

This is a POC for demonstration. To extend:

1. **Add More Agents**: Extend `ProcurementAgents` class
2. **Custom Tools**: Create `@tool` decorated functions
3. **New Routing Logic**: Add conditional edges to graph
4. **Monitoring**: Integrate with Datadog, New Relic, etc.

## 📝 License

This implementation is provided as-is for educational and demonstration purposes.

## ❓ FAQ

### Q: How does the graph pause for human approval?
A: Uses `interrupt_before=["legal_node"]` in graph compilation. When the graph reaches this point, it pauses and returns control to the application. External systems update the state with human decision, then the graph resumes.

### Q: Can I run multiple workflows concurrently?
A: Yes! Each workflow gets a unique `thread_id`. The checkpointer (especially PostgreSQL) supports concurrent execution.

### Q: What's the difference between this and a simple chain?
A: This uses a **StateGraph** with persistent state, routing decisions, and interruptions. A chain would be linear without branching or pauses.

### Q: How do I integrate with real vendor APIs?
A: Replace `mock_vendor_search` with a real `@tool` that calls your vendor API. Same for budget validation.

### Q: Can I deploy this as a REST API?
A: Yes! Wrap the executor in FastAPI:
```python
@app.post("/procurement/start")
async def start_procurement(request: dict):
    executor = ProcurementWorkflowExecutor(graph, memory)
    state = executor.start_workflow(...)
    return {"thread_id": executor.thread_id, "state": state}
```

## 🚨 Troubleshooting

### Graph doesn't pause at approval point
- Verify `interrupt_before=["legal_node"]` in graph compilation
- Check that state reaches approval_gate node
- Review logs for routing decisions

### LangSmith traces not appearing
- Verify `LANGSMITH_API_KEY` is set: `echo $LANGSMITH_API_KEY`
- Check project name: `LANGSMITH_PROJECT="procurement_system"`
- Ensure `langsmith` package installed: `pip list | grep langsmith`

### State not persisting across sessions
- MemorySaver only persists while process is running
- For persistence: upgrade to PostgresSaver
- Verify checkpointer is passed to `graph.compile(checkpointer=...)`

### LLM calls failing
- Check `ANTHROPIC_API_KEY` is valid
- Verify model name: `claude-3-5-sonnet-20241022`
- Check rate limits: https://console.anthropic.com

---

**Ready to build production-grade multi-agent systems?** Start with `python procurement_system.py` and explore the execution walkthrough!