LangChain と Antigravity で組むAIパイプライン — 権限境界・失敗回復・コストから逆算する設計

# pipelines/memory/memory_factory.py
from langchain.memory import (
    ConversationBufferMemory,
    ConversationSummaryMemory,
    ConversationEntityMemory,
    ConversationBufferWindowMemory,
)
from enum import Enum
 
class MemoryStrategy(Enum):
    BUFFER = "buffer"              # すべてのメッセージ
    BUFFER_WINDOW = "buffer_window"  # 最後のN個のメッセージ
    SUMMARY = "summary"            # 要約された履歴
    ENTITY = "entity"              # エンティティトラッキング
 
class MemoryFactory:
    """戦略と制約に基づいてメモリインスタンスを作成"""
 
    @staticmethod
    def create_memory(
        strategy: MemoryStrategy,
        max_tokens: int = 4000,
        llm=None,
        **kwargs
    ):
        if strategy == MemoryStrategy.BUFFER:
            # シンプル、低レイテンシ、無制限の成長
            return ConversationBufferMemory(
                memory_key="chat_history",
                return_messages=True,
                **kwargs
            )
 
        elif strategy == MemoryStrategy.BUFFER_WINDOW:
            # 固定メモリフットプリント、最近のコンテキストを保持
            k = kwargs.get("k", 5)  # 保持するメッセージ数
            return ConversationBufferWindowMemory(
                memory_key="chat_history",
                k=k,
                return_messages=True,
                **kwargs
            )
 
        elif strategy == MemoryStrategy.SUMMARY:
            # 要約を通じた圧縮（LLMが必須）
            return ConversationSummaryMemory(
                llm=llm,
                memory_key="chat_history",
                return_messages=True,
                **kwargs
            )
 
        elif strategy == MemoryStrategy.ENTITY:
            # エンティティとその属性を追跡
            return ConversationEntityMemory(
                llm=llm,
                memory_key="chat_history",
                entity_extraction_prompt=kwargs.get("entity_extraction_prompt"),
                **kwargs
            )
 
# トークン制限での使用
class TokenAwareMemory:
    """厳格なトークン予算でメモリを管理"""
 
    def __init__(self, max_tokens: int = 4000, llm=None):
        self.max_tokens = max_tokens
        self.llm = llm
        self._memory = None
        self._total_tokens = 0
        self._strategy = None
 
    def add_message(self, role: str, content: str):
        """メッセージを追加し、必要に応じて戦略を切り替え"""
        # トークンを推定（おおよそ：英語では1トークン≈4文字）
        estimated_tokens = len(content) // 4
 
        if self._total_tokens + estimated_tokens > self.max_tokens * 0.8:
            # 制限に近づいている、要約戦略に切り替え
            if self._strategy != MemoryStrategy.SUMMARY:
                self._switch_to_summary()
 
        self._memory.save_context(
            {"input": role},
            {"output": content}
        )
        self._total_tokens += estimated_tokens
 
    def _switch_to_summary(self):
        """バッファから要約メモリに移行"""
        # 現在のメモリからメッセージを抽出
        messages = self._memory.chat_memory.messages if hasattr(
            self._memory, 'chat_memory'
        ) else []
 
        # 要約メモリを作成
        self._memory = MemoryFactory.create_memory(
            MemoryStrategy.SUMMARY,
            llm=self.llm,
            max_tokens=self.max_tokens
        )
        self._strategy = MemoryStrategy.SUMMARY
 
        # メッセージを再追加（要約をトリガー）
        for msg in messages:
            self.add_message(msg.type, msg.content)

# pipelines/routing/router.py
from dataclasses import dataclass
from typing import Callable, List
from enum import Enum
 
@dataclass
class Route:
    """パイプラインルートを表示"""
    name: str
    condition: Callable[[str], bool]
    chain: Any  # 実行するLangChainチェーン
    description: str
    priority: int = 0
 
class DynamicRouter(PipelineComponent):
    """リクエストを特化したチェーンにルーティング"""
 
    def __init__(self, routes: List[Route], config: Dict[str, Any] = None):
        super().__init__("DynamicRouter", config)
        # 優先度でソート（高い順）
        self.routes = sorted(routes, key=lambda r: r.priority, reverse=True)
        self.fallback_chain = config.get("fallback_chain")
 
    async def execute(self, context: StageContext) -> StageContext:
        selected_route = None
 
        # 各ルートの条件を評価
        for route in self.routes:
            try:
                if route.condition(context.input_text):
                    selected_route = route
                    break
            except Exception as e:
                context.metadata.setdefault("routing_errors", []).append({
                    "route": route.name,
                    "error": str(e)
                })
                continue
 
        # ルートがマッチしなかった場合フォールバックを使用
        if not selected_route:
            if not self.fallback_chain:
                raise ValueError("ルートがマッチしもフォールバックチェーンが設定されていません")
            selected_route = Route(
                name="fallback",
                condition=lambda x: True,
                chain=self.fallback_chain,
                description="フォールバックチェーン"
            )
 
        context.metadata["selected_route"] = selected_route.name
        context.metadata["chain_description"] = selected_route.description
 
        return context

