Интерактивные агенты в стиле REPL

Что такое агент в стиле REPL

Базовый цикл REPL

def run_agent_repl(inputs):
    print('Agent started. Type /exit to quit.')
    conversation_history = []
    for user_input in inputs:
        user_input = user_input.strip()
        if not user_input:
            continue
        if user_input == '/exit':
            print('Goodbye!')
            break
        conversation_history.append({'role': 'user', 'content': user_input})
        agent_reply = f'[Agent reply to: {user_input}]'
        conversation_history.append({'role': 'assistant', 'content': agent_reply})
        print(f'Agent: {agent_reply}')

run_agent_repl(['Hello agent', 'What can you do?', '/exit'])

Корректная обработка KeyboardInterrupt

def run_agent_repl():
    print('Agent ready. Press Ctrl+C to exit.')
    conversation_history = []

    try:
        while True:
            user_input = input('You: ').strip()
            if not user_input:
                continue

            conversation_history.append({'role': 'user', 'content': user_input})
            reply = '[mocked agent response]'
            conversation_history.append({'role': 'assistant', 'content': reply})
            print(f'Agent: {reply}')

    except KeyboardInterrupt:
        print('\nSession ended. Goodbye!')
    except EOFError:
        # Triggered when stdin is closed (piped input ends)
        print('\nEnd of input. Exiting.')

if __name__ == '__main__':
    import builtins
    _demo_inputs = iter(['What agents can you build?', EOFError])
    def _fake_input(prompt=''):
        val = next(_demo_inputs)
        if val is EOFError:
            raise EOFError
        print(prompt + val)
        return val
    builtins.input = _fake_input
    run_agent_repl()

Сохранение истории многошагового разговора

import openai

client = openai.OpenAI(api_key='YOUR_API_KEY')

def run_chat_repl():
    history = [
        {'role': 'system', 'content': 'You are a helpful AI assistant.'}
    ]

    print('Chat started. Type /exit to quit.')

    try:
        while True:
            user_input = input('You: ').strip()
            if not user_input or user_input == '/exit':
                break

            history.append({'role': 'user', 'content': user_input})

            response = client.chat.completions.create(
                model='gpt-4o-mini',
                messages=history
            )

            reply = response.choices[0].message.content
            history.append({'role': 'assistant', 'content': reply})
            print(f'Agent: {reply}')

    except KeyboardInterrupt:
        print('\nGoodbye!')

readline: история и редактирование ввода

try:
    import readline  # available on Unix/Mac, not Windows
    # Enable persistent history between sessions
    import os
    HISTORY_FILE = os.path.expanduser('~/.agent_history')
    try:
        readline.read_history_file(HISTORY_FILE)
    except FileNotFoundError:
        pass

    import atexit
    atexit.register(readline.write_history_file, HISTORY_FILE)
    readline.set_history_length(500)
    print('History enabled (use up/down arrows)')
except ImportError:
    pass  # readline not available on Windows

# After this, input() automatically has history support

Встроенные команды со слешем

def handle_command(cmd: str, history: list) -> bool:
    'Returns True if the command was handled, False if it is a regular message'
    if cmd == '/exit' or cmd == '/quit':
        print('Goodbye!')
        raise SystemExit(0)

    elif cmd == '/help':
        print('Commands: /exit, /clear, /history, /help')
        return True

    elif cmd == '/clear':
        # Keep only the system message
        system = history[0] if history and history[0]['role'] == 'system' else None
        history.clear()
        if system:
            history.append(system)
        print('[Conversation cleared]')
        return True

    elif cmd == '/history':
        for i, msg in enumerate(history):
            if msg['role'] != 'system':
                print(f'  [{msg["role"]}] {msg["content"][:80]}')
        return True

    return False  # not a command, treat as regular message

