שילוב OpenAI Agents SDK: העברות, מעקות בטיחות, מעקב

הטמעת לקוח B2B חדש בשירות SaaS אורכת שבועיים. היא דורשת תיאום בין שלוש מחלקות: הטמעה, אינטגרציה ותמיכה. בעיות אופייניות: ניתוב ידני של פניות, דליפת מידע אישי דרך המודל, וחוסר שקיפות בקבלת החלטות. אוטמטנו תהליך זה באמצעות מערכת מרובת סוכנים המבוססת על OpenAI Agents SDK (להלן SDK) — חבילת Python רשמית מ-OpenAI לבניית סוכני AI. תוצאה: הזמן עד לאינטגרציה הראשונה ירד מ-14 ל-3 ימים, פניות התמיכה ירדו ב-54%, ועלויות התפעול ב-40%.

הפתרון מבוסס על האבסטרקציות Agent, Runner, Handoffs, Guardrails וניטור מובנה. ה-SDK מחליף קריאות ישירות ל-Assistants API בממשק טיפוסי, מה שמפשט את התחזוקה והבדיקות. הניסיון שלנו: 5 שנים בפיתוח AI/ML, למעלה מ-20 פרויקטים על OpenAI API. אנו מיישמים פתרונות turnkey עם זמן תגובה מובטח של סוכן של p99 < 2 שניות. קבלו ייעוץ — המהנדס שלנו יעריך את התרחיש שלכם.

כיצד OpenAI Agents SDK פותר בעיות במערכות מרובות סוכנים?

ללא ה-SDK, מפתחים נתקלים בשלושה אתגרים עיקריים:

• כאוס עם סוכנים מרובים: ניתוב ידני של פניות והעברת קונטקסט.
• דליפת PII: המודל עלול לחשוף מספרי כרטיסים או פרטי דרכון ללא פילטרים נוספים.
• חוסר שקיפות: בלתי אפשרי להבין מדוע סוכן קיבל החלטה.

ה-SDK פותר这些问题 עם העברות אוטומטיות (handoffs), guardrails מובנים, וניטור OpenTelemetry. מערכת מרובת סוכנים על Agents SDK מעבדת בקשות פי 3 מהר יותר מסוכן מונוליטי (בהתבסס על benchmark שלנו של 1000 בקשות). חיסכון בקריאות API: עד $30k בשנה בעומס של 10k בקשות ליום.

למה Handoffs חשובים לסקייל?

Handoffs מאפשרים האצלת משימות לסוכנים מתמחים תוך שמירה על קונטקסט השיחה. זה קריטי למערכות production בעומס גבוה: סוכן triage על מודל קל (gpt-4o-mini) מנתב פניות, בעוד סוכנים מורכבים (gpt-4o) מטפלים רק בתחום שלהם. ללא handoffs, כל סוכן חייב לדעת לעשות הכל, מה שמוביל לעלייה ב-latency ובעלויות. בפרויקט שלנו, ארכיטקטורת ה-handoffs הפחיתה את ה-p99 latency ב-40%.

דוגמאות ליישום סוכנים

ה-SDK מותקן עם pip install openai-agents. תיעוד מלא נמצא ב-מאגר הרשמי.

סוכן בסיסי עם כלים

import asyncio
from openai import AsyncOpenAI
from agents import Agent, Runner, function_tool, RunConfig
from agents.models.openai_responses import OpenAIResponsesModel

client = AsyncOpenAI()

@function_tool
def get_weather(city: str, unit: str = "celsius") -> str:
    """Получить текущую погоду в городе.

    Args:
        city: Название города
        unit: Единица температуры (celsius/fahrenheit)
    """
    data = weather_api.get(city=city, unit=unit)
    return f"Погода в {city}: {data['temp']}°, {data['description']}"

@function_tool
def create_calendar_event(
    title: str,
    date: str,
    duration_minutes: int,
    attendees: list[str],
) -> str:
    """Создать событие в корпоративном календаре."""
    event = calendar_api.create(
        title=title,
        date=date,
        duration=duration_minutes,
        attendees=attendees,
    )
    return f"Событие создано: {event['id']}, ссылка: {event['meet_link']}"

