| # 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) |
|
|
| ```python |
| 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:** |
| ```python |
| @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 |
| ```python |
| @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 |
| ```python |
| @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 |
| ```python |
| # 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 |
| ```python |
| 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):** |
| ```python |
| # 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:** |
| ```python |
| # 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: |
| ```python |
| 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:** |
| 1. **Parallel Tool Calls**: Use asyncio for mock_vendor_search + validate_budget |
| 2. **Model Selection**: Use claude-3-haiku for analysis (cheaper, faster) |
| 3. **Prompt Optimization**: Reduce tokens in analysis prompt |
| 4. **Caching**: Cache vendor list if query repeats |
| 5. **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 |
| ```python |
| 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 |
| ```python |
| @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 |
| ```python |
| 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 |
| ```python |
| # 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: |
|
|
| 1. **Modularity**: Independent agents, tools, and routing |
| 2. **Observability**: Full tracing via LangSmith |
| 3. **Reliability**: State persistence and recovery |
| 4. **Extensibility**: Easy to add agents, tools, and routing |
| 5. **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. |
|
|