if __name__ == '__main__':
    demo_history = [
        {'role': 'system', 'content': 'You are a helpful agent.'},
        {'role': 'user', 'content': 'Hi there'},
        {'role': 'assistant', 'content': 'Hello! How can I help?'},
    ]
    handle_command('/help', demo_history)
    handle_command('/history', demo_history)

Ограничение длины истории разговора

def trim_history(history: list, max_turns: int = 10) -> list:
    system_messages = [m for m in history if m['role'] == 'system']
    non_system = [m for m in history if m['role'] != 'system']

    # Keep the last max_turns messages (each turn = 1 user + 1 assistant)
    max_messages = max_turns * 2
    trimmed = non_system[-max_messages:]

    return system_messages + trimmed

# Usage in the REPL loop:
# history.append({'role': 'user', 'content': user_input})
# trimmed = trim_history(history, max_turns=10)
# response = client.chat.completions.create(model='gpt-4o-mini', messages=trimmed)
print('Trimming history prevents token count from growing unboundedly')

Отображение приветственного баннера

def print_banner():
    banner = '''
====================================
  AI Research Agent v1.0
====================================
  Type your question to begin.
  Commands: /help  /clear  /exit
====================================
    '''
    print(banner)

# Or with rich colors:
# from rich.console import Console
# from rich.panel import Panel
# console = Console()
# console.print(Panel('[bold cyan]AI Research Agent[/] v1.0', subtitle='Type /help for commands'))

print_banner()

Отображение индикаторов обработки

import sys
import threading
import time

def thinking_spinner(stop_event: threading.Event):
    frames = ['|', '/', '-', '\\']
    i = 0
    while not stop_event.is_set():
        sys.stdout.write(f'\rAgent is thinking... {frames[i % 4]}')
        sys.stdout.flush()
        time.sleep(0.1)
        i += 1
    sys.stdout.write('\r' + ' ' * 30 + '\r')  # clear the line
    sys.stdout.flush()

def call_agent_with_spinner(query: str) -> str:
    stop = threading.Event()
    spinner = threading.Thread(target=thinking_spinner, args=(stop,))
    spinner.start()

    # result = agent.run(query)  # the actual slow call
    time.sleep(1)  # simulating work
    result = 'The answer is 42.'

    stop.set()
    spinner.join()
    return result

if __name__ == '__main__':
    answer = call_agent_with_spinner('What is 6 * 7?')
    print(f'\nFinal answer: {answer}')

Сохранение сеанса в файл

import json
import datetime
import os

def save_session(history: list, directory: str = 'sessions'):
    os.makedirs(directory, exist_ok=True)
    timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
    filename = os.path.join(directory, f'session_{timestamp}.json')

    with open(filename, 'w') as f:
        json.dump({
            'saved_at': timestamp,
            'messages': history
        }, f, indent=2)

    print(f'Session saved to {filename}')

# Usage in REPL on /save command:
# save_session(conversation_history)

if __name__ == '__main__':
    import tempfile
    demo_dir = tempfile.mkdtemp()
    demo_history = [
        {'role': 'user', 'content': 'Hello'},
        {'role': 'assistant', 'content': 'Hi! How can I help?'},
    ]
    save_session(demo_history, directory=demo_dir)

Полный шаблон агента REPL

def run_repl(agent):
    history = [{'role': 'system', 'content': 'You are a helpful AI assistant.'}]
    print('Agent ready. Type /help for commands.')
    try:
        while True:
            user_input = input('You: ').strip()
            if not user_input:
                continue
            if user_input.startswith('/'):
                handle_command(user_input, history)
                continue
            history.append({'role': 'user', 'content': user_input})
            trimmed = trim_history(history, max_turns=10)
            reply = agent.run(trimmed)
            history.append({'role': 'assistant', 'content': reply})
            print(f'Agent: {reply}')
    except KeyboardInterrupt:
        print('\nSession ended.')
        save_session(history)

Проверка знаний: агенты в стиле REPL

Итоги: интерактивные агенты в стиле REPL