assistant = Agent(
    name="Corporate Assistant",
    instructions="""Ты — корпоративный ассистент. Помогай сотрудникам планировать встречи, находить информацию, решать задачи. При необходимости — используй инструменты.""",
    model="gpt-4o",
    tools=[get_weather, create_calendar_event],
)

async def main():
    result = await Runner.run(
        assistant,
        input="Запланируй встречу с командой на завтра в 14:00 на 1 час",
    )
    print(result.final_output)

asyncio.run(main())

Handoffs: העברה בין סוכנים מתמחים

from agents import Agent, handoff, Runner

triage_agent = Agent(
    name="Triage",
    instructions="""Классифицируй запрос и передай соответствующему агенту.
Billing → billing_agent
Technical → tech_agent
General → general_agent""",
    model="gpt-4o-mini",
)

billing_agent = Agent(
    name="Billing Support",
    instructions="Помогай с вопросами выставления счётов, платежами, подписками.",
    model="gpt-4o",
    tools=[get_invoice, process_refund, update_payment_method],
)

tech_agent = Agent(
    name="Technical Support",
    instructions="Решай технические проблемы: API, интеграции, ошибки.",
    model="gpt-4o",
    tools=[check_api_status, get_error_logs, create_bug_report],
)

general_agent = Agent(
    name="General Support",
    instructions="Отвечай на общие вопросы о продукте.",
    model="gpt-4o-mini",
    tools=[search_docs],
)

triage_agent.handoffs = [
    handoff(billing_agent, tool_name_override="transfer_to_billing"),
    handoff(tech_agent, tool_name_override="transfer_to_technical"),
    handoff(general_agent, tool_name_override="transfer_to_general"),
]

result = await Runner.run(
    triage_agent,
    input="Я получил двойное списание за прошлый месяц",
)

אילו סיכונים Guardrails ממתנים?

Guardrails מגנים מפני דליפת PII, תגובות רעילות, וניצול לרעה של המודל. InputGuardrails חוסמים בקשות עם מידע רגיש לפני עיבוד; OutputGuardrails מסננים תגובות לפני שליחה למשתמש. בפיננסים ובריאות, זו דרישה מחייבת. ללא guardrails, המודל עלול לחשוף בטעות נתוני לקוחות — אירוע אחד כזה יכול לגרור קנסות של עד 4% מהמחזור.

Guardrails: פילטרי קלט ופלט

from agents import Agent, InputGuardrail, OutputGuardrail, GuardrailFunctionOutput

async def pii_detection_guardrail(ctx, agent, input) -> GuardrailFunctionOutput:
    pii_check_agent = Agent(
        name="PII Checker",
        instructions="Проверь, содержит ли текст персональные данные (номера карт, паспортов, СНИЛС).",
        model="gpt-4o-mini",
        output_type={"contains_pii": bool, "pii_types": list[str]},
    )
    result = await Runner.run(pii_check_agent, input=input)
    contains_pii = result.final_output.get("contains_pii", False)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=contains_pii,
    )

async def content_safety_guardrail(ctx, agent, output) -> GuardrailFunctionOutput:
    violations = await compliance_checker.check(output)
    return GuardrailFunctionOutput(
        output_info=violations,
        tripwire_triggered=len(violations) > 0,
    )

safe_agent = Agent(
    name="Safe Assistant",
    instructions="Отвечай на вопросы о финансовых продуктах.",
    model="gpt-4o",
    input_guardrails=[InputGuardrail(guardrail_function=pii_detection_guardrail)],
    output_guardrails=[OutputGuardrail(guardrail_function=content_safety_guardrail)],
)

פלט מובנה עם טיפוסים

from pydantic import BaseModel
from typing import Literal

class CustomerAnalysis(BaseModel):
    customer_id: str
    churn_risk: Literal["low", "medium", "high"]
    churn_probability: float
    key_risk_factors: list[str]
    recommended_actions: list[str]
    priority_contact: bool

analysis_agent = Agent(
    name="Churn Analyst",
    instructions="""Анализируй данные клиента и оцени риск оттока. Учитывай: активность за последние 30 дней, NPS, количество обращений в поддержку, использование функций продукта.""",
    model="gpt-4o",
    tools=[get_customer_activity, get_support_history, get_product_usage],
    output_type=CustomerAnalysis,
)

