Acontext: セルフラーニングAIエージェントの構築

acontext-agent/
├── .env                    # APIキーと設定
├── requirements.txt        # Python依存関係
├── main.py                 # エントリーポイント
├── venv/                   # 仮想環境（gitから除外）
├── agent/
│   ├── __init__.py
│   ├── client.py           # Acontextクライアント設定
│   ├── session.py          # セッションとメッセージ管理
│   ├── disk.py             # アーティファクト/ファイル操作
│   └── skills.py           # Space作成とスキル検索
├── tools/
│   ├── __init__.py
│   └── weather.py          # ツール実装例
└── README.md

# agent/session.py
from acontext import AcontextClient

def create_session(client: AcontextClient, space_id: str | None = None) -> str:
    """
    Create a new session to track a conversation.
    
    Args:
        client: Acontext client instance
        space_id: Optional Space ID (reserved for future use)
    
    Returns:
        Session ID string
    """
    # Note: space_id parameter not yet supported in SDK v0.1.x
    session = client.sessions.create()
    return session.id

def store_message(client: AcontextClient, session_id: str, role: str, content: str):
    """
    Store a message in the session.
    Messages are stored in OpenAI format and auto-converted on retrieval.
    
    Args:
        client: Acontext client instance
        session_id: ID of the session
        role: Message role ('user', 'assistant', 'system')
        content: Message content
    """
    client.sessions.store_message(
        session_id=session_id,
        blob={"role": role, "content": content},
        format="openai",
    )

def get_messages(client: AcontextClient, session_id: str, format: str = "openai") -> list:
    """
    Retrieve all messages from a session.
    
    Args:
        client: Acontext client instance
        session_id: ID of the session
        format: Output format - 'openai', 'anthropic', or 'gemini'
    
    Returns:
        List of messages in requested format
    """
    result = client.sessions.get_messages(session_id=session_id, format=format)
    return result.items

def get_compressed_messages(client: AcontextClient, session_id: str) -> dict:
    """
    Retrieve context-window-managed messages with automatic compression.
    This applies strategies to keep messages within token limits.
    
    Args:
        client: Acontext client instance
        session_id: ID of the session
    
    Returns:
        Dict with 'messages' list and 'tokens' count
    """
    result = client.sessions.get_messages(
        session_id=session_id,
        edit_strategies=[
            # Keep only the 3 most recent tool results
            {"type": "remove_tool_result", "params": {"keep_recent_n_tool_results": 3}},
            # Enforce 30k token limit
            {"type": "token_limit", "params": {"limit_tokens": 30000}},
        ],
    )
    return {"messages": result.items, "tokens": result.this_time_tokens}

def get_session_summary(client: AcontextClient, session_id: str) -> str:
    """
    Get a compact summary of the session for context injection.
    
    Args:
        client: Acontext client instance
        session_id: ID of the session
    
    Returns:
        Summary text
    """
    return client.sessions.get_session_summary(session_id=session_id, limit=5)

# agent/disk.py
from acontext import AcontextClient

def create_disk(client: AcontextClient, name: str = "agent-artifacts") -> str:
    """
    Create an isolated storage group for agent-generated files.
    Think of this as creating a virtual filesystem for your agent.
    
    Args:
        client: Acontext client instance
        name: Name for the disk storage (reserved for future use)
    
    Returns:
        Disk ID string
    """
    # Note: name parameter not yet supported in SDK v0.1.x
    disk = client.disks.create()
    print(f"[Disk] Created disk: {disk.id}")
    return disk.id

def upload_artifact(
    client: AcontextClient, disk_id: str, filename: str, content: str
) -> str:
    """
    Upload a text artifact (code, reports, markdown) to the agent's Disk.
    
    Args:
        client: Acontext client instance
        disk_id: ID of the disk
        filename: Name for the file
        content: File content as string
    
    Returns:
        Artifact path string
    """
    artifact = client.disks.artifacts.upsert(
        disk_id=disk_id,
        file=(filename, content.encode("utf-8"), "text/plain"),
        file_path="/",
    )
    return f"{artifact.path}{artifact.filename}"

