Supply Chain

Global event monitoring with risk analysis, mitigation strategies, and persistence

Multi-agent supply chain risk management: event monitoring → impact analysis → strategy generation. Includes web search for real-time news, SQLite persistence, and API deployment.

Setup

# Create environment
python3 -m venv venv && source venv/bin/activate

# Install packages
pip install praisonaiagents praisonai

# Set API keys
export OPENAI_API_KEY="your-key"
export TAVILY_API_KEY="your-key"  # For news monitoring

Create Sample Data

# Create supplier database
cat > suppliers.json << 'EOF'
[
  {"id": "SUP001", "name": "Taiwan Chips Co", "region": "Asia", "products": ["semiconductors"], "risk_score": 7},
  {"id": "SUP002", "name": "German Motors", "region": "Europe", "products": ["engines", "parts"], "risk_score": 3},
  {"id": "SUP003", "name": "Brazil Raw Materials", "region": "Americas", "products": ["steel", "aluminum"], "risk_score": 5}
]
EOF

Run: Python Code

from praisonaiagents import Agent, AgentTeam, Task, tool
from pydantic import BaseModel
from typing import List
import json

# Structured output
class RiskAssessment(BaseModel):
    event_type: str
    affected_suppliers: List[str]
    risk_level: str  # critical, high, medium, low
    estimated_delay_days: int
    mitigation_strategies: List[str]

# Database persistence is configured via memory={} parameter

# Tools
@tool
def search_global_news(region: str) -> str:
    """Search for supply chain disruption news in a region."""
    from tavily import TavilyClient
    import os
    client = TavilyClient(api_key=os.getenv("TAVILY_API_KEY"))
    results = client.search(f"supply chain disruption {region} 2024", max_results=3)
    return json.dumps([{"title": r["title"], "content": r["content"][:200]} for r in results["results"]])

@tool
def check_supplier_risk(supplier_id: str) -> str:
    """Check risk profile for a supplier."""
    with open("suppliers.json") as f:
        suppliers = json.load(f)
    for s in suppliers:
        if s["id"] == supplier_id:
            return json.dumps(s)
    return json.dumps({"error": "Supplier not found"})

@tool
def get_backup_suppliers(product: str) -> str:
    """Find backup suppliers for a product."""
    backups = {
        "semiconductors": ["Korea Chips Ltd", "Japan Semi Inc"],
        "engines": ["US Motors Corp", "Mexico Auto Parts"],
        "steel": ["India Steel Co", "Australia Mining Ltd"]
    }
    return json.dumps({"product": product, "backups": backups.get(product, [])})

# Agents
monitor = Agent(
    name="EventMonitor",
    instructions="Monitor global news for supply chain disruptions. Use search_global_news tool.",
    tools=[search_global_news],
    memory={
        "db": "sqlite:///supply_chain.db",
        "session_id": "supply-chain-monitor"
    }
)

analyzer = Agent(
    name="RiskAnalyzer",
    instructions="Analyze supplier risk and impact. Use check_supplier_risk tool.",
    tools=[check_supplier_risk]
)

strategist = Agent(
    name="Strategist",
    instructions="Generate mitigation strategies. Use get_backup_suppliers for alternatives.",
    tools=[get_backup_suppliers]
)

# Tasks
monitor_task = Task(
    description="Search for supply chain news in {region}",
    agent=monitor,
    expected_output="List of relevant disruption events"
)

analyze_task = Task(
    description="Analyze risk for suppliers in affected region",
    agent=analyzer,
    expected_output="Risk assessment for each supplier"
)

strategy_task = Task(
    description="Generate mitigation strategies for identified risks",
    agent=strategist,
    expected_output="Structured risk assessment with strategies",
    output_pydantic=RiskAssessment
)

# Run
agents = AgentTeam(agents=[monitor, analyzer, strategist], tasks=[monitor_task, analyze_task, strategy_task])
result = agents.start(region="Asia")
print(result)

Run: CLI

# Monitor specific region
praisonai "Check supply chain risks in Asia" --web-search --verbose

# With persistence
praisonai "Analyze Taiwan semiconductor supply risk" --memory --user-id supply_team

# Full analysis
praisonai "Monitor global supply chain and generate mitigation report" --web-search --telemetry

Run: agents.yaml

Create agents.yaml:

framework: praisonai
topic: "supply chain risk management"
roles:
  monitor:
    role: Global Event Monitor
    goal: Track supply chain disruptions worldwide
    backstory: Expert at monitoring geopolitical and natural events
    tools:
      - tavily_search
    tasks:
      monitor_events:
        description: |
          Search for supply chain disruptions:
          - Natural disasters
          - Political unrest
          - Trade policy changes
          - Pandemic impacts
        expected_output: List of current disruption events
        
  analyzer:
    role: Risk Analyst
    goal: Assess supply chain impact
    backstory: Expert at quantifying supply chain risks
    tasks:
      analyze_risk:
        description: |
          For each event, assess:
          - Affected suppliers and products
          - Estimated delay (days)
          - Cost impact (low/medium/high/critical)
          - Risk score (1-10)
        expected_output: Risk assessment matrix
        
  strategist:
    role: Mitigation Strategist
    goal: Develop risk mitigation plans
    backstory: Expert at supply chain optimization
    tasks:
      generate_strategies:
        description: |
          For high-risk items:
          - Identify backup suppliers
          - Calculate buffer stock needs
          - Recommend logistics alternatives
        expected_output: Actionable mitigation plan

Run:

praisonai agents.yaml --verbose

Monitor & Verify

# View monitoring history
praisonai --history 10 --user-id supply_team

# Check metrics
praisonai --metrics

# Export report
praisonai --save supply_chain_report

Serve API

from praisonaiagents import Agent, tool
import json

@tool
def quick_risk_check(region: str, product: str) -> str:
    """Quick supply chain risk check."""
    risk_matrix = {
        ("Asia", "semiconductors"): {"risk": "high", "delay_days": 30},
        ("Europe", "engines"): {"risk": "medium", "delay_days": 14},
        ("Americas", "steel"): {"risk": "low", "delay_days": 7}
    }
    result = risk_matrix.get((region, product), {"risk": "unknown", "delay_days": 0})
    return json.dumps(result)

agent = Agent(
    name="SupplyChainAPI",
    instructions="Assess supply chain risks for regions and products.",
    tools=[quick_risk_check]
)

agent.launch(path="/supply-risk", port=8000)

Test:

curl -X POST http://localhost:8000/supply-risk \
  -H "Content-Type: application/json" \
  -d '{"message": "Check risk for semiconductors from Asia"}'

Cleanup

rm -f supply_chain.db suppliers.json
deactivate

Features Demonstrated

Feature	Implementation
Multi-agent	Monitor → Analyzer → Strategist
Web Search	Tavily for real-time news
Structured Output	Pydantic RiskAssessment
Custom Tools	Supplier lookup, backup finder
DB Persistence	SQLite via db()
CLI	--web-search for live data
YAML Config	3-agent risk pipeline
API Endpoint	agent.launch()