# pipelines/resilience/retry_policy.py
import asyncio
from typing import Callable, Any, TypeVar
from enum import Enum
 
T = TypeVar('T')
 
class RetryStrategy(Enum):
    EXPONENTIAL = "exponential"
    LINEAR = "linear"
    FIXED = "fixed"
 
class ResilientComponent(PipelineComponent):
    """再試行ロジック付きコンポーネントの基底クラス"""
 
    def __init__(
        self,
        name: str,
        config: Dict[str, Any] = None,
        max_retries: int = 3,
        strategy: RetryStrategy = RetryStrategy.EXPONENTIAL,
        base_delay: float = 1.0,
        max_delay: float = 30.0
    ):
        super().__init__(name, config)
        self.max_retries = max_retries
        self.strategy = strategy
        self.base_delay = base_delay
        self.max_delay = max_delay
 
    async def execute_with_retry(self, func: Callable[..., Any], *args, **kwargs):
        """自動再試行とバックオフで実行"""
        last_exception = None
 
        for attempt in range(self.max_retries + 1):
            try:
                return await func(*args, **kwargs)
            except Exception as e:
                last_exception = e
 
                if attempt == self.max_retries:
                    break
 
                # 遅延を計算
                delay = self._calculate_delay(attempt)
 
                # 再試行をログ
                self.execution_metrics["errors"].append({
                    "attempt": attempt + 1,
                    "error": str(e),
                    "retry_delay": delay
                })
 
                await asyncio.sleep(delay)
 
        raise last_exception
 
    def _calculate_delay(self, attempt: int) -> float:
        """再試行戦略に基づいて遅延を計算"""
        if self.strategy == RetryStrategy.EXPONENTIAL:
            delay = self.base_delay * (2 ** attempt)
        elif self.strategy == RetryStrategy.LINEAR:
            delay = self.base_delay * (attempt + 1)
        else:  # FIXED
            delay = self.base_delay
 
        return min(delay, self.max_delay)
 
class CircuitBreaker:
    """繰り返しの再試行からカスケード障害を防止"""
 
    class State(Enum):
        CLOSED = "closed"      # 通常動作
        OPEN = "open"          # 失敗、リクエストを拒否
        HALF_OPEN = "half_open"  # 復旧をテスト中
 
    def __init__(
        self,
        failure_threshold: int = 5,
        recovery_timeout: int = 60,
        success_threshold: int = 2
    ):
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout
        self.success_threshold = success_threshold
        self.state = self.State.CLOSED
        self.failure_count = 0
        self.success_count = 0
        self.last_failure_time = None
 
    async def execute(self, func: Callable, *args, **kwargs):
        """サーキットブレーカー保護で関数を実行"""
        if self.state == self.State.OPEN:
            if self._should_attempt_recovery():
                self.state = self.State.HALF_OPEN
                self.success_count = 0
            else:
                raise RuntimeError("サーキットブレーカーはOPEN です")
 
        try:
            result = await func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise
 
    def _on_success(self):
        self.failure_count = 0
        if self.state == self.State.HALF_OPEN:
            self.success_count += 1
            if self.success_count >= self.success_threshold:
                self.state = self.State.CLOSED
 
    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = self.State.OPEN
 
    def _should_attempt_recovery(self) -> bool:
        if not self.last_failure_time:
            return True
        elapsed = time.time() - self.last_failure_time
        return elapsed >= self.recovery_timeout

# deploy/antigravity_pipeline.py
from antigravity import Agent, Pipeline, Config
import structlog
 
logger = structlog.get_logger()
 
class AIAssistantPipeline(Agent):
    """Antigravityによって管理される本番AIパイプライン"""
 
    def __init__(self, config: PipelineConfig):
        super().__init__(name="ai-assistant-pipeline", version="1.0.0")
        self.config = config
 
        # コンポーネントを初期化
        self.validator = InputValidator({"min_length": 10})
        self.retriever = ContextRetriever(vector_store, {"top_k": 5})
        self.router = DynamicRouter(routes, {"fallback_chain": default_chain})
        self.circuit_breaker = CircuitBreaker(failure_threshold=5)
 
        # メトリクストラッキング
        self.total_requests = 0
        self.successful_requests = 0
        self.failed_requests = 0
 
    async def execute(self, user_input: str, session_id: str) -> Dict[str, Any]:
        """完全なパイプラインを実行"""
        self.total_requests += 1
 
        context = StageContext(
            input_text=user_input,
            metadata={"session_id": session_id, "timestamp": datetime.utcnow().isoformat()}
        )
 
        try:
            # パイプラインステージ
            context = await self.validator.run(context)
            context = await self.retriever.run(context)
            context = await self.router.run(context)
 
            # サーキットブレーカーで選択されたチェーンを実行
            response = await self.circuit_breaker.execute(
                self._execute_chain,
                context
            )
 
            self.successful_requests += 1
            return {
                "status": "success",
                "response": response,
                "metadata": context.metadata,
                "execution_time": sum(context.execution_time.values())
            }
 
        except Exception as e:
            self.failed_requests += 1
            logger.error(
                "pipeline_execution_failed",
                error=str(e),
                session_id=session_id,
                component=getattr(e, 'component', 'unknown')
            )
            return {
                "status": "error",
                "error": str(e),
                "metadata": context.metadata
            }

