README.md

metadata

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

# 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:

# 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

# 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

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

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

from advanced_examples import VendorComparison

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

Metrics & Analytics

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

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:

@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:

{
    "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:

   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):

from langgraph.checkpoint.memory import MemorySaver
checkpointer = MemorySaver()  # In-memory only

Production:

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

Real Tool Integration

# 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

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

# 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

# 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

# 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

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

# 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

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
• LangSmith Tracing Guide
• Claude API Documentation
• LangChain Tools Reference

🤝 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:

@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!