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