Introduction
In this AI agent design, we focus on an Inventory Management & Restocking Agent. This agent is designed to help businesses optimize their inventory control by tracking stock levels, identifying products that need restocking, and triggering automatic orders based on predefined thresholds. The agent ensures that restocking decisions are in line with company policies, and it automatically generates the necessary purchase orders. It never claims success until the action is completed and receipts are provided, ensuring that every step in the process is tracked for audit purposes.
The Use Case
Inventory management is a crucial process for any retail or warehouse-based business. The AI agent checks stock levels of various products, compares them against predefined minimum stock thresholds, and automatically triggers purchase orders for restocking. It can integrate with existing ERP (Enterprise Resource Planning) systems to ensure that inventory is consistently managed and replenished in a timely manner, preventing stockouts and ensuring that customers’ demands are met.
Prompt Contract (agent interface)
# file: contracts/inventory_management_v1.yaml
role: "InventoryManagementAgent"
scope: >
Monitor inventory levels and restock when necessary. Trigger purchase orders when thresholds are breached.
Ask once for missing fields (product_id, current_stock, minimum_stock_level, reorder_quantity).
Propose tool calls; never assert success without a receipt.
output:
type: object
required: [summary, decision, inventory_details, citations, next_steps, tool_proposals]
properties:
summary: {type: string, maxWords: 100}
decision: {type: string, enum: ["approve","reject","need_approval","need_more_info"]}
inventory_details:
type: object
required: [product_id, current_stock, minimum_stock_level, reorder_quantity, restock_order_created]
properties:
product_id: {type: string}
current_stock: {type: integer}
minimum_stock_level: {type: integer}
reorder_quantity: {type: integer}
restock_order_created: {type: boolean}
citations: {type: array, items: {type: string}}
next_steps: {type: array, items: {type: string}, maxItems: 5}
tool_proposals:
type: array
items:
type: object
required: [name, args, preconditions, idempotency_key]
properties:
name: {type: string, enum: [CheckStockLevel, CheckRestockingPolicy, CreateRestockOrder, RequestApproval]}
args: {type: object}
preconditions: {type: string}
idempotency_key: {type: string}
policy_id: "inventory_policy.v4"
citation_rule: "1–2 minimal-span claim_ids per factual sentence"
decoding:
narrative: {top_p: 0.92, temperature: 0.72, stop: ["\n\n## "]}
bullets: {top_p: 0.82, temperature: 0.45}
Example claims (context provided to the model)
[
{"claim_id":"policy:inventory:min_stock_level","text":"Products must have at least 10 units in stock at all times.",
"effective_date":"2025-01-01","source_id":"doc:inventory_policy_v4","span":"at least 10 units in stock"},
{"claim_id":"policy:inventory:reorder_quantity","text":"Restock orders should always be in multiples of 100 units.",
"effective_date":"2025-01-01","source_id":"doc:inventory_policy_v4","span":"multiples of 100 units"},
{"claim_id":"policy:inventory:restock_threshold","text":"Restock threshold for popular products is 20 units.",
"effective_date":"2025-01-01","source_id":"doc:inventory_policy_v4","span":"Restock threshold for popular products is 20 units"}
]
Tool Interfaces (typed, with receipts)
# tools.py
from pydantic import BaseModel
from typing import Optional, List, Dict
from datetime import datetime
class CheckStockLevelArgs(BaseModel):
product_id: str
current_stock: int
minimum_stock_level: int
class CheckRestockingPolicyArgs(BaseModel):
product_id: str
current_stock: int
reorder_quantity: int
class CreateRestockOrderArgs(BaseModel):
product_id: str
reorder_quantity: int
supplier: str
class RequestApprovalArgs(BaseModel):
product_id: str
reorder_quantity: int
approver_id: str
class ToolReceipt(BaseModel):
tool: str
ok: bool
ref: str
message: str = ""
data: Optional[Dict] = None
# adapters.py (demo logic)
from tools import *
from datetime import datetime
MIN_STOCK_LEVEL = 10 # units
RESTOCK_THRESHOLD = 20 # units
REORDER_MULTIPLE = 100 # units
PRODUCTS = {
"prod-001": {"stock": 15, "min_stock_level": 10, "reorder_quantity": 100, "supplier": "SupplierA"},
"prod-002": {"stock": 5, "min_stock_level": 10, "reorder_quantity": 100, "supplier": "SupplierB"}
}
def check_stock_level(a: CheckStockLevelArgs) -> ToolReceipt:
product = PRODUCTS.get(a.product_id)
if not product:
return ToolReceipt(tool="CheckStockLevel", ok=False, ref="product-not-found", message="Product not found")
stock_needed = max(0, a.minimum_stock_level - a.current_stock)
return ToolReceipt(tool="CheckStockLevel", ok=True, ref=f"stock-{a.product_id}",
message="Stock checked", data={"stock_needed": stock_needed})
def check_restocking_policy(a: CheckRestockingPolicyArgs) -> ToolReceipt:
if a.current_stock >= a.reorder_quantity:
return ToolReceipt(tool="CheckRestockingPolicy", ok=True, ref="policy-ok", message="Reorder quantity meets policy")
return ToolReceipt(tool="CheckRestockingPolicy", ok=False, ref="policy-reorder-issue",
message="Reorder quantity is not met")
def create_restock_order(a: CreateRestockOrderArgs) -> ToolReceipt:
return ToolReceipt(tool="CreateRestockOrder", ok=True, ref=f"restock-{a.product_id}",
message="Restock order created", data={"reorder_quantity": a.reorder_quantity})
def request_approval(a: RequestApprovalArgs) -> ToolReceipt:
return ToolReceipt(tool="RequestApproval", ok=True, ref=f"approval-{a.product_id}",
message=f"Approval requested for {a.reorder_quantity} units.", data={"approver_id": a.approver_id})
Agent Loop (proposal → verification → execution → receipts)
# agent_inventory.py
import uuid, json
from datetime import datetime
from typing import Any, Dict, List
from tools import *
from adapters import *
ALLOWED_TOOLS = {"CheckStockLevel", "CheckRestockingPolicy", "CreateRestockOrder", "RequestApproval"}
def new_idem() -> str:
return f"idem-{uuid.uuid4()}"
def verify_proposal(p: Dict[str, Any]) -> str:
required = {"name","args","preconditions","idempotency_key"}
if not required.issubset(p): return "Missing proposal fields"
if p["name"] not in ALLOWED_TOOLS: return "Tool not allowed"
return ""
def execute(p: Dict[str, Any]) -> ToolReceipt:
n, a = p["name"], p["args"]
if n == "CheckStockLevel": return check_stock_level(CheckStockLevelArgs(**a))
if n == "CheckRestockingPolicy": return check_restocking_policy(CheckRestockingPolicyArgs(**a))
if n == "CreateRestockOrder": return create_restock_order(CreateRestockOrderArgs(**a))
if n == "RequestApproval": return request_approval(RequestApprovalArgs(**a))
return ToolReceipt(tool=n, ok=False, ref="none", message="Unknown tool")
# --- Model shim returning a plan per contract (replace with your LLM call) ---
def call_model(contract_yaml: str, claims: List[Dict[str,Any]], product_data: Dict[str,Any]) -> Dict[str,Any]:
stock_needed = max(0, product_data["min_stock_level"] - product_data["stock"])
decision = "approve" if stock_needed == 0 else "request_approval"
return {
"summary": f"Product {product_data['product_id']} reviewed for restocking.",
"decision": decision,
"inventory_details": {
"product_id": product_data["product_id"],
"current_stock": product_data["stock"],
"minimum_stock_level": product_data["min_stock_level"],
"reorder_quantity": product_data["reorder_quantity"],
"restock_order_created": False
},
"citations": ["policy:inventory:min_stock_level", "policy:inventory:reorder_quantity", "policy:inventory:restock_threshold"],
"next_steps": ["Check stock level", "Check restocking policy", "Request approval if needed", "Create restock order"],
"tool_proposals": [
{"name":"CheckStockLevel",
"args":{"product_id":product_data["product_id"],"current_stock":product_data["stock"],
"minimum_stock_level":product_data["min_stock_level"]},
"preconditions":"Check stock level and calculate restocking need.","idempotency_key": new_idem()},
{"name":"CheckRestockingPolicy",
"args":{"product_id":product_data["product_id"],"current_stock":product_data["stock"],
"reorder_quantity":product_data["reorder_quantity"]},
"preconditions":"Ensure policy compliance for reorder quantity.","idempotency_key": new_idem()},
{"name":"RequestApproval",
"args":{"product_id":product_data["product_id"],"reorder_quantity":product_data["reorder_quantity"],
"approver_id":"M002"},
"preconditions":"Request approval for the restocking order.","idempotency_key": new_idem()},
{"name":"CreateRestockOrder",
"args":{"product_id":product_data["product_id"],"reorder_quantity":product_data["reorder_quantity"],
"supplier":product_data["supplier"]},
"preconditions":"Create restocking order.","idempotency_key": new_idem()}
]
}
def render_response(model_json: Dict[str,Any], receipts: List[ToolReceipt]) -> str:
idx = {r.tool:r for r in receipts}
lines = [model_json["summary"], ""]
lines.append(f"Decision: {model_json['decision']}")
lines.append(f"Inventory Details: {model_json['inventory_details']}")
lines.append("")
lines.append("Next steps:")
for s in model_json["next_steps"]:
lines.append(f"• {s}")
if idx.get("CreateRestockOrder") and idx["CreateRestockOrder"].ok:
lines.append(f"\nRestock order created: {idx['CreateRestockOrder'].message}")
if idx.get("RequestApproval") and idx["RequestApproval"].ok:
lines.append(f"Approval requested: {idx['RequestApproval'].message}")
lines.append("\nCitations: " + ", ".join(model_json["citations"]))
return "\n".join(lines)
def handle(product_data: Dict[str,Any]) -> str:
contract = open("contracts/inventory_management_v1.yaml").read()
claims: List[Dict[str,Any]] = [] # load real claims
plan = call_model(contract, claims, product_data)
receipts: List[ToolReceipt] = []
for prop in plan["tool_proposals"]:
reason = verify_proposal(prop)
if reason:
receipts.append(ToolReceipt(tool=prop["name"], ok=False, ref="blocked", message=reason)); continue
rec = execute(prop)
receipts.append(rec)
if not rec.ok and prop["name"] in {"CreateRestockOrder"}:
break
return render_response(plan, receipts)
if __name__ == "__main__":
example_product_data = {
"product_id":"prod-123",
"stock": 50,
"min_stock_level": 10,
"reorder_quantity": 100,
"supplier": "SupplierA"
}
print(handle(example_product_data))
The Prompt You’d Send to the Model (concise and testable)
System:
You are InventoryManagementAgent. Follow the contract:
- Ask once if product_id, current_stock, minimum_stock_level, or reorder_quantity is missing.
- Cite 1–2 claim_ids per factual sentence using provided claims.
- Propose tools; never assert success without a receipt.
- Output JSON with keys: summary, decision, inventory_details, citations[], next_steps[], tool_proposals[].
Claims (eligible only):
[ ... JSON array of inventory policy claims like above ... ]
User:
Please process this inventory for restocking:
{"product_id":"prod-123","stock":50,"min_stock_level":10,"reorder_quantity":100,"supplier":"SupplierA"}
How to adapt quickly
Integrate the stock level checks, restocking policy validation, and order creation functions with your existing ERP or inventory management system. Add idempotency and transactional integrity for stock operations, ensuring no accidental double orders. Load claims from company policies regarding inventory levels, reordering policies, and supplier relationships. Deploy the contract, policy bundle, and decoder settings behind a feature flag for testing, canary deployments, and rollback support.