# Multi-Agent Procurement System: Setup & Deployment Guide ## Quick Start ### 1. Prerequisites - Python 3.10+ - pip or poetry for dependency management ### 2. Installation ```bash # Create virtual environment python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate # Install dependencies pip install -r requirements.txt ``` ### 3. Configure Environment Variables Create a `.env` file in the project root: ```bash # LangChain / Anthropic Configuration ANTHROPIC_API_KEY="sk-ant-xxx" # LangSmith Configuration (Optional but recommended) LANGSMITH_API_KEY="your_langsmith_key_here" LANGSMITH_PROJECT="procurement_system" LANGSMITH_ENDPOINT="https://api.smith.langchain.com" # Optional: Debugging DEBUG=False LOG_LEVEL="INFO" ``` ### 4. Run the System ```bash python procurement_system.py ``` --- ## Architecture Deep Dive ### Graph Flow Diagram ``` ┌─────────────────────────────────────────────────────────────────┐ │ PROCUREMENT WORKFLOW │ └─────────────────────────────────────────────────────────────────┘ START │ ▼ ┌──────────────┐ │ ResearchNode │ ◄─── Vendor Search Tool │ (Agent 1) │ └──────┬───────┘ │ Returns: vendor_options[] ▼ ┌──────────────┐ │ AnalysisNode │ ◄─── Budget Validator Tool │ (Agent 2) │ LLM Recommendation └──────┬───────┘ │ ┌─────────────┼──────────────┐ │ │ Within Exceeds Budget Budget │ │ ▼ ▼ ┌──────────────┐ ┌─────────┐ │ ApprovalGate │ │ END │ │ (Interrupts) │ │ REJECT │ └──────┬───────┘ └─────────┘ │ *** GRAPH PAUSES HERE *** External system updates: state.human_approved = True/False │ ▼ ┌────────────────────┐ │ ApprovalDecRouter │ └────────┬───────────┘ │ ┌───────┼────────┐ │ │ Approved Rejected │ │ ▼ ▼ ┌─────────┐ ┌────────┐ │LegalNode│ │ END │ │(Agent 3)│ │REJECT │ └────┬────┘ └────────┘ │ ▼ ┌───┐ │END│ (Contract generated) └───┘ ``` ### State Evolution **Initial State:** ```python { "procurement_request": "...", "vendor_options": [], "selected_vendor": {}, "budget_limit": 5500.0, "analysis_approved": False, "human_approved": False, "contract_draft": "", "logs": ["Workflow initiated"] } ``` **After ResearchNode:** ```python { # ... previous fields "vendor_options": [ { "vendor_id": "VENDOR_001", "name": "CloudTech Solutions", "price_per_month": 5000, "capabilities": [...], ... }, # ... more vendors ] } ``` **After AnalysisNode (Budget OK):** ```python { # ... previous fields "selected_vendor": { "vendor_id": "VENDOR_001", "name": "CloudTech Solutions", "price_per_month": 5000, ... }, "analysis_approved": True } ``` **After ApprovalGate (INTERRUPTED):** ``` Graph PAUSES at interrupt_before=["legal_node"] Human reviews state and updates via executor.approve_vendor(approval=True) ``` **After LegalNode:** ```python { # ... all previous fields "contract_draft": "# PURCHASE AGREEMENT\n\n## Parties\n...", "human_approved": True } ``` --- ## LangSmith Integration Setup ### Why LangSmith? For a multi-agent system, observability is critical: - **Trace each LLM call**: See what each agent is asking the LLM - **Track tool invocations**: Log vendor search and budget validation calls - **Monitor latency**: Identify bottlenecks in agent reasoning - **Debug state transitions**: Understand why the graph routed to a specific node - **Audit trail**: Full compliance logging for procurement decisions ### Step-by-Step LangSmith Setup #### 1. Create LangSmith Account - Visit: https://smith.langchain.com - Sign up with email or GitHub - Create organization (if needed) #### 2. Generate API Key - Navigate to Settings (bottom-left profile) - Click "API keys" - Create new API key - Copy the key (format: `lsk_xyz...`) #### 3. Configure Environment ```bash # Set environment variables export LANGSMITH_API_KEY="lsk_your_api_key_here" export LANGSMITH_PROJECT="procurement_system" # Verify configuration python -c "from langsmith import Client; c = Client(); print('✓ LangSmith connected')" ``` #### 4. Run System with Tracing ```bash python procurement_system.py ``` #### 5. View Traces in LangSmith Dashboard - Go to: https://smith.langchain.com/projects - Select project: "procurement_system" - Each workflow run appears as a trace tree showing: - **ResearchNode execution**: `mock_vendor_search` tool call - **AnalysisNode**: LLM invocation with vendor analysis prompt - **AnalysisNode**: `validate_budget` tool call - **Approval interruption**: State checkpoint - **LegalNode**: Contract generation LLM call ### LangSmith Dashboard Features #### Traces Tab ``` Run #1: Procurement for "Cloud Infrastructure" ├── ResearchNode │ └── Tool: mock_vendor_search │ └── Input: "cloud infrastructure" │ └── Output: [3 vendors found] │ ├── AnalysisNode │ ├── LLM Call: ChatAnthropic │ │ └── Tokens: 342 prompt, 198 completion │ │ └── Cost: $0.00X │ │ └── Latency: 1.23s │ └── Tool: validate_budget │ └── Status: APPROVED (within budget) │ └── ApprovalGate: INTERRUPTED ``` #### Analytics - **Agent performance**: Which agent takes longest? - **LLM costs**: Total spend on this project - **Tool success rate**: How often do tools succeed? - **Latency breakdown**: Time spent in each node #### Dataset Features (for continuous improvement) - Log LLM calls as training examples - Create feedback loops for agent improvements - Track prompt performance over time --- ## Production Deployment Considerations ### 1. Checkpointer Upgrade **Development (current):** ```python checkpointer = MemorySaver() # In-memory only, lost on restart ``` **Production:** ```python from langgraph.checkpoint.postgres import PostgresSaver checkpointer = PostgresSaver( conn_string="postgresql://user:password@localhost:5432/langgraph" ) ``` This persists all state to a database, allowing: - Recovery from crashes - Long-running workflows - Concurrent workflow management ### 2. LLM Configuration **Development:** ```python llm = ChatAnthropic(model="claude-3-5-sonnet-20241022", temperature=0.3) ``` **Production - Consider:** ```python llm = ChatAnthropic( model="claude-3-5-sonnet-20241022", temperature=0.1, # Lower for consistency max_tokens=4096, timeout=30, max_retries=3 # Retry failed calls ) ``` ### 3. Tool Enhancements **Mock tools (current):** ```python @tool def mock_vendor_search(query: str) -> list[dict]: # Returns hardcoded data ``` **Production - Real implementations:** ```python @tool def real_vendor_search(query: str) -> list[dict]: """Search live vendor database or API.""" response = requests.get( "https://vendor-api.example.com/search", params={"q": query} ) return response.json() @tool def real_budget_validator(vendor_id: str, budget: float) -> dict: """Query actual ERP system for budget allocation.""" erp = ERPClient(api_key=os.getenv("ERP_API_KEY")) allocation = erp.get_budget_allocation() return {"within_budget": allocation >= budget} ``` ### 4. Error Handling & Resilience ```python def research_node(state: ProcurementState) -> Command: """Enhanced with error handling.""" try: vendors = self.vendor_search_tool(state["procurement_request"]) if not vendors: raise ValueError("No vendors found for request") except Exception as e: new_log = f"ERROR in ResearchNode: {str(e)}" return Command( update={"logs": [new_log]}, goto=END # Graceful failure ) return Command( update={"vendor_options": vendors, "logs": [...]}, goto="analysis_node" ) ``` ### 5. Approval Gate Enhancement **Current (manual):** ```python state_update = {"human_approved": True} self.graph.update_state(config, state_update) ``` **Production (with workflow engine):** ```python # Integrate with Slack bot, email approval link, web dashboard async def wait_for_human_approval(thread_id: str, timeout_seconds=86400): """Poll approval status from database.""" while True: approval = db.get_approval_status(thread_id) if approval is not None: return approval await asyncio.sleep(60) ``` ### 6. Logging & Monitoring ```python import logging logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) logger = logging.getLogger(__name__) # In each node: logger.info(f"ResearchNode: Processing request: {request}") logger.info(f"AnalysisNode: Budget validation completed: {validation}") ``` ### 7. Async Execution ```python # For long-running workflows, use async execution from langgraph.types import StreamMode async def run_async(): config = {"configurable": {"thread_id": thread_id}} async for event in graph.astream(initial_state, config): logger.info(f"Event: {event}") ``` --- ## Troubleshooting ### Issue: Graph doesn't pause at approval_gate **Cause:** `interrupt_before=["legal_node"]` pauses *before* legal_node, not at approval_gate. **Solution:** Current design is correct. The graph pauses when it reaches the logical point before executing legal_node (which only runs if approved). ### Issue: State not persisting across restarts **Cause:** Using MemorySaver (in-memory only). **Solution:** Upgrade to PostgresSaver for production: ```bash pip install "langgraph[postgres]" ``` ### Issue: LLM calls not appearing in LangSmith **Cause:** LANGSMITH_API_KEY not set. **Solution:** ```bash export LANGSMITH_API_KEY="your_key" echo $LANGSMITH_API_KEY # Verify it's set python procurement_system.py ``` ### Issue: Vendor search returns empty list **Cause:** Query doesn't match mock data keywords. **Solution:** Use "cloud" or "software" in procurement request: ```python executor.start_workflow( procurement_request="We need enterprise cloud infrastructure", # ✓ Works budget_limit=5000 ) ``` --- ## Testing the System ### Unit Test Example ```python import pytest from procurement_system import ProcurementAgents, ProcurementState @pytest.fixture def llm(): from langchain_anthropic import ChatAnthropic return ChatAnthropic(model="claude-3-5-sonnet-20241022") def test_research_node(llm): agents = ProcurementAgents(llm) state = ProcurementState( procurement_request="cloud infrastructure", vendor_options=[], selected_vendor={}, budget_limit=5000, analysis_approved=False, human_approved=False, contract_draft="", logs=[] ) result = agents.research_node(state) assert len(result.update["vendor_options"]) > 0 assert result.goto == "analysis_node" def test_analysis_node_budget_exceeded(llm): agents = ProcurementAgents(llm) state = ProcurementState( procurement_request="", vendor_options=[{"name": "Expensive", "price_per_month": 10000}], selected_vendor={}, budget_limit=1000, analysis_approved=False, human_approved=False, contract_draft="", logs=[] ) result = agents.analysis_node(state) assert result.goto == END assert result.update["analysis_approved"] == False # Run tests pytest test_procurement.py -v ``` --- ## Next Steps 1. **Test locally** with mock tools 2. **Integrate real tools**: Connect to actual vendor APIs 3. **Deploy checkpointer**: Use PostgreSQL for state persistence 4. **Add web UI**: Build frontend for approval gate (Streamlit, FastAPI + React) 5. **Monitor in LangSmith**: Track performance and costs 6. **Add compliance logging**: Audit trail for procurement regulations --- ## References - [LangGraph Documentation](https://langchain-ai.github.io/langgraph/) - [LangSmith Tracing Guide](https://docs.smith.langchain.com/) - [Claude Models API](https://docs.anthropic.com/en/api/getting-started) - [LangChain Tools](https://python.langchain.com/docs/modules/tools/)