result = await Runner.run(
    analysis_agent,
    input=f"Проанализируй клиента ID: {customer_id}",
)

analysis: CustomerAnalysis = result.final_output
print(f"Риск оттока: {analysis.churn_risk} ({analysis.churn_probability:.0%})")

מה מספק ניטור סוכנים (Tracing)?

Tracing הוא הדרך היחידה לנפות באגים בתרחישים מרובי סוכנים. ה-SDK מתעד כל קריאה, handoff ו-guardrail, כולל latency ו-tokens. אינטגרציית OpenTelemetry שולחת נתונים ל-Jaeger, Grafana או New Relic. ללא tracing, לא תראו היכן התרחש עיכוב או איזה סוכן גרם לשגיאה.

ניטור ומעקב

from agents.tracing import set_tracing_provider
from agents.tracing.opentelemetry import OpenTelemetryTracingProvider
from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter

otlp_exporter = OTLPSpanExporter(endpoint="http://jaeger:4318/v1/traces")
set_tracing_provider(OpenTelemetryTracingProvider(exporter=otlp_exporter))

from agents import trace

async def run_with_trace():
    with trace("customer_support_session"):
        result = await Runner.run(
            triage_agent,
            input="Проблема с подключением к API",
            run_config=RunConfig(
                trace_include_sensitive_data=False,
                workflow_name="customer_support",
            ),
        )
        return result

מקרה מעשי: אוטומציה של הטמעת B2B

לקוח: שירות B2B גדול עם למעלה מ-2000 לקוחות. הטמעה אופיינית ארכה שבועיים ודרשה תיאום של 3 מחלקות: מכירות, אינטגרציה ותמיכה.

ארכיטקטורה:

• מתאם הטמעה (triage): מקבל בקשה, מנתב
• סוכן הגדרת חשבון: הגדרת חשבון, תפקידים, SSO
• סוכן אינטגרציה: סיוע באינטגרציית API, יצירת דוגמאות קוד
• סוכן הדרכה: תוכן הדרכה מותאם אישית
• סוכן הצלחה: מעקב, ניטור מדדי אימוץ

שרשרת handoff: מתאם → הגדרת חשבון → אינטגרציה → הדרכה → הצלחה. תוצאות:

חיסכון בעלויות תמיכה: כ-$30k בשנה.

תהליך היישום ולוח זמנים

שלבי יישום

לוח זמנים כולל: בין 2 ל-4 שבועות בהתאם למורכבות.

לוח זמנים לפי רכיב

• סוכן בסיסי עם כלים: 2–4 ימים
• ארכיטקטורת handoff עם 3–5 סוכנים: 1–2 שבועות
• Guardrails ובדיקות בטיחות: 3–5 ימים
• ניטור ומעקב: 3–5 ימים
• פריסה ל-production: שבוע

העלות מחושבת בנפרד לכל פרויקט.

טעויות עיצוב נפוצות

• שימוש באותו מודל לכל הסוכנים: triage צריך להיות זול (gpt-4o-mini), משימות מורכבות על gpt-4o.
• חוסר ב-output guardrails: PII עלול לדלוף לתגובות. תמיד הוסיפו OutputGuardrail.
• הוראות ארוכות מדי: המודל מאבד פוקוס. חלקו סוכנים לפי התמחות.
• התעלמות מ-tracing: בלתי אפשרי לנפות באגים בתרחישים מרובי סוכנים בלעדיו.

מה כלול בעבודה

• תיעוד ארכיטקטורת סוכנים (מפות סוכנים, דיאגרמות handoff)
• ניטור מוגדר (OpenTelemetry / OpenAI tracing)
• Guardrails מוכנים (PII, בטיחות תוכן, בדיקות מותאמות אישית)
• הדרכת צוות הלקוח (סדנה של יום אחד)
• שבועיים של תמיכה לאחר הפריסה

הבטחות ותמיכה

אנו מבטיחים SLA לתגובת סוכן של p99 < 2 שניות. מהנדסי OpenAI מוסמכים. ניסיון יישום בפיננסים ובריאות. צרו קשר כדי להעריך את התרחיש שלכם — קבלו ייעוץ על ארכיטקטורת סוכנים.