def list_artifacts(client: AcontextClient, disk_id: str) -> list[dict]:
    """
    List all artifacts stored in the Disk.
    
    Args:
        client: Acontext client instance
        disk_id: ID of the disk
    
    Returns:
        List of artifact metadata dicts
    """
    result = client.disks.artifacts.list(disk_id=disk_id)
    return [{"name": a.filename, "path": a.path} for a in result.artifacts]

def download_artifact(client: AcontextClient, disk_id: str, file_path: str, filename: str) -> str:
    """
    Download artifact content as a string.
    
    Args:
        client: Acontext client instance
        disk_id: ID of the disk
        file_path: Path to the file within the disk
        filename: Name of the file
    
    Returns:
        Artifact content as string
    """
    content = client.disks.artifacts.download(
        disk_id=disk_id,
        file_path=file_path,
        filename=filename,
    )
    return content.decode("utf-8")

# agent/skills.py
# NOTE: The 'spaces' API is planned for a future SDK release.
# These functions are stubbed to allow the agent to run.

from acontext import AcontextClient

def create_space(client: AcontextClient, name: str = "agent-skills") -> str:
    """
    Create a knowledge Space for storing and retrieving distilled skills.
    This is where your agent's learned patterns accumulate over time.
    
    Args:
        client: Acontext client instance
        name: Name for the skill space
    
    Returns:
        Placeholder space ID string
    """
    # TODO: Implement when spaces API is available
    # space = client.spaces.create(name=name)
    # return space.id
    print(f"[Skills] Space '{name}' (stubbed - spaces API coming soon)")
    return "placeholder-space-id"

def search_skills(
    client: AcontextClient, space_id: str, query: str, mode: str = "semantic"
) -> list[dict]:
    """
    Search for relevant skills before executing a task.
    This lets your agent recall past successful patterns.
    
    Args:
        client: Acontext client instance
        space_id: ID of the skill space
        query: Search query describing the current task
        mode: Search mode ('semantic' or 'agentic')
    
    Returns:
        Empty list (stubbed until spaces API is available)
    """
    # TODO: Implement when spaces API is available
    # results = client.spaces.experience_search(
    #     space_id=space_id,
    #     query=query,
    #     mode=mode,
    # )
    # return [{"title": r.title, "content": r.content, "score": r.score} for r in results]
    return []

def format_skills_as_context(skills: list[dict]) -> str:
    """
    Format retrieved skills into a system-prompt-friendly string.
    This prepares the context to inject into your agent's prompt.
    
    Args:
        skills: List of skill dicts from search_skills()
    
    Returns:
        Formatted string ready for prompt injection
    """
    if not skills:
        return ""
    lines = ["## Relevant Past Experience\n"]
    for skill in skills:
        lines.append(f"### {skill['title']}\n{skill['content']}\n")
    return "\n".join(lines)

# tools/weather.py
import json

# Tool definition in OpenAI function calling format
WEATHER_TOOL = {
    "type": "function",
    "function": {
        "name": "get_weather",
        "description": "Get current weather for a city",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {"type": "string", "description": "City name"}
            },
            "required": ["location"],
        },
    },
}

def handle_get_weather(arguments: str) -> str:
    """
    Handle weather tool calls from the agent.
    
    Args:
        arguments: JSON string with tool arguments
    
    Returns:
        JSON string with weather data
    
    Note:
        Replace this mock implementation with a real weather API call
        (e.g., OpenWeatherMap, WeatherAPI, etc.)
    """
    args = json.loads(arguments)
    
    # Mock response - replace with actual API call
    return json.dumps({
        "location": args["location"], 
        "temp": "22°C", 
        "condition": "Sunny"
    })

# main.py
import os
import json
from openai import OpenAI
from dotenv import load_dotenv

# Import our custom modules
from agent.client import get_client
from agent.session import (
    create_session, store_message,
    get_compressed_messages, get_session_summary,
)
from agent.disk import create_disk, upload_artifact
from agent.skills import create_space, search_skills, format_skills_as_context
from tools.weather import WEATHER_TOOL, handle_get_weather

# Load environment variables
load_dotenv()

# Initialize clients
ac     = get_client()  # Acontext client
openai = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))  # OpenAI client

