"""
Pattern 36: Market-Based Pattern
Agents operate in an economic market, trading resources and services based on supply/demand.
Uses pricing mechanisms, bidding, and market equilibrium for resource allocation.
"""
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
class TransactionType(Enum):
BUY = "buy"
SELL = "sell"
@dataclass
class MarketAgent:
"""Agent participating in market economy."""
name: str
role: str
currency: float = 1000.0 # Starting budget
services: List[str] = field(default_factory=list)
transactions: List[Dict[str, Any]] = field(default_factory=list)
@dataclass
class ServiceListing:
"""Service offered in marketplace."""
provider: str
service: str
price: float
quality: float # 0.0-1.0
available: bool = True
class MarketBasedPattern:
"""
Market-Based Pattern - Economic coordination mechanism.
Features:
- Supply and demand dynamics
- Price-based resource allocation
- Competitive bidding
- Market equilibrium
- Transaction history
"""
def __init__(self, model_name: str = "gpt-4"):
self.llm = ChatOpenAI(model=model_name, temperature=0.7)
self.agents: Dict[str, MarketAgent] = {}
self.marketplace: List[ServiceListing] = []
self.transactions: List[Dict[str, Any]] = []
def add_agent(self, agent: MarketAgent):
"""Add agent to market."""
self.agents[agent.name] = agent
print(f"✓ Agent {agent.name} joined market with {agent.currency} currency")
def list_service(self, provider_name: str, service: str, price: float, quality: float):
"""Agent lists service in marketplace."""
if provider_name in self.agents:
listing = ServiceListing(provider_name, service, price, quality)
self.marketplace.append(listing)
self.agents[provider_name].services.append(service)
print(f"✓ {provider_name} listed: {service} @ {price} (quality: {quality})")
return listing
return None
def find_best_offer(self, service_type: str, max_price: float) -> Optional[ServiceListing]:
"""Find best service offer based on price-quality ratio."""
candidates = [l for l in self.marketplace
if service_type.lower() in l.service.lower()
and l.price Dict[str, Any]:
"""Execute market transaction."""
if buyer_name not in self.agents:
return {"success": False, "error": "Buyer not found"}
buyer = self.agents[buyer_name]
# Find best offer
best_offer = self.find_best_offer(service_type, max_price)
if not best_offer:
return {"success": False, "error": "No suitable offers"}
if buyer.currency 2: # High demand
listing.price *= 1.2
print(f"↑ Price increase: {listing.service} → {listing.price:.2f}")
def get_market_stats(self) -> Dict[str, Any]:
"""Get market statistics."""
return {
"total_agents": len(self.agents),
"active_listings": sum(1 for l in self.marketplace if l.available),
"total_transactions": len(self.transactions),
"total_volume": sum(tx['price'] for tx in self.transactions),
"average_price": sum(tx['price'] for tx in self.transactions) / len(self.transactions) if self.transactions else 0
}
def run(self, scenario: str) -> Dict[str, Any]:
"""Execute market scenario."""
return {
"pattern": "Market-Based",
"scenario": scenario,
**self.get_market_stats()
}
# Example usage
if __name__ == "__main__":
print("=== Market-Based Pattern Demo ===\n")
market = MarketBasedPattern()
# Add agents
print("--- Agents Joining Market ---")
agents = [
MarketAgent("Alice", "Developer", 1000, []),
MarketAgent("Bob", "Designer", 1200, []),
MarketAgent("Carol", "Tester", 800, []),
]
for agent in agents:
market.add_agent(agent)
# List services
print("\n--- Service Listings ---")
market.list_service("Alice", "API Development", 300, 0.9)
market.list_service("Alice", "Code Review", 100, 0.85)
market.list_service("Bob", "UI Design", 250, 0.95)
market.list_service("Bob", "Logo Design", 150, 0.8)
market.list_service("Carol", "Automated Testing", 200, 0.88)
# Execute transactions
print("\n--- Market Transactions ---")
tx1 = market.execute_transaction("Bob", "Code Review", 150)
if tx1['success']:
print(f"✓ {tx1['buyer']} bought {tx1['service']} from {tx1['seller']} for {tx1['price']}")
tx2 = market.execute_transaction("Carol", "UI Design", 300)
if tx2['success']:
print(f"✓ {tx2['buyer']} bought {tx2['service']} from {tx2['seller']} for {tx2['price']}")
tx3 = market.execute_transaction("Alice", "Automated Testing", 250)
if tx3['success']:
print(f"✓ {tx3['buyer']} bought {tx3['service']} from {tx3['seller']} for {tx3['price']}")
# Show agent balances
print("\n--- Agent Balances ---")
for name, agent in market.agents.items():
print(f"{name}: {agent.currency:.2f} currency ({len(agent.transactions)} transactions)")
# Market statistics
print("\n--- Market Statistics ---")
stats = market.get_market_stats()
print(f"Total Agents: {stats['total_agents']}")
print(f"Active Listings: {stats['active_listings']}")
print(f"Total Transactions: {stats['total_transactions']}")
print(f"Total Volume: {stats['total_volume']:.2f}")
print(f"Average Price: {stats['average_price']:.2f}")
# Price adjustments
print("\n--- Price Adjustments (Supply/Demand) ---")
market.adjust_prices()
result = market.run("Service marketplace")
print(f"\n✓ {result['pattern']} pattern executed")
result = pattern.run("Test input")
print(f"Result: {result}")
Agentic Architecture Pattern: Market Based
Pattern 36: 36 Market Based – Line-by-Line Explanation
Overview
Pattern 36: Market-Based Pattern
Agents operate in an economic market, trading resources and services based on supply/demand. Uses pricing mechanisms, bidding, and market equilibrium for resource allocation.
Code Structure Analysis
File Information
- Pattern Number: 36
- Pattern Name: 36 Market Based
- File Path:
patterns\organizational\36_market_based.py - Total Lines: 220
Detailed Code Explanation
Module Docstring
"""
Pattern 36: Market-Based Pattern
Agents operate in an economic market, trading resources and services based on supply/demand.
Uses pricing mechanisms, bidding, and market equilibrium for resource allocation.
"""
Explanation: This module docstring describes the purpose and key features of the 36 Market Based pattern.
Imports Section
from typing import Dict, Any, List, Optional
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime: Date and time utilities
– from enum import Enum: Enumeration support for fixed sets of values
– from langchain_core.prompts import ChatPromptTemplate: LangChain framework integration
– from langchain_openai import ChatOpenAI: LangChain framework integration
Enumerations
TransactionType (Line 16)
class TransactionType(Enum):
BUY
SELL
Purpose: No enum docstring.
Values:
– BUY: Represents buy
– SELL: Represents sell
Classes
TransactionType (Line 16)
Description: No class docstring.
MarketAgent (Line 22)
Description: Agent participating in market economy.
ServiceListing (Line 32)
Description: Service offered in marketplace.
MarketBasedPattern (Line 41)
Description: Market-Based Pattern – Economic coordination mechanism.
Features: – Supply and demand dynamics – Price-based resource allocation – Competitive bidding – Market equilibrium – Transaction history
Methods:
__init__ (Line 53)
def __init__(self, model_name):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
– model_name: LLM model to use (e.g., “gpt-4”)
add_agent (Line 59)
def add_agent(self, agent):
"""Add agent to market."""
Purpose: Add agent to market.
Parameters:
– self: Instance reference
– agent: Parameter for agent
list_service (Line 64)
def list_service(self, provider_name, service, price, quality):
"""Agent lists service in marketplace."""
Purpose: Agent lists service in marketplace.
Parameters:
– self: Instance reference
– provider_name: Parameter for provider_name
– service: Parameter for service
– price: Parameter for price
– quality: Parameter for quality
find_best_offer (Line 74)
def find_best_offer(self, service_type, max_price):
"""Find best service offer based on price-quality ratio."""
Purpose: Find best service offer based on price-quality ratio.
Parameters:
– self: Instance reference
– service_type: Parameter for service_type
– max_price: Parameter for max_price
execute_transaction (Line 87)
def execute_transaction(self, buyer_name, service_type, max_price):
"""Execute market transaction."""
Purpose: Execute market transaction.
Parameters:
– self: Instance reference
– buyer_name: Parameter for buyer_name
– service_type: Parameter for service_type
– max_price: Parameter for max_price
adjust_prices (Line 125)
def adjust_prices(self):
"""Market adjusts prices based on demand."""
Purpose: Market adjusts prices based on demand.
Parameters:
– self: Instance reference
get_market_stats (Line 140)
def get_market_stats(self):
"""Get market statistics."""
Purpose: Get market statistics.
Parameters:
– self: Instance reference
run (Line 150)
def run(self, scenario):
"""Execute market scenario."""
Purpose: Execute market scenario.
Parameters:
– self: Instance reference
– scenario: Parameter for scenario
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Initialize the pattern
pattern = TransactionType(name="example", model_name="gpt-4")
# Configure and execute
result = pattern.execute(data={"key": "value"})
# Check results
print(result)
Key Concepts
- LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
- Workflow Orchestration: LangGraph manages execution flow through nodes
- State Management: TypedDict defines workflow state structure
- Error Handling: Comprehensive error handling and fallback mechanisms
- Extensibility: Designed for easy customization and extension
Extension Points
This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities
Leave a Reply