エージェントの失敗とループへの対処

エージェントが失敗してループする理由

最大反復回数の制限

from langchain.agents import AgentExecutor

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=10,             # Hard stop after 10 steps
    max_execution_time=30.0,       # Also stop after 30 wall-clock seconds
    early_stopping_method='generate',  # Ask the model for a partial answer at the limit
    verbose=True
)

早期停止：最終回答を強制する

# The 'generate' early_stopping_method adds this system instruction
# when max_iterations is reached:
#
# 'You have {N} steps remaining but the task is not complete.
#  Give your best final answer based on the information gathered so far.'
#
# Contrast with 'force' which abruptly terminates without generating an answer.

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    max_iterations=7,
    early_stopping_method='generate'
)

解析エラーの適切な処理

agent_executor = AgentExecutor(
    agent=agent,
    tools=tools,
    handle_parsing_errors=True,
    # Custom error message fed back to the model:
    # handle_parsing_errors='Please format your response as Thought/Action/Action Input.'
)

# When a parse error occurs, the executor automatically adds:
# Observation: Could not parse LLM output. Please follow the format:
#   Thought: ...
#   Action: tool_name
#   Action Input: ...

反復ループの検出と中断

from collections import Counter

class LoopDetector:
    def __init__(self, max_repeats: int = 2):
        self.max_repeats = max_repeats
        self.call_counts = Counter()

    def check(self, tool_name: str, tool_input: str) -> bool:
        key = f'{tool_name}:{tool_input}'
        self.call_counts[key] += 1
        if self.call_counts[key] > self.max_repeats:
            return True  # Loop detected
        return False

    def get_warning(self) -> str:
        return ('You have called this tool with the same input multiple times. '
                'Try a different approach, different search terms, or a different tool.')

ツールレベルのエラー処理

from langchain_core.tools import tool
import requests

@tool
def get_company_data(company_name: str) -> str:
    '''Retrieve company information from the business database.
    Input: company name as a string.
    '''
    try:
        resp = requests.get(
            f'https://api.example.com/companies/{company_name}',
            timeout=5
        )
        if resp.status_code == 404:
            return f'No company found with name "{company_name}". Try the exact legal name or ticker symbol.'
        if resp.status_code == 429:
            return 'Rate limit exceeded. Wait 60 seconds before trying again.'
        resp.raise_for_status()
        return resp.json().get('summary', 'No summary available.')
    except requests.Timeout:
        return 'The database is not responding. Try searching the web instead.'

API障害時の指数バックオフ

import time
import requests
from langchain_core.tools import tool

@tool
def reliable_search(query: str) -> str:
    '''Search with automatic retry on failure. Input: search query string.'''
    max_retries = 3
    for attempt in range(max_retries):
        try:
            resp = requests.get(
                'https://api.duckduckgo.com/',
                params={'q': query, 'format': 'json'},
                timeout=10
            )
            resp.raise_for_status()
            data = resp.json()
            return data.get('AbstractText', 'No results found.')
        except requests.RequestException as e:
            if attempt < max_retries - 1:
                wait = 2 ** attempt  # 1s, 2s, 4s
                time.sleep(wait)
            else:
                return f'Search failed after {max_retries} attempts: {str(e)}'

エージェントレベルのタイムアウト予算

import asyncio

async def run_with_timeout(user_input: str, timeout_seconds: float = 30.0) -> str:
    try:
        result = await asyncio.wait_for(
            agent_executor.ainvoke({'input': user_input}),
            timeout=timeout_seconds
        )
        return result['output']
    except asyncio.TimeoutError:
        return ('I am taking longer than expected to answer this question. '
                'Please try again with a simpler question, or check back later.')

分析のための失敗ログ

import logging
import json

logger = logging.getLogger('agent')

def run_and_log(user_input: str) -> str:
    try:
        result = agent_executor.invoke(
            {'input': user_input},
            return_intermediate_steps=True
        )
        if not result.get('output'):
            logger.warning('Agent returned empty output', extra={
                'input': user_input,
                'steps': len(result.get('intermediate_steps', []))
            })
        return result['output']
    except Exception as e:
        logger.error('Agent failed with exception', extra={
            'input': user_input,
            'error': str(e),
            'error_type': type(e).__name__
        })
        return 'I encountered an error. Please try rephrasing your question.'

プロンプトへの復旧ヒントの注入

失敗シナリオのテスト

クイックチェック

レッスンのまとめ