Интеграция OpenAI Assistants API для разработки агентов

Интеграция OpenAI Assistants API для разработки агентов

OpenAI Assistants API — managed-сервис для создания агентов с персистентным состоянием: Threads (хранят историю диалогов), Files (загруженные документы), Code Interpreter (выполнение Python в sandbox), File Search (встроенный RAG). В отличие от Chat Completions API, Assistants берут на себя управление памятью и жизненным циклом.

Базовая интеграция

from openai import OpenAI
import time

client = OpenAI()

# Создание ассистента
assistant = client.beta.assistants.create(
    name="Corporate Support Assistant",
    instructions="""Ты — ассистент техподдержки компании TechCorp.
Помогай клиентам решать проблемы. Используй загруженные документы как источник истины.
При необходимости выполняй код для расчётов и анализа данных.""",
    model="gpt-4o",
    tools=[
        {"type": "file_search"},        # Встроенный RAG
        {"type": "code_interpreter"},   # Выполнение Python
    ],
)

# Создание Thread (хранит историю)
thread = client.beta.threads.create()

# Добавление сообщения
message = client.beta.threads.messages.create(
    thread_id=thread.id,
    role="user",
    content="Как настроить SSO для корпоративного аккаунта?",
)

# Запуск ассистента
run = client.beta.threads.runs.create(
    thread_id=thread.id,
    assistant_id=assistant.id,
)

# Ожидание завершения
while run.status in ["queued", "in_progress"]:
    run = client.beta.threads.runs.retrieve(thread_id=thread.id, run_id=run.id)
    time.sleep(0.5)

# Получение ответа
messages = client.beta.threads.messages.list(thread_id=thread.id)
print(messages.data[0].content[0].text.value)

File Search: загрузка и индексирование документов

# Создание Vector Store (хранилище для RAG)
vector_store = client.beta.vector_stores.create(name="Support Docs")

# Загрузка документов
with open("docs/user-guide.pdf", "rb") as f:
    file = client.beta.vector_stores.files.upload_and_poll(
        vector_store_id=vector_store.id,
        file=f,
    )

# Привязка к ассистенту
assistant = client.beta.assistants.update(
    assistant_id=assistant.id,
    tool_resources={"file_search": {"vector_store_ids": [vector_store.id]}},
)

Кастомные функции + streaming

import json

# Ассистент с кастомными функциями
assistant_with_tools = client.beta.assistants.create(
    name="CRM Assistant",
    model="gpt-4o",
    tools=[{
        "type": "function",
        "function": {
            "name": "get_customer_info",
            "description": "Получить информацию о клиенте из CRM",
            "parameters": {
                "type": "object",
                "properties": {
                    "customer_id": {"type": "string"},
                },
                "required": ["customer_id"],
            },
        },
    }],
)

# Обработка function calls с streaming
def handle_run_with_functions(thread_id: str, assistant_id: str):
    with client.beta.threads.runs.stream(
        thread_id=thread_id,
        assistant_id=assistant_id,
    ) as stream:
        for event in stream:
            if event.event == "thread.run.requires_action":
                # Выполняем функции
                tool_outputs = []
                for tool_call in event.data.required_action.submit_tool_outputs.tool_calls:
                    if tool_call.function.name == "get_customer_info":
                        args = json.loads(tool_call.function.arguments)
                        result = crm.get_customer(args["customer_id"])
                        tool_outputs.append({
                            "tool_call_id": tool_call.id,
                            "output": json.dumps(result),
                        })

                # Отправляем результаты
                stream.submit_tool_outputs_and_stream(tool_outputs)

            elif event.event == "thread.message.delta":
                for delta in event.data.delta.content:
                    if delta.type == "text":
                        print(delta.text.value, end="", flush=True)

Практический кейс: корпоративный FAQ-ассистент

Задача: HR-отдел получал 50+ повторяющихся вопросов в день (отпуска, документы, льготы). Один HR-менеджер тратил 2 часа ежедневно на ответы.

Архитектура: Assistants API + File Search (15 регламентов загружены в Vector Store) + интеграция в корпоративный Slack.

Результаты:

• Автономные ответы на 73% вопросов
• Время внедрения: 5 дней (vs 2 недели на кастомный RAG)
• HR-менеджер высвободил 1.5 часа/день

Ограничения Assistants API: высокая стоимость хранения (Vector Store платный), нет контроля над алгоритмом чанкинга, сложнее настроить гибридный поиск. Для production-RAG с высокими требованиями к качеству — предпочтителен кастомный LangChain/LlamaIndex стек.

Сроки

• Базовый ассистент + File Search: 1–3 дня
• Кастомные функции + streaming: 3–5 дней
• Production-деплой с мониторингом: 1 неделя