アシスタントから自律エージェントへ

自律性のスペクトラム

レベル1：純粋なチャットボット

import anthropic

# Level 1: Pure chatbot — single turn, no memory, no tools
def pure_chatbot(user_message: str) -> str:
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')
    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=512,
        messages=[{'role': 'user', 'content': user_message}]
    )
    return response.content[0].text

# What it has:
#   - Language understanding
#   - Knowledge from training
# What it lacks:
#   - Memory (no history between sessions)
#   - Tools (cannot access external data)
#   - Goals (no objective to pursue)
#   - Proactivity (only responds, never initiates)

response = pure_chatbot('What is the capital of Australia?')
print(response)

レベル2：ツール拡張型エージェント

import anthropic

# Level 2: Tool-augmented agent
def tool_augmented_agent(user_message: str, conversation_history: list) -> str:
    client = anthropic.Anthropic(api_key='YOUR_API_KEY')

    tools = [
        {
            'name': 'search_web',
            'description': 'Search the web for current information',
            'input_schema': {'type': 'object',
                             'properties': {'query': {'type': 'string'}},
                             'required': ['query']}
        }
    ]

    conversation_history.append({'role': 'user', 'content': user_message})

    response = client.messages.create(
        model='claude-opus-4-5',
        max_tokens=1024,
        tools=tools,
        messages=conversation_history
    )

    # Handle tool use...
    return response.content[-1].text if response.stop_reason == 'end_turn' else '[tool called]'

# What changed vs Level 1:
#   + Tools (external data access)
#   + Session memory (conversation history)
# Still lacking:
#   - Persistent cross-session memory
#   - Goals (still reactive)
#   - Proactivity

レベル3：目標指向型エージェント

# Level 3: Goal-directed agent
class GoalDirectedAgent:
    def __init__(self, goal: str, tools: list, client):
        self.goal = goal
        self.tools = tools
        self.client = client
        self.memory = []  # persistent across steps
        self.plan = self._make_plan()

    def _make_plan(self) -> list:
        response = self.client.messages.create(
            model='claude-opus-4-5', max_tokens=512,
            messages=[{'role': 'user', 'content':
                f'Goal: {self.goal}\n'
                'Create a numbered list of steps to achieve this goal. '
                'Each step should be a single tool call or reasoning step.'
            }]
        )
        return response.content[0].text

    def step(self) -> str:
        # Execute next planned step
        return 'step executed'

# What changed vs Level 2:
#   + Goal: has an objective to pursue
#   + Planning: decomposes goal into sub-tasks
#   + Persistent memory across steps
# Still lacking:
#   - Self-directed (still initiated by human)
#   - Self-improvement

if __name__ == '__main__':
    class FakeContent:
        def __init__(self, text):
            self.text = text

    class FakeResponse:
        def __init__(self, text):
            self.content = [FakeContent(text)]

    class FakeMessages:
        def create(self, **kwargs):
            return FakeResponse('1. Search web for topic\n2. Summarize findings\n3. Draft report')

    class FakeClient:
        def __init__(self):
            self.messages = FakeMessages()

    agent = GoalDirectedAgent(goal='Write a market report', tools=[], client=FakeClient())
    print('Goal:', agent.goal)
    print('Plan:')
    print(agent.plan)

レベル4：自己主導型エージェント

import time

# Level 4: Self-directed agent (simplified sketch)
class SelfDirectedAgent:
    def __init__(self, mission: str, client):
        self.mission = mission
        self.client = client
        self.goals_queue = []
        self.long_term_memory = []
        self.running = False

    def start(self):
        self.running = True
        self._generate_initial_goals()
        while self.running:
            self._observe_environment()
            self._prioritise_goals()
            if self.goals_queue:
                goal = self.goals_queue.pop(0)
                self._pursue_goal(goal)
            time.sleep(60)  # autonomous monitoring loop

    def _observe_environment(self):
        # Agent monitors for events without being asked
        print('Observing environment...')

    def _generate_initial_goals(self):
        # Agent decomposes its mission into actionable goals
        self.goals_queue = ['Monitor inbox', 'Check project status']

    def _prioritise_goals(self):
        # Agent re-orders goals based on new observations
        pass

    def _pursue_goal(self, goal: str):
        print(f'Pursuing: {goal}')

if __name__ == '__main__':
    agent = SelfDirectedAgent(mission='Manage my inbox proactively', client=None)
    agent._generate_initial_goals()
    print('Initial goals:', agent.goals_queue)
    agent._observe_environment()
    goal = agent.goals_queue.pop(0)
    agent._pursue_goal(goal)

各レベルで変わる点：メモリ

MEMORY_BY_LEVEL = {
    'L1_chatbot': {
        'scope': 'context_window_only',
        'persistence': 'none',
        'implementation': 'messages list in current API call'
    },
    'L2_tool_augmented': {
        'scope': 'session',
        'persistence': 'in-memory (lost on restart)',
        'implementation': 'conversation_history list'
    },
    'L3_goal_directed': {
        'scope': 'task',
        'persistence': 'persists for task duration',
        'implementation': 'SQLite or Redis with task state'
    },
    'L4_self_directed': {
        'scope': 'long_term',
        'persistence': 'indefinite',
        'implementation': 'vector DB (episodic) + structured DB (semantic) + prompt cache (procedural)'
    }
}

for level, info in MEMORY_BY_LEVEL.items():
    print(f'{level}: {info["implementation"]}')

