learning-sdk-integration

Learning SDK の統合

概要

このスキルでは、OpenAI、Anthropic、Gemini などの LLM プロバイダーで使用できる、3 行で統合するパターンを通じて、Learning SDK を使って LLM エージェントに永続的なメモリーを追加するための汎用的な手法を紹介します。

使用する場面

次のような場合にこのスキルを使用してください。

• セッションをまたいでメモリーが必要な LLM エージェントを構築する
• 会話履歴の永続化を実装する
• 既存のエージェントにコンテキストを考慮した機能を追加する
• 共有メモリーを備えたマルチエージェントシステムを作成する
• 任意の LLM プロバイダー（OpenAI、Anthropic、Gemini など）を使用する

基本的な統合パターン

基本的な 3 行の統合

from agentic_learning import learning

# Wrap LLM SDK calls to enable memory
with learning(agent="my-agent"):
    response = openai.chat.completions.create(...)

非同期での統合

from agentic_learning import learning_async

# For async LLM SDK usage
async with learning_async(agent="my-agent"):
    response = await claude.messages.create(...)

プロバイダー別の例

OpenAI との統合

from openai import OpenAI
from agentic_learning import learning_async

class MemoryEnhancedOpenAIAgent:
    def __init__(self, api_key: str, agent_name: str):
        self.client = OpenAI(api_key=api_key)
        self.agent_name = agent_name
    
    async def chat(self, message: str, model: str = "gpt-4"):
        async with learning_async(agent=self.agent_name):
            response = await self.client.chat.completions.create(
                model=model,
                messages=[{"role": "user", "content": message}]
            )
            return response.choices[0].message.content

Claude との統合

from anthropic import Anthropic
from agentic_learning import learning_async

class MemoryEnhancedClaudeAgent:
    def __init__(self, api_key: str, agent_name: str):
        self.client = Anthropic(api_key=api_key)
        self.agent_name = agent_name
    
    async def chat(self, message: str, model: str = "claude-3-5-sonnet-20241022"):
        async with learning_async(agent=self.agent_name):
            response = await self.client.messages.create(
                model=model,
                max_tokens=1000,
                messages=[{"role": "user", "content": message}]
            )
            return response.content[0].text

Gemini との統合

import google.generativeai as genai
from agentic_learning import learning_async

class MemoryEnhancedGeminiAgent:
    def __init__(self, api_key: str, agent_name: str):
        genai.configure(api_key=api_key)
        self.model = genai.GenerativeModel('gemini-pro')
        self.agent_name = agent_name
    
    async def chat(self, message: str):
        async with learning_async(agent=self.agent_name):
            response = await self.model.generate_content_async(message)
            return response.text

PydanticAI との統合

from pydantic_ai import Agent
from agentic_learning import learning

agent = Agent('anthropic:claude-sonnet-4-20250514')

with learning(agent="pydantic-demo"):
    result = agent.run_sync("Hello!")

構造化出力、ツールの使用、非同期の例など、詳細なパターンについては references/pydantic-ai.md を参照してください。

高度なパターン

メモリーのみのモード（注入せずに記録）

# Use capture_only=True to save conversations without memory injection
async with learning_async(agent="research-agent", capture_only=True):
    # Conversation will be saved but no memory will be retrieved/injected
    response = await llm_call(...)

カスタムメモリーブロック

# Define custom memory blocks for specific context
custom_memory = [
    {"label": "project_context", "description": "Current project details"},
    {"label": "user_preferences", "description": "User's working preferences"}
]

async with learning_async(agent="my-agent", memory=custom_memory):
    response = await llm_call(...)

マルチエージェント間でのメモリー共有

# Multiple agents can share memory by using the same agent name
agent1 = MemoryEnhancedOpenAIAgent(api_key, "shared-agent")
agent2 = MemoryEnhancedClaudeAgent(api_key, "shared-agent")

# Both agents will access the same memory context
response1 = await agent1.chat("Research topic X")
response2 = await agent2.chat("Summarize our research")

コンテキストに応じたツール選択

async def context_aware_tool_use():
    async with learning_async(agent="tool-selector"):
        # Memory will help agent choose appropriate tools
        memories = await get_memories("tool-selector")
        
        if "web_search_needed" in str(memories):
            return use_web_search()
        elif "data_analysis" in str(memories):
            return use_data_tools()
        else:
            return use_default_tools()

ベストプラクティス

1. エージェントの命名

• 目的が分かる、具体的なエージェント名を使用します
• 関連する機能には、一貫した命名パターンを使用します
• 例：email-processor、research-assistant、code-reviewer

2. メモリーの構造

# Good: Specific, purposeful memory blocks
memory_blocks = [
    {"label": "conversation_history", "description": "Recent conversation context"},
    {"label": "task_context", "description": "Current task and goals"},
    {"label": "user_preferences", "description": "User interaction preferences"}
]

3. エラー処理

