エージェントループでよくある失敗

エージェントループとその失敗モード

失敗1：無限ループ

# Example of an agent in an infinite loop:
# Step 1: reasoning='Need to search for Python docs'
#         tool='search_web', args={'query': 'Python documentation'}
# Step 2: reasoning='Search result was unhelpful, try again'
#         tool='search_web', args={'query': 'Python documentation'}
# Step 3: reasoning='Search result was unhelpful, try again'
#         tool='search_web', args={'query': 'Python documentation'}
# ... repeats until max_iterations or token budget is exhausted

print('Symptom: same tool + same arguments appearing repeatedly in steps')
print('Fix: detect repeated (tool, args) pairs and break the loop')

失敗2：行き詰まった状態

# Example of a stuck agent:
# Step 1: tool='search_web', args={'query': 'topic A'}
# Step 2: tool='search_web', args={'query': 'topic B'}  # different args
# Step 3: tool='search_web', args={'query': 'topic A'}  # back to first
# Step 4: tool='read_document', args={'url': '...'}
# Step 5: tool='search_web', args={'query': 'topic A'}
# ... no FINAL_ANSWER ever produced

print('Symptom: agent takes many steps but never calls FINAL_ANSWER')
print('Fix: max_iterations guard + force final answer if limit is near')

失敗3：最終回答の欠落

# An agent that never concludes:
def run_agent_bad(query: str, max_steps: int = 20) -> str:
    for step in range(max_steps):
        action = llm_decide_action(query, history)

        if action['type'] == 'tool':
            result = execute_tool(action)
            history.append(result)
        # BUG: No check for 'final_answer' type!
        # The agent loops until max_steps, returning None

    return None  # never actually returns an answer

# Fix: explicitly check for final_answer signal
def run_agent_good(query: str, max_steps: int = 20) -> str:
    for step in range(max_steps):
        action = llm_decide_action(query, history)
        if action['type'] == 'final_answer':
            return action['answer']  # exit cleanly
        execute_tool(action)
    return 'Reached step limit without a conclusion.'

失敗4：ツール呼び出しの解析エラー

import json

def safe_parse_tool_call(arguments_str: str) -> dict:
    try:
        return json.loads(arguments_str)
    except json.JSONDecodeError as e:
        print(f'Failed to parse tool arguments: {e}')
        print(f'Raw: {arguments_str}')
        return None

def execute_step(tool_call) -> str:
    args = safe_parse_tool_call(tool_call.function.arguments)
    if args is None:
        # Feed the error back to the LLM in the next step
        return f'ERROR: Could not parse tool arguments. Raw: {tool_call.function.arguments}'
    return run_tool(tool_call.function.name, args)

失敗5：ツールが有用なデータを返さない

def run_agent_with_empty_result_handling(query: str) -> str:
    for step in range(20):
        action = decide_next_action(query, history)

        if action['type'] == 'final_answer':
            return action['answer']

        result = execute_tool(action['tool'], action['args'])

        # Detect empty results and provide context
        if not result or result.strip() == '':
            observation = f'Tool {action["tool"]} returned no data. Try a different approach or different arguments.'
        elif 'error' in result.lower():
            observation = f'Tool error: {result}. Consider a different tool or query.'
        else:
            observation = result

        history.append({'tool': action['tool'], 'result': observation})

    return 'Could not complete task within step limit.'

失敗6：存在しないツール名の生成

REGISTERED_TOOLS = {
    'search_web': search_web_function,
    'get_weather': get_weather_function,
    'calculate': calculate_function
}

def dispatch_tool(tool_name: str, args: dict) -> str:
    if tool_name not in REGISTERED_TOOLS:
        available = ', '.join(REGISTERED_TOOLS.keys())
        return (
            f'ERROR: Unknown tool "{tool_name}". '
            f'Available tools: {available}. '
            f'Please use one of the available tools.'
        )

    tool_fn = REGISTERED_TOOLS[tool_name]
    return tool_fn(**args)

失敗7：トークン予算の枯渇

def truncate_tool_result(result: str, max_chars: int = 2000) -> str:
    if len(result) <= max_chars:
        return result
    truncated = result[:max_chars]
    return f'{truncated}\n... [result truncated to {max_chars} chars]'

def add_observation_to_history(history: list, tool_name: str, result: str):
    safe_result = truncate_tool_result(result, max_chars=2000)
    history.append({
        'role': 'tool',
        'content': safe_result,
        'tool_name': tool_name
    })
    print(f'[Step] Tool={tool_name}, Result length={len(result)} (stored {len(safe_result)})')

if __name__ == '__main__':
    demo_history = []
    add_observation_to_history(demo_history, 'search_web', 'x' * 3000)

失敗モードをプログラムで検出する

def diagnose_agent_failure(steps: list) -> str:
    if not steps:
        return 'No steps recorded'

    # Check for infinite loop: same (tool, args) repeated
    seen = {}
    for s in steps:
        key = (s.get('tool'), str(s.get('args')))
        seen[key] = seen.get(key, 0) + 1
    repeated = {k: v for k, v in seen.items() if v > 2}
    if repeated:
        return f'INFINITE_LOOP: repeated actions: {repeated}'

    # Check for missing final answer
    has_answer = any(s.get('type') == 'final_answer' for s in steps)
    if not has_answer and len(steps) >= 15:
        return 'STUCK_STATE: many steps taken but no final answer'

    # Check for parse errors
    errors = [s for s in steps if 'ERROR' in str(s.get('result', ''))]
    if len(errors) > 2:
        return f'TOOL_ERROR: {len(errors)} tool errors in pipeline'

    return 'OK'

if __name__ == '__main__':
    demo_steps = [{'tool': 'search_web', 'args': {'q': 'weather'}} for _ in range(3)]
    print('Diagnosis:', diagnose_agent_failure(demo_steps))

シンプルなステップ数上限を実装する

def run_agent_with_budget(query: str, max_steps: int = 15) -> dict:
    history = []
    for step in range(1, max_steps + 1):
        print(f'[Step {step}/{max_steps}]')

        action = decide_next_action(query, history)

        if action['type'] == 'final_answer':
            return {
                'status': 'success',
                'answer': action['answer'],
                'steps_taken': step
            }

        result = execute_tool(action['tool'], action['args'])
        history.append({'step': step, 'tool': action['tool'], 'result': result})

        if step == max_steps - 1:
            # Warn the agent it must conclude
            history.append({'role': 'system',
                            'content': 'You must provide a FINAL_ANSWER on the next step.'})

    return {'status': 'timeout', 'answer': None, 'steps_taken': max_steps}

クイックリファレンス：失敗モードと修正方法

理解度チェック：エージェントループの失敗

振り返り：エージェントループでよくある失敗