各レベルで変わる点：計画

PLANNING_BY_LEVEL = {
    'L1': 'None — single response',
    'L2': 'Single-step tool selection (which tool to call now)',
    'L3': 'Multi-step plan (goal -> ordered sub-tasks -> tool calls)',
    'L4': 'Hierarchical plan (mission -> goals -> tasks -> actions) + replan on failure'
}

# L3 multi-step planning example:
def plan_goal(goal: str, client) -> list:
    import anthropic
    response = client.messages.create(
        model='claude-opus-4-5', max_tokens=512,
        messages=[{'role': 'user', 'content':
            f'Break this goal into 3-5 concrete steps:\nGoal: {goal}\n'
            'Return JSON: {"steps": [{"step": int, "action": str, "tool": str}]}'
        }]
    )
    import json
    return json.loads(response.content[0].text)

for lvl, desc in PLANNING_BY_LEVEL.items():
    print(f'{lvl}: {desc}')

レベル別の人間による監督要件

OVERSIGHT_BY_LEVEL = {
    'L1_chatbot': [
        'None required (output only)'
    ],
    'L2_tool_augmented': [
        'Review tool permissions (read-only vs write)',
        'Audit logs of tool calls'
    ],
    'L3_goal_directed': [
        'Human approval before irreversible actions',
        'Plan review before execution starts',
        'Progress checkpoints',
        'Full audit trail'
    ],
    'L4_self_directed': [
        'Human approval before high-impact actions',
        'Real-time action streaming to oversight dashboard',
        'Emergency stop mechanism',
        'Anomaly detection on goal drift',
        'Regular review of long-term memory state',
        'Corrigibility: agent must accept shutdown'
    ]
}

for level, requirements in OVERSIGHT_BY_LEVEL.items():
    print(f'{level}:')
    for req in requirements:
        print(f'  - {req}')

プロアクティブ性：重要な転換点

# Proactive monitoring pattern
import asyncio
from datetime import datetime

class ProactiveMonitor:
    def __init__(self, agent_fn, check_fn, interval_seconds: int = 60):
        self.agent_fn = agent_fn
        self.check_fn = check_fn
        self.interval = interval_seconds

    async def run(self):
        print(f'Proactive monitor started, checking every {self.interval}s')
        while True:
            try:
                events = await self.check_fn()
                for event in events:
                    print(f'[{datetime.utcnow().isoformat()}] Event: {event}')
                    await self.agent_fn(event)
            except Exception as e:
                print(f'Monitor error: {e}')
            await asyncio.sleep(self.interval)

# Example: agent monitors for new emails every 5 minutes
# and proactively drafts replies or flags urgent ones
async def example_setup():
    monitor = ProactiveMonitor(
        agent_fn=lambda e: print(f'Agent handling: {e}'),
        check_fn=lambda: [],  # replace with real inbox check
        interval_seconds=300
    )
    # await monitor.run()

if __name__ == '__main__':
    import asyncio

    async def demo():
        async def check_fn():
            return ['New email from boss@example.com']

        async def agent_fn(event):
            print(f'Agent drafting reply for: {event}')

        monitor = ProactiveMonitor(agent_fn=agent_fn, check_fn=check_fn, interval_seconds=1)
        try:
            await asyncio.wait_for(monitor.run(), timeout=0.3)
        except asyncio.TimeoutError:
            pass

    asyncio.run(demo())

レベル別のエラー回復

# Error recovery strategies by autonomy level

def chatbot_error_recovery(error: Exception) -> str:
    return 'I\'m sorry, I encountered an error. Please try again.'

def tool_agent_error_recovery(tool_name: str, error: Exception, retries: int) -> str:
    if retries < 3:
        return f'Retrying {tool_name} ({retries+1}/3)'
    return f'Tool {tool_name} unavailable after 3 retries, skipping'

def goal_agent_error_recovery(failed_step: dict, plan: list, client) -> list:
    import anthropic, json
    client_obj = anthropic.Anthropic(api_key='YOUR_API_KEY')
    response = client_obj.messages.create(
        model='claude-opus-4-5', max_tokens=256,
        messages=[{'role': 'user', 'content':
            f'Step failed: {failed_step}\nRemaining plan: {plan}\n'
            'Revise the remaining plan to work around the failure. Return JSON plan.'
        }]
    )
    return json.loads(response.content[0].text)

print('Error recovery patterns defined for each autonomy level')

適切なレベルの選択

def recommend_autonomy_level(requirements: dict) -> str:
    needs_realtime = requirements.get('realtime_data', False)
    needs_multistep = requirements.get('multi_step_tasks', False)
    needs_unsupervised = requirements.get('runs_unsupervised', False)
    needs_initiative = requirements.get('initiates_actions', False)

    if needs_initiative and needs_unsupervised:
        return 'L4_self_directed (high oversight required)'
    if needs_multistep:
        return 'L3_goal_directed (plan review recommended)'
    if needs_realtime:
        return 'L2_tool_augmented (audit logs required)'
    return 'L1_chatbot (minimal oversight)'

# Examples:
print(recommend_autonomy_level({
    'realtime_data': True, 'multi_step_tasks': False,
    'runs_unsupervised': False, 'initiates_actions': False
}))
print(recommend_autonomy_level({
    'realtime_data': True, 'multi_step_tasks': True,
    'runs_unsupervised': True, 'initiates_actions': True
}))

理解度チェック

まとめ：アシスタントから自律型エージェントへ