Multi-Agent Procurement System - Architecture Document
Executive Summary
This document provides a comprehensive architectural overview of the Multi-Agent Autonomous Procurement System. It is designed for architects, engineers, and technical leads evaluating this system for production deployment.
Key Design Principles:
- Modularity: Each agent has a single, well-defined responsibility
- Observability: Full tracing of all LLM calls and agent decisions
- Human-in-the-Loop: Critical decisions subject to human approval
- State Persistence: Complete workflow state maintained across interruptions
- Production-Ready: Follows enterprise patterns for error handling and monitoring
System Overview
High-Level Architecture
┌─────────────────────────────────────────────────────────────────────────┐
│ MULTI-AGENT PROCUREMENT SYSTEM │
└─────────────────────────────────────────────────────────────────────────┘
External Inputs Agents & Tools State Management
│ │ │
├─ Procurement Request ├─ ResearchAgent ───┐ ├─ ProcurementState
├─ Budget Limit │ └─ VendorSearch │ │ ├─ vendor_options
└─ (Human Approval) │ │ │ ├─ selected_vendor
├─ AnalysisAgent ─┐ │ │ ├─ analysis_approved
│ ├─ LLM Eval │ │ │ ├─ human_approved
│ └─ BudgetCheck │ │ │ └─ contract_draft
│ │ │ │
├─ ApprovalGate ───┼─┼──→ MemorySaver
│ └─ (Interrupts) │ │ │ (Checkpointer)
│ │ │ │
├─ LegalAgent ──┐ │ │ │
│ └─ ContractGen │ │ │
│ ▼ ▼ │
└─────────────────────── └
Command Routing
Data Flow
Initial Request
│
▼
research_node: "Find vendors"
│ Tool: mock_vendor_search
▼ Returns: vendor_options[]
analysis_node: "Evaluate & validate budget"
├─ LLM Call: Recommend vendor
├─ Tool: validate_budget
│ Returns: within_budget decision
│
├─ If within budget:
│ ▼
│ approval_gate: "Request human approval"
│ │ State Checkpoint Created
│ │ *** GRAPH PAUSES ***
│ │ (External: Human updates state)
│ ▼
│ approval_decision: "Route based on approval"
│ │
│ ├─ If approved:
│ │ ▼
│ │ legal_node: "Generate contract"
│ │ │ LLM Call: Create purchase agreement
│ │ ▼
│ │ contract_draft → END
│ │
│ └─ If rejected:
│ ▼
│ END
│
└─ If exceeds budget:
▼
END (Rejection)
Component Architecture
1. State Schema (TypedDict)
class ProcurementState(TypedDict):
"""Complete graph state with type safety."""
# Input Fields
procurement_request: str # What: Raw requirement
budget_limit: float # Budget: Maximum spend
# Agent Output Fields
vendor_options: list[dict] # Research output: candidates
selected_vendor: dict # Analysis output: chosen vendor
# Decision Fields
analysis_approved: bool # Analysis: Budget check result
human_approved: bool # Approval gate: Human decision
# Final Output
contract_draft: str # Legal output: Purchase agreement
# Audit
logs: list[str] # Complete decision history
Design Rationale:
- Strict Types: TypedDict enables IDE autocomplete and type checking
- Annotated[list, add]: Custom operator for list accumulation in logs
- Immutable Pattern: State never modified directly; new values via Command updates
- Audit Trail: Every decision logged with timestamp
2. Agents
ResearchAgent
Responsibility: Vendor Discovery
Input: procurement_request: str
Process:
1. Parse requirement
2. Call vendor search tool
3. Return structured vendor list
Output: vendor_options: list[dict]
Error Handling: Return empty list if search fails
Tool Integration:
@tool
def mock_vendor_search(query: str) -> list[dict]:
"""Mock vendor database search."""
# Returns: [{"vendor_id", "name", "price_per_month", "capabilities", ...}]
# Production: Replace with real APIs
def real_vendor_search(query: str) -> list[dict]:
return requests.get("https://api.vendors.com/search").json()
AnalysisAgent
Responsibility: Vendor Evaluation & Budget Validation
Input:
- vendor_options: list[dict]
- budget_limit: float
- procurement_request: str
Process:
1. Use LLM to analyze vendors
├─ Prompt: "Recommend best vendor considering price, reputation, capabilities"
├─ Temperature: 0.3 (low randomness for consistency)
└─ Max tokens: 2048
2. Parse LLM response for recommendation
3. Extract selected vendor from options
4. Call budget_validator tool
└─ Returns: {within_budget: bool, variance: float, status: "APPROVED|REJECTED"}
5. Route based on budget result
├─ Within budget → approval_gate
└─ Exceeded → END
Output:
- selected_vendor: dict
- analysis_approved: bool
Routing:
- goto="approval_gate" (if approved)
- goto=END (if rejected)
LLM Considerations:
- Model: claude-3-5-sonnet-20241022 (balanced performance/cost)
- Temperature: 0.3 (consistent financial analysis)
- System Prompt: Embedded in PromptTemplate (financial context)
- Token Limit: 2048 (sufficient for vendor analysis + response)
LegalAgent
Responsibility: Contract Generation
Input:
- selected_vendor: dict
- procurement_request: str
- budget_limit: float
Process:
1. Format vendor details for legal document
2. Call LLM with contract template
├─ Prompt: "Generate professional purchase contract in markdown"
├─ Includes: Parties, services, pricing, terms, warranties
└─ Output: Markdown document
3. Format and return contract
Output:
- contract_draft: str (markdown)
Routing:
- goto=END (always terminates)
Contract Template Includes:
- Header and document information
- Parties (Company & Vendor)
- Services/Goods description
- Pricing and payment terms
- Term and termination clauses
- Liabilities and warranties
- Confidentiality
- Signature block
3. Tools
Vendor Search Tool
@tool
def mock_vendor_search(query: str) -> list[dict]:
"""
Discover vendors matching procurement request.
Mock Database Structure:
- Categories: cloud_infrastructure, software_licensing
- Per vendor: id, name, price, capabilities, reputation, terms
Production Integration:
- Query vendor API
- Filter by region, compliance, integration support
- Return ranked by relevance/price
"""
Budget Validator Tool
@tool
def validate_budget(vendor_price: float, budget_limit: float) -> dict:
"""
Validate vendor pricing against budget.
Returns:
{
"within_budget": True/False,
"vendor_price": float,
"budget_limit": float,
"variance": float, # budget - price
"variance_percentage": float,
"status": "APPROVED" | "REJECTED"
}
Production Integration:
- Query ERP system for available budget
- Check cost center allocation
- Validate against PO limits
- Return approval status
"""
4. Graph Structure
Nodes
START
↓
research_node (ResearchAgent)
↓ Command.goto="analysis_node"
analysis_node (AnalysisAgent)
├─ Command.goto="approval_gate" (within budget)
└─ Command.goto=END (exceeds budget)
↓
approval_gate (Interruption)
↓ Command.goto="approval_decision"
approval_decision (Router)
├─ Route to "legal_node" (if human_approved)
└─ Route to END (if not approved)
↓
legal_node (LegalAgent)
↓ Command.goto=END
END
Edges
# Linear flow
graph.add_edge(START, "research_node")
# Command-based routing (agents return Command with explicit goto)
# research_node → analysis_node (via Command)
# analysis_node → {approval_gate OR END} (via Command)
# Explicit edge for approval flow
graph.add_edge("approval_gate", "approval_decision")
# Conditional routing
graph.add_conditional_edges(
"approval_decision",
lambda state: "legal_node" if state.get("human_approved") else END
)
# Terminal edge
graph.add_edge("legal_node", END)
Checkpointing Configuration
checkpointer = MemorySaver() # Development
# or
checkpointer = PostgresSaver() # Production
compiled_graph = graph.compile(
checkpointer=checkpointer,
interrupt_before=["legal_node"]
)
# Interruption Semantics:
# - Graph executes all nodes until reaching legal_node
# - Before executing legal_node, graph PAUSES
# - Control returns to application
# - State persisted to checkpointer
# - External system updates state.human_approved
# - graph.update_state(config, {"human_approved": True})
# - graph.stream(None, config) resumes execution
# - Continues with approval_decision and legal_node
Execution Model
Single Execution Cycle
Timeline: T=0ms to T=60ms
T+0ms Input: procurement_request, budget_limit
Initial State created
T+5ms research_node executes
│ Input: procurement_request
├─ Call: mock_vendor_search(request)
│ LangSmith: Log tool invocation
└─ Output: vendor_options populated, goto="analysis_node"
T+10ms analysis_node executes
│ Input: vendor_options, budget_limit, request
├─ Call: LLM with vendor analysis prompt
│ LangSmith: Log LLM call (tokens, latency, cost)
├─ Parse: Extract recommendation
├─ Call: validate_budget(vendor_price, budget)
│ LangSmith: Log tool invocation
└─ Decide:
within_budget → goto="approval_gate"
exceeded → goto=END
T+15ms (if within budget) approval_gate executes
│ Input: selected_vendor, logs
├─ Create log entry
└─ Return Command(goto="approval_decision")
T+16ms *** GRAPH PAUSES ***
State checkpointed to MemorySaver
Control returns to application
External: Human reviews state
External: Calls executor.approve_vendor(approval=True)
External: graph.update_state(config, {"human_approved": True})
T+60s+ External: graph.stream(None, config) called to resume
T+61ms approval_decision executes
│ Input: human_approved=True (updated state)
└─ Route: goto="legal_node"
T+70ms legal_node executes
│ Input: selected_vendor, request
├─ Call: LLM with contract generation prompt
│ LangSmith: Log LLM call
├─ Format: Markdown contract
└─ Output: contract_draft populated, goto=END
T+75ms Graph reaches END
Execution complete
Final state returned to application
State Checkpoint Mechanics
Checkpoint Creation (T+16ms):
# MemorySaver internal representation
checkpointer.storage = {
"procurement_20250628_145230": {
"values": {
# Full state at interruption point
"procurement_request": "...",
"vendor_options": [...],
"selected_vendor": {...},
"budget_limit": 5500.0,
"analysis_approved": True,
"human_approved": False, # Checkpoint created here
"contract_draft": "",
"logs": [...]
},
"next": "approval_decision", # Next node to execute
"task_id": "...",
"channel": "input"
}
}
State Retrieval & Update:
# Retrieve saved state
config = {"configurable": {"thread_id": "procurement_20250628_145230"}}
saved_state = graph.get_state(config) # Full state recovered
# Update with human decision
graph.update_state(config, {"human_approved": True})
# State in checkpoint is now:
# {..., "human_approved": True, ...}
# Resume from checkpoint
graph.stream(None, config) # Continues from next=approval_decision
Observability & Monitoring
LangSmith Integration
Automatic Tracing:
Every LLM call, tool invocation, and state transition is logged to LangSmith.
Trace Structure:
┌─ Procurement Workflow Run
│ ├─ research_node
│ │ └─ Tool: mock_vendor_search
│ │ ├─ Input: {"query": "cloud infrastructure"}
│ │ ├─ Output: [3 vendors]
│ │ └─ Latency: 2ms
│ │
│ ├─ analysis_node
│ │ ├─ LLM: ChatAnthropic
│ │ │ ├─ Prompt: "Analyze vendors..."
│ │ │ ├─ Response: "Based on analysis..."
│ │ │ ├─ Tokens: 342 in, 198 out
│ │ │ ├─ Cost: $0.0032
│ │ │ └─ Latency: 1.23s
│ │ │
│ │ └─ Tool: validate_budget
│ │ ├─ Input: {"vendor_price": 5000, "budget": 5500}
│ │ ├─ Output: {"within_budget": true, ...}
│ │ └─ Latency: 1ms
│ │
│ ├─ approval_gate
│ │ └─ Status: INTERRUPTED (awaiting human approval)
│ │
│ └─ legal_node
│ └─ LLM: ChatAnthropic
│ ├─ Prompt: "Generate contract..."
│ ├─ Response: "# PURCHASE AGREEMENT..."
│ ├─ Tokens: 287 in, 856 out
│ ├─ Cost: $0.0056
│ └─ Latency: 1.89s
Analytics Available:
- Traces: View every run with full execution tree
- Metrics: Token usage, latency by agent, cost per run
- Datasets: Log examples for continuous improvement
- Feedback: Mark runs as good/bad for analysis
- Comparison: Compare LLM models or prompts
Audit Trail
State logs maintain complete decision history:
state["logs"] = [
"Workflow initiated",
"[2025-06-28T14:52:30] ResearchNode: Analyzing request: We need enterprise cloud... | Found 3 vendors.",
"[2025-06-28T14:52:31] AnalysisNode: Validating 3 vendors against budget $5500.00 | Selected: CloudTech Solutions | Status: APPROVED",
"[2025-06-28T14:52:31] ApprovalGate: Awaiting human approval for CloudTech Solutions",
"[2025-06-28T14:52:35] LegalNode: Drafting contract for CloudTech Solutions | Contract generated successfully"
]
Compliance Features:
- Timestamped decision points
- Vendor selection rationale
- Budget validation evidence
- Human approval tracking
- Contract generation audit
Failure Modes & Recovery
Failure Mode: Research Tool Fails
research_node → Exception in mock_vendor_search
↓
Error: "No vendors found for query"
↓
Handling:
- Log error: {"error": "vendor_search failed", "request": "..."}
- Return empty vendor_options
- Next node (analysis_node) handles gracefully or routes to END
- User notified with audit trail
Failure Mode: LLM Call Fails
analysis_node → Timeout in ChatAnthropic.invoke()
↓
Error: "Model request timed out"
↓
Handling:
- LangChain builtin: Automatic retry with exponential backoff
- Max retries: 3
- If all retry: Exception bubbles up
- Can be caught in node: return Command(goto=END, update={"error": "..."})
Failure Mode: Graph Paused, Process Crashes
T+16ms: Graph paused, state checkpointed
T+30s: Process crashes
T+50s: New process starts with same thread_id
↓
Recovery:
1. Create new graph instance
2. Call: saved_state = graph.get_state(config)
→ Retrieves full state from checkpoint
3. Update: graph.update_state(config, {"human_approved": True})
→ Modifies checkpointed state
4. Resume: graph.stream(None, config)
→ Continues from approval_decision node
5. Final state: Returned with contract_draft populated
Zero-Loss Recovery: No lost state due to checkpointing.
Failure Mode: Budget Exceeds
analysis_node → validate_budget(6000, 5000) → {"within_budget": False}
↓
Handling:
- Set analysis_approved = False
- Route to END (not approval_gate)
- Contract never generated
- Audit log shows rejection reason
- User notified: "Budget exceeded by $1000"
Scalability Considerations
Current Architecture (MemorySaver)
- Single Process: State held in memory
- Single Machine: No inter-process state sharing
- Suitable For: POC, development, low-volume deployments
Production Architecture (PostgresSaver)
Recommended Setup:
Multiple Worker Processes ↔ PostgreSQL Database
(Shared State)
↓ Checkpointing
↓ Shared thread storage
Benefits:
- Horizontal Scaling: Add workers for concurrent workflows
- Persistence: State survives process restarts
- Distributed: Workers on different machines
- Concurrent Workflows: 1000s of independent procurement requests
Estimated Capacity:
- 1 PostgreSQL instance: 10,000+ concurrent workflow states
- 1 worker process: 10-50 concurrent workflows (I/O limited)
- Scale out: Add more workers as needed
Performance Optimization
Latency Breakdown (Current):
research_node: ~5ms (mock tool, instant)
analysis_node: ~1,200ms (1 LLM call)
├─ LLM latency: ~1,200ms
└─ Tool latency: ~1ms
legal_node: ~1,900ms (1 LLM call)
├─ LLM latency: ~1,900ms (longer response)
└─ Tool latency: ~0ms
Total: ~3.1 seconds execution
Human approval: 60+ seconds (external)
Optimization Opportunities:
- Parallel Tool Calls: Use asyncio for mock_vendor_search + validate_budget
- Model Selection: Use claude-3-haiku for analysis (cheaper, faster)
- Prompt Optimization: Reduce tokens in analysis prompt
- Caching: Cache vendor list if query repeats
- Batch Processing: Run multiple workflows with thread pools
Cost Optimization
Current Pricing (Claude 3.5 Sonnet):
- Input tokens: $0.003 per 1K
- Output tokens: $0.015 per 1K
Per-Workflow Cost:
- Analysis node: ~$0.0032 (342 in + 198 out)
- Legal node: ~0.0056 (287 in + 856 out)
- Total: ~$0.0088 per complete workflow
Cost Reduction:
- Switch to Haiku: ~90% cost reduction
- Optimize prompts: Reduce token usage
- Cache responses: Avoid repeated calls
- Batch similar requests: Amortize overhead
Security Architecture
Data Protection
- State Checkpoint: Stored in checkpointer (MemorySaver or PostgreSQL)
- Production: Encrypt at rest (PostgreSQL TLS)
- Secrets: Never logged or traced
- Audit: All decisions logged with timestamps
Access Control
- Graph Access: Gated by thread_id (implicit isolation)
- State Updates: Only via authenticated graph.update_state() calls
- Token Storage: Never persisted (LLM calls ephemeral)
- API Keys: Environment variables, not in code
Compliance
- Audit Trail: Complete decision history in state
- Timestamping: All actions timestamped
- Approval Tracking: human_approved flag with external update tracking
- Data Retention: Checkpoint retention policy configurable
Extension Points
Adding New Agents
def new_agent_node(state: ProcurementState) -> Command:
"""New specialized agent."""
# Process state
new_state = {"field": value}
return Command(update=new_state, goto="next_node")
# Add to graph
graph.add_node("new_agent", new_agent_node)
graph.add_edge("previous_node", "new_agent")
Adding New Tools
@tool
def new_tool(param: str) -> dict:
"""Specialized tool for domain."""
# Implement business logic
return {"result": "..."}
# Use in agent nodes
result = self.new_tool(state["field"])
Custom Routing
def custom_router(state: ProcurementState) -> str:
"""Complex routing logic."""
if state["field"] == "value":
return "node_a"
else:
return "node_b"
graph.add_conditional_edges("source", custom_router)
Integration with External Systems
# Send approval request
await send_slack_notification(state)
# Sync to document management
await upload_contract_to_docusign(state["contract_draft"])
# Log to compliance system
await log_to_audit_trail(state["logs"])
Conclusion
This architecture provides a production-grade foundation for multi-agent procurement systems. Key strengths:
- Modularity: Independent agents, tools, and routing
- Observability: Full tracing via LangSmith
- Reliability: State persistence and recovery
- Extensibility: Easy to add agents, tools, and routing
- Compliance: Complete audit trail and approval tracking
The system is suitable for immediate POC deployment and scales to production with PostgreSQL checkpointing and real tool integrations.