ReActをゼロから実装する

ループを自分で構築する

Step 1: Define Tools

import json, math

def calculator(expression: str) -> str:
    try:
        return str(eval(expression, {'__builtins__': {}}, vars(math)))
    except Exception as e:
        return f'Error: {e}'

def search_wikipedia(query: str) -> str:
    # placeholder — call a real search API
    return f'Wikipedia summary for: {query}'

TOOLS = {
    'calculator': calculator,
    'search_wikipedia': search_wikipedia,
}

print(calculator('2 + 2 * sqrt(16)'))
print(search_wikipedia('agents'))

Step 2: Tool Schemas

schemas = [
    {'type': 'function', 'function': {
        'name': 'calculator',
        'description': 'Evaluate a math expression like 2 + 2 * sqrt(16)',
        'parameters': {'type': 'object', 'properties': {'expression': {'type': 'string'}}, 'required': ['expression']}
    }},
    {'type': 'function', 'function': {
        'name': 'search_wikipedia',
        'description': 'Search Wikipedia and return a summary',
        'parameters': {'type': 'object', 'properties': {'query': {'type': 'string'}}, 'required': ['query']}
    }}
]
import json
print(json.dumps(schemas, indent=2))

Step 3: System Prompt

system = '''
You are a research assistant. Use the available tools when needed.
Think step by step. If you have enough information, give a final answer.
'''
print(system.strip())

Step 4: The Loop

from openai import OpenAI
client = OpenAI()

def react(question, max_steps=10):
    messages = [
        {'role': 'system', 'content': system},
        {'role': 'user', 'content': question}
    ]
    for step in range(max_steps):
        r = client.chat.completions.create(
            model='gpt-4o-mini',
            messages=messages,
            tools=schemas,
        )
        msg = r.choices[0].message
        messages.append(msg)
        if not msg.tool_calls:
            return msg.content
        for tc in msg.tool_calls:
            args = json.loads(tc.function.arguments)
            try:
                result = TOOLS[tc.function.name](**args)
            except Exception as e:
                result = f'Tool error: {e}'
            messages.append({'role': 'tool', 'tool_call_id': tc.id, 'content': str(result)})
    return 'Step limit reached.'

Step 5: Run It

answer = react('What is the population of Tokyo times the square root of pi?')
print(answer)

ステップ 6：ログを追加する

def react_verbose(question, max_steps=10):
    messages = [...]
    for step in range(max_steps):
        r = client.chat.completions.create(model='gpt-4o-mini', messages=messages, tools=schemas)
        msg = r.choices[0].message
        print(f'[step {step}] thought: {msg.content}')
        if msg.tool_calls:
            for tc in msg.tool_calls:
                print(f'[step {step}] action: {tc.function.name}({tc.function.arguments})')
        ...

ステップ 7：ツールとしてのステップカウンター

system = f'You have at most {MAX_STEPS} steps. Step {current_step}/{MAX_STEPS}. ...'

ステップ 8：最終回答の形式

tools.append({'type': 'function', 'function': {
    'name': 'final_answer',
    'description': 'Submit the final answer to the user.',
    'parameters': {'type': 'object', 'properties': {'answer': {'type': 'string'}}, 'required': ['answer']}
}})
# Loop terminates when final_answer is called.

ステップ 9：ツールエラー

try:
    result = TOOLS[name](**args)
except Exception as e:
    result = f'TOOL ERROR: {type(e).__name__}: {e}'
# Model sees the error and can try a different approach.

ステップ 10：ツール呼び出しの並列実行

import asyncio

async def run_tools(tool_calls):
    async def run_one(tc):
        args = json.loads(tc.function.arguments)
        return tc.id, TOOLS[tc.function.name](**args)
    return await asyncio.gather(*[run_one(tc) for tc in tool_calls])

ステップ 11：会話の長さを制限する

if total_tokens(messages) > 6000:
    messages = trim_messages(messages)

ステップ 12：本番環境への対応

ツールエラーの処理

復習