AI向けプラットフォームエンジニアリング: 自律エージェントのためのインフラストラクチャ設計

[Agent Platform Engineering Architecture]

  User / Event Trigger
         │
         ▼
  ┌──────────────────────────────────────────────────────────┐
  │ Agent Gateway & Rate Limiter (Token Budget Enforcement)  │
  └────────────────────────────┬─────────────────────────────┘
                               │
         ▼                     ▼                      ▼
  ┌──────────────┐      ┌──────────────┐       ┌──────────────┐
  │ Agent Worker │      │ Agent Worker │       │ Agent Worker │
  │ (Reasoning)  │      │ (Reasoning)  │       │ (Reasoning)  │
  └──────┬───────┘      └──────┬───────┘       └──────┬───────┘
         │                     │                      │
         ├─────────────────────┼──────────────────────┤
         ▼                     ▼                      ▼
  ┌────────────────┐    ┌─────────────────┐    ┌──────────────────┐
  │ Isolated E2B / │    │ OpenTelemetry   │    │ Postgres / Redis │
  │ Firecracker VM │    │ Tracing & Evals │    │ Checkpoint Store │
  │ (Code Sandbox) │    │ (Audit Logging) │    │ (State Recovery) │
  └────────────────┘    └─────────────────┘    └──────────────────┘

# Production Agent Sandbox Invocation using Ephemeral MicroVMs
from e2b_code_interpreter import Sandbox

def execute_agent_code(python_code: str) -> dict:
    """Executes arbitrary agent-generated code in an isolated microVM sandbox."""
    # Instantiates a fresh Firecracker microVM in <200ms
    with Sandbox(template="python-data-science") as sandbox:
        try:
            execution = sandbox.run_code(
                python_code,
                timeout=30, # Hard ceiling prevents infinite CPU loops
            )
            return {
                "stdout": execution.logs.stdout,
                "stderr": execution.logs.stderr,
                "error": execution.error,
                "exit_code": 0 if not execution.error else 1,
            }
        except Exception as e:
            return {"error": f"Sandbox execution timeout: {str(e)}", "exit_code": -1}

from opentelemetry import trace

tracer = trace.get_tracer("ai.platform.agent", "1.0.0")

def execute_agent_step(agent_id: str, step_index: int, prompt: str, model: str):
    with tracer.start_as_current_span(f"agent.step.{step_index}") as span:
        span.set_attribute("gen_ai.system", "anthropic")
        span.set_attribute("gen_ai.request.model", model)
        span.set_attribute("agent.id", agent_id)
        span.set_attribute("agent.step_index", step_index)
        
        # Scrub PII before attaching to trace span
        sanitized_prompt = scrub_pii(prompt)
        span.set_attribute("gen_ai.prompt", sanitized_prompt)

        response = call_llm_with_tools(prompt)
        
        # Capture usage metrics for real-time cost attribution
        span.set_attribute("gen_ai.usage.prompt_tokens", response.usage.prompt_tokens)
        span.set_attribute("gen_ai.usage.completion_tokens", response.usage.completion_tokens)
        span.set_attribute("agent.tools_called", [t.name for t in response.tool_calls])
        
        return response

# Redis-Backed Sliding Window Token Circuit Breaker
import redis

r = redis.Redis.from_url(os.getenv("REDIS_URL"))

def check_token_quota(tenant_id: str, requested_tokens: int, max_hourly_budget: int = 250_000):
    key = f"quota:tokens:{tenant_id}"
    current_spent = r.incrby(key, requested_tokens)
    
    # Set 1-hour expiration on initial spend
    if current_spent == requested_tokens:
        r.expire(key, 3600)
        
    if current_spent > max_hourly_budget:
        # Trip the circuit breaker
        raise Exception(f"Hourly token quota exceeded for tenant {tenant_id}. Execution halted.")