# pipeline/permissions.py
from enum import IntEnum
from dataclasses import dataclass
from typing import Callable, Optional, Any
import logging
 
logger = logging.getLogger(__name__)
 
 
class Tier(IntEnum):
    OBSERVE = 1        # 副作用なし
    REVERSIBLE = 2     # 取り消せる書き込み
    IRREVERSIBLE = 3   # 取り消せない
 
 
class PermissionDenied(RuntimeError):
    """承認されなかった操作"""
 
 
@dataclass
class PermissionPolicy:
    """どの階層まで自動実行を許すか"""
    auto_max: Tier = Tier.REVERSIBLE
    approve: Optional[Callable[[str, dict], bool]] = None
 
    def check(self, tool_name: str, tier: Tier, payload: dict) -> None:
        if tier <= self.auto_max:
            logger.info("auto-approved", extra={"tool": tool_name, "tier": int(tier)})
            return
        if self.approve is None:
            raise PermissionDenied(f"{tool_name}: L{int(tier)} に承認ハンドラがありません")
        if not self.approve(tool_name, payload):
            raise PermissionDenied(f"{tool_name}: 人間の承認が得られませんでした")
        logger.warning("human-approved", extra={"tool": tool_name, "tier": int(tier)})
 
 
def guarded(tier: Tier, policy: PermissionPolicy):
    """ツール関数に権限階層を結びつけるデコレータ"""
    def wrap(fn: Callable[..., Any]) -> Callable[..., Any]:
        def inner(**kwargs: Any) -> Any:
            policy.check(fn.__name__, tier, kwargs)
            return fn(**kwargs)
        inner.__name__ = fn.__name__
        inner.__doc__ = fn.__doc__
        inner.permission_tier = tier  # type: ignore[attr-defined]
        return inner
    return wrap

# pipeline/tools.py
from langchain_core.tools import StructuredTool
from pipeline.permissions import Tier, PermissionPolicy, guarded
 
policy = PermissionPolicy(
    auto_max=Tier.REVERSIBLE,
    approve=lambda name, payload: input(f"{name} を実行しますか? {payload} [y/N] ") == "y",
)
 
 
@guarded(Tier.OBSERVE, policy)
def search_docs(query: str) -> str:
    """ドキュメントを検索して該当箇所を返す"""
    return retriever.invoke(query)[0].page_content
 
 
@guarded(Tier.IRREVERSIBLE, policy)
def publish_release(tag: str) -> str:
    """本番へリリースを公開する"""
    return deploy_client.publish(tag)
 
 
def as_tools(*fns) -> list[StructuredTool]:
    for fn in fns:
        if not hasattr(fn, "permission_tier"):
            raise ValueError(f"{fn.__name__} に権限階層が宣言されていません")
    return [StructuredTool.from_function(fn) for fn in fns]

# pipeline/checkpoint.py
import sqlite3
import json
from contextlib import closing
from typing import Iterable, Iterator, Any
 
SCHEMA = """
CREATE TABLE IF NOT EXISTS progress (
    run_id TEXT NOT NULL,
    item_key TEXT NOT NULL,
    result TEXT,
    PRIMARY KEY (run_id, item_key)
);
"""
 
 
class Checkpoint:
    def __init__(self, path: str = "pipeline_state.db") -> None:
        self.conn = sqlite3.connect(path, isolation_level=None)
        self.conn.execute("PRAGMA journal_mode=WAL")
        self.conn.executescript(SCHEMA)
 
    def done_keys(self, run_id: str) -> set[str]:
        cur = self.conn.execute(
            "SELECT item_key FROM progress WHERE run_id = ?", (run_id,)
        )
        return {row[0] for row in cur}
 
    def record(self, run_id: str, item_key: str, result: Any) -> None:
        self.conn.execute(
            "INSERT OR REPLACE INTO progress VALUES (?, ?, ?)",
            (run_id, item_key, json.dumps(result, ensure_ascii=False)),
        )
 
 
def resumable(
    run_id: str,
    items: Iterable[tuple[str, Any]],
    ckpt: Checkpoint,
) -> Iterator[tuple[str, Any]]:
    """未処理のアイテムだけを流す。中断しても次回はここから続く"""
    already = ckpt.done_keys(run_id)
    for key, payload in items:
        if key in already:
            continue
        yield key, payload

# pipeline/run.py
from pipeline.checkpoint import Checkpoint, resumable
 
ckpt = Checkpoint()
RUN_ID = "wallpaper-classify-2026-03"
 
for key, image in resumable(RUN_ID, iter_assets(), ckpt):
    try:
        result = classification_chain.invoke({"image": image})
    except RateLimitError:
        # 途中で落ちても、記録済みの分は次回スキップされる
        raise
    ckpt.record(RUN_ID, key, result)