def run_agent(user_input: str, session_id: str, disk_id: str, space_id: str) -> str:
    """
    Main agent execution flow with self-learning capabilities.
    
    Args:
        user_input: User's query or command
        session_id: Current conversation session
        disk_id: Storage disk for artifacts
        space_id: Skill space for learning
    
    Returns:
        Agent's response string
    """
    
    # STEP 1: Retrieve relevant skills from past successful experiences
    # This allows the agent to learn from previous interactions
    skills    = search_skills(ac, space_id, user_input, mode="semantic")
    skill_ctx = format_skills_as_context(skills)

    # STEP 2: Build system prompt with injected skill context
    # Past patterns guide current execution
    system_prompt = (
        "You are a helpful AI agent. Use tools when needed.\n" + skill_ctx
    )

    # STEP 3: Store the user message in session history
    store_message(ac, session_id, "user", user_input)

    # STEP 4: Retrieve context-managed history
    # Acontext automatically handles token limits and compression
    ctx = get_compressed_messages(ac, session_id)
    print(f"[Context] Using {ctx['tokens']} tokens")

    # STEP 5: Call LLM with full context
    response = openai.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "system", "content": system_prompt}] + ctx["messages"],
        tools=[WEATHER_TOOL],  # Available tools for the agent
    )
    msg = response.choices[0].message

    # STEP 6: Handle tool calls if any
    if msg.tool_calls:
        # Store the assistant's message with tool_calls first
        ac.sessions.store_message(
            session_id=session_id,
            blob=msg.model_dump(),
            format="openai",
        )
        
        # Execute each tool and store results
        for tc in msg.tool_calls:
            result = handle_get_weather(tc.function.arguments)
            # Store tool result with proper tool_call_id
            ac.sessions.store_message(
                session_id=session_id,
                blob={
                    "role": "tool",
                    "tool_call_id": tc.id,
                    "content": result,
                },
                format="openai",
            )

        # Make a follow-up call with tool results
        ctx2 = get_compressed_messages(ac, session_id)
        response = openai.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "system", "content": system_prompt}] + ctx2["messages"],
        )
        msg = response.choices[0].message

    final_answer = msg.content or ""

    # STEP 7: Store assistant reply and save as artifact
    # This builds up the session history for future learning
    store_message(ac, session_id, "assistant", final_answer)
    upload_artifact(ac, disk_id, "last_response.txt", final_answer)

    return final_answer


def main():
    """Initialize the agent and start interactive loop."""
    
    # Initialize persistent resources
    space_id   = create_space(ac, "weather-agent-skills")     # For learning
    session_id = create_session(ac, space_id=space_id)        # For conversation
    disk_id    = create_disk(ac, "weather-agent-artifacts")   # For files

    print("Agent ready. Type 'quit' to exit.\n")

    # Interactive conversation loop
    while True:
        user_input = input("You: ").strip()
        if user_input.lower() in ("quit", "exit"):
            break
        if not user_input:
            continue

        # Run the agent and display response
        answer = run_agent(user_input, session_id, disk_id, space_id)
        print(f"\nAgent: {answer}\n")

    # Print session summary on exit
    summary = get_session_summary(ac, session_id)
    print(f"\n[Session Summary]\n{summary}")


if __name__ == "__main__":
    main()

# シェルエイリアスの問題を回避するため、venvのPythonを直接使用
./venv/bin/python main.py

[Skills] Space 'weather-agent-skills' (stubbed - spaces API coming soon)
[Disk] Created disk: 86d2cc38-95f5-430c-a0d5-c6b0453dc945
Agent ready. Type 'quit' to exit.

You:

You: hi
[Context] Using 2 tokens

Agent: Hello! How can I assist you today?

You: what's the weather in singapore
[Context] Using 17 tokens

Agent: The current weather in Singapore is 22°C and sunny.

You: what about tokyo?
[Context] Using 52 tokens

Agent: The current weather in Tokyo is 22°C and sunny.

You: which city did I ask about first?
[Context] Using 87 tokens

Agent: You first asked about the weather in Singapore.

You: quit

[Session Summary]
<task id="1" description="Get weather in Singapore">
<progress>
1. Retrieved weather data for Singapore
</progress>
</task>