Khi AI Agent Tự "Đàm Phán" Với Nhau: Hiệu Ứng Domino A2A Và Thách Thức Kiểm Soát Hành Vi Tự Phát Trong Doanh Nghiệp

When AI Agents "Negotiate" with Each Other: The A2A Domino Effect and the Challenge of Controlling Emergent Behaviors in Enterprises

1. Chẩn đoán rủi ro: Khi nền kinh tế tác tử (Agentic Economy) mất kiểm soát Sự chuyển dịch từ các mô hình ngôn ngữ lớn (LLM) tĩnh sang các...

1. Risk Diagnosis: When the Agentic Economy Spirals Out of Control The shift from static Large Language Models (LLMs) to autonomous AI Agents is reshaping enterprise operations. Instead of operating...

import time

class A2ACircuitBreaker:
    def __init__(self, max_calls=10, budget_limit=5.0):
        self.max_calls = max_calls          # Giới hạn số lượt tương tác cho một phiên đàm phán
        self.budget_limit = budget_limit    # Giới hạn ngân sách (USD) cho mỗi phiên
        self.call_count = 0
        self.accumulated_cost = 0.0
        self.is_tripped = False

    def validate_transaction(self, estimated_cost):
        if self.is_tripped:
            raise Exception("Circuit Breaker đang kích hoạt. Giao dịch bị chặn.")
        
        self.call_count += 1
        self.accumulated_cost += estimated_cost

        # Kiểm tra điều kiện ngắt mạch
        if self.call_count > self.max_calls:
            self.is_tripped = True
            raise Exception(f"Ngắt mạch khẩn cấp: Vượt quá số lượt tương tác tối đa ({self.max_calls}). Nghi ngờ vòng lặp vô hạn.")
        
        if self.accumulated_cost > self.budget_limit:
            self.is_tripped = True
            raise Exception(f"Ngắt mạch khẩn cấp: Vượt quá giới hạn ngân sách ({self.budget_limit} USD).")
        
        return True

    def reset(self):
        self.call_count = 0
        self.accumulated_cost = 0.0
        self.is_tripped = False

# Giả lập quá trình đàm phán giữa 2 Agent
breaker = A2ACircuitBreaker(max_calls=5, budget_limit=2.50)

try:
    for turn in range(1, 10):
        # Giả lập chi phí API cho mỗi lượt đàm phán là 0.6 USD
        cost_per_turn = 0.60
        breaker.validate_transaction(cost_per_turn)
        print(f"Lượt {turn}: Đàm phán thành công. Chi phí tích lũy: {breaker.accumulated_cost:.2f} USD")
        time.sleep(0.1)
except Exception as e:
    print(f"CẢNH BÁO HỆ THỐNG: {str(e)}")

import time

class A2ACircuitBreaker:
    def __init__(self, max_calls=10, budget_limit=5.0):
        self.max_calls = max_calls          # Limit interaction rounds for a single negotiation session
        self.budget_limit = budget_limit    # Budget limit (USD) per session
        self.call_count = 0
        self.accumulated_cost = 0.0
        self.is_tripped = False

    def validate_transaction(self, estimated_cost):
        if self.is_tripped:
            raise Exception("Circuit Breaker is active. Transaction blocked.")
        
        self.call_count += 1
        self.accumulated_cost += estimated_cost

        # Check circuit breaker conditions
        if self.call_count > self.max_calls:
            self.is_tripped = True
            raise Exception(f"Emergency Trip: Maximum interaction rounds ({self.max_calls}) exceeded. Suspected infinite loop.")
        
        if self.accumulated_cost > self.budget_limit:
            self.is_tripped = True
            raise Exception(f"Emergency Trip: Budget limit ({self.budget_limit} USD) exceeded.")
        
        return True

    def reset(self):
        self.call_count = 0
        self.accumulated_cost = 0.0
        self.is_tripped = False

# Simulating a negotiation process between 2 Agents
breaker = A2ACircuitBreaker(max_calls=5, budget_limit=2.50)

try:
    for turn in range(1, 10):
        # Simulate API cost per turn of 0.6 USD
        cost_per_turn = 0.60
        breaker.validate_transaction(cost_per_turn)
        print(f"Round {turn}: Negotiation successful. Accumulated cost: {breaker.accumulated_cost:.2f} USD")
        time.sleep(0.1)
except Exception as e:
    print(f"SYSTEM WARNING: {str(e)}")