Agentic Architecture Pattern: Market Based

Written by

in


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

Comments

Leave a Reply

Your email address will not be published. Required fields are marked *