learning-sdk-integration

学习 SDK 集成

概述

此技能通过一种适用于 OpenAI、Anthropic、Gemini 和其他大模型提供方的三行集成模式，提供使用 Learning SDK 为大模型 Agent 添加持久记忆的通用模式。

适用场景

在以下情况下使用此技能：

• 构建需要跨会话记忆的大模型 Agent
• 实现对话历史持久化
• 为现有 Agent 添加上下文感知能力
• 创建共享记忆的多 Agent 系统
• 使用任何大模型提供方（OpenAI、Anthropic、Gemini 等）

核心集成模式

基本三行集成

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(...)

多 Agent 记忆共享

# 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. Agent 命名

• 使用能体现用途的描述性 Agent 名称
• 对于相关功能，使用一致的命名模式
• 示例：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)

集成示例

通用研究 Agent

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. 记忆未显示：确保各次调用使用一致的 Agent 名称
2. 性能问题：对于仅记录日志的场景，请使用 capture_only=True
3. 上下文溢出：对于长时间运行的会话，请定期清除记忆
4. 异步冲突：使用异步 SDK 调用时，务必使用 learning_async
5. 提供方兼容性：检查 SDK 版本是否与 Agentic Learning 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