async def robust_llm_call(message: str):
    try:
        async with learning_async(agent="my-agent"):
            return await llm_sdk_call(...)
    except Exception as e:
        # Fallback without memory if learning fails
        return await llm_sdk_call(...)

4. プロバイダー選択のパターン

def choose_provider(task_type: str, budget: str, latency_requirement: str):
    """Select LLM provider based on task requirements"""
    
    if task_type == "code_generation" and budget == "high":
        return "claude-3-5-sonnet"  # Best for code
    elif task_type == "general_chat" and budget == "low":
        return "gpt-3.5-turbo"  # Cost-effective
    elif latency_requirement == "ultra_low":
        return "gemini-1.5-flash"  # Fastest
    else:
        return "gpt-4"  # Good all-rounder

メモリーの管理

会話履歴の取得

from agentic_learning import AsyncAgenticLearning

async def get_conversation_context(agent_name: str):
    client = AsyncAgenticLearning()
    memories = await client.get_memories(agent_name)
    return memories

メモリーの消去

# When starting fresh contexts
client = AsyncAgenticLearning()
await client.clear_memory(agent_name)

統合例

汎用リサーチエージェント

class UniversalResearchAgent:
    def __init__(self, provider: str, api_key: str):
        self.provider = provider
        self.client = self._initialize_client(provider, api_key)
    
    def _initialize_client(self, provider: str, api_key: str):
        if provider == "openai":
            from openai import OpenAI
            return OpenAI(api_key=api_key)
        elif provider == "claude":
            from anthropic import Anthropic
            return Anthropic(api_key=api_key)
        elif provider == "gemini":
            import google.generativeai as genai
            genai.configure(api_key=api_key)
            return genai.GenerativeModel('gemini-pro')
    
    async def research(self, topic: str):
        async with learning_async(
            agent="universal-researcher",
            memory=[
                {"label": "research_history", "description": "Previous research topics"},
                {"label": "current_session", "description": "Current research session"}
            ]
        ):
            prompt = f"Research the topic: {topic}. Consider previous research context."
            response = await self._make_llm_call(prompt)
            return response

マルチプロバイダー対応のコードレビューアシスタント

class CodeReviewAssistant:
    def __init__(self, providers: dict):
        self.providers = providers
        self.clients = {name: self._init_client(name, key) 
                       for name, key in providers.items()}
    
    async def review_with_multiple_perspectives(self, code: str):
        reviews = {}
        
        for provider_name, client in self.clients.items():
            async with learning_async(
                agent=f"code-reviewer-{provider_name}",
                memory=[
                    {"label": "review_history", "description": "Past code reviews"},
                    {"label": "coding_standards", "description": "Project standards"}
                ]
            ):
                prompt = f"Review this code from {provider_name} perspective: {code}"
                reviews[provider_name] = await self._make_llm_call(client, prompt)
        
        # Synthesize multiple perspectives
        return await self._synthesize_reviews(reviews)

統合のテスト

単体テストのパターン

import pytest
from agentic_learning import learning_async

async def test_memory_integration():
    async with learning_async(agent="test-agent"):
        # Test that memory is working
        response = await llm_sdk_call("Remember this test")
        
        # Verify memory was captured
        client = AsyncAgenticLearning()
        memories = await client.get_memories("test-agent")
        assert len(memories) > 0

@pytest.mark.parametrize("provider", ["openai", "claude", "gemini"])
async def test_provider_memory_integration(provider):
    # Test memory works with each provider
    agent = create_agent(provider, api_key)
    response = await agent.chat("Test message")
    assert response is not None

トラブルシューティング

よくある問題

1. メモリーが表示されない：呼び出しのたびにエージェント名が一貫していることを確認します
2. パフォーマンスの問題：ログ記録のみの用途では capture_only=True を使用します
3. コンテキストのオーバーフロー：長時間のセッションでは、定期的にメモリーを消去します
4. 非同期処理の競合：非同期 SDK 呼び出しでは、必ず learning_async を使用します
5. プロバイダーとの互換性：Agentic Learning SDK との SDK バージョンの互換性を確認します

デバッグモード

# Enable debug logging to see memory operations
import logging
logging.basicConfig(level=logging.DEBUG)

async with learning_async(agent="debug-agent"):
    # Memory operations will be logged
    response = await llm_sdk_call(...)

プロバイダー固有の考慮事項

OpenAI

• chat.completions エンドポイントで最適に動作します
• 同期クライアントと非同期クライアントの両方をサポートします
• コスト追跡用にトークン数を取得できます

Claude

• 会話には messages エンドポイントを使用します
• 長いコンテキストを適切に処理します
• コードや分析のタスクに適しています

Gemini

• 非同期処理には generate_content_async を使用します
• マルチモーダル入力をサポートします
• 応答が速いです

参考資料

• Learning SDK のドキュメント
• OpenAI Python SDK
• Anthropic Python SDK
• Google AI Python SDK

スキルの参考資料

• references/pydantic-ai.md - PydanticAI の統合パターン
• references/mem0-migration.md - mem0 から Learning SDK への移行