OpenAI의 AI Agent 전략

┌────────────────────────────────────────────────────────────┐
│                    Consumer Products                        │
│   ChatGPT Agent Mode  │  Operator  │  Deep Research        │
├────────────────────────────────────────────────────────────┤
│                    Developer Tools                          │
│     Agents SDK  │  Codex  │  MCP Support                   │
├────────────────────────────────────────────────────────────┤
│                    Responses API                            │
│   web_search │ file_search │ code_interpreter │ computer_use│
├────────────────────────────────────────────────────────────┤
│                    Foundation Models                         │
│   GPT-4.1  │  o3/o4-mini  │  GPT-5/5.2  │  codex-1        │
└────────────────────────────────────────────────────────────┘

from agents import Agent, Runner

# Agent 정의 — instructions로 역할과 행동 규칙 지정
sales_agent = Agent(
    name="Sales Agent",
    instructions="""당신은 영업팀 전문 상담사입니다.
    - 제품 가격, 라이선스, 할인 정책에 대해 정확히 답변하세요.
    - 기술적 질문은 직접 답하지 말고, 기술 상담 에이전트로 넘기세요.
    - 항상 고객의 회사 규모와 요구사항을 먼저 파악하세요.""",
    model="gpt-4.1",
    tools=[pricing_tool, crm_lookup_tool],
)

from agents import Agent, handoff

tech_agent = Agent(
    name="Tech Support Agent",
    instructions="기술적인 질문에 답변합니다. 가격 질문은 영업 에이전트로 넘기세요.",
    tools=[docs_search_tool, code_executor_tool],
    handoffs=[handoff(sales_agent)],  # 영업 에이전트로 핸드오프 가능
)

# Triage Agent — 모든 요청을 분류하여 적절한 에이전트로 라우팅
triage_agent = Agent(
    name="Triage Agent",
    instructions="""사용자 요청을 분석하여 적절한 에이전트로 라우팅하세요:
    - 가격, 라이선스, 할인 → Sales Agent
    - 기술 지원, 버그, 설정 → Tech Support Agent
    - 판단이 어려우면 추가 질문으로 의도를 파악하세요.""",
    handoffs=[
        handoff(sales_agent),
        handoff(tech_agent),
    ],
)

from agents import Agent, InputGuardrail, GuardrailFunctionOutput
from pydantic import BaseModel

class SafetyCheck(BaseModel):
    is_safe: bool
    reason: str

# Guardrail Agent — 입력이 안전한지 판단하는 별도 에이전트
safety_agent = Agent(
    name="Safety Checker",
    instructions="사용자 입력이 안전한지 판단하세요. 개인정보, 욕설, 유해 콘텐츠가 있으면 unsafe로 분류하세요.",
    output_type=SafetyCheck,
)

async def check_safety(ctx, agent, input_text):
    result = await Runner.run(safety_agent, input_text, context=ctx)
    return GuardrailFunctionOutput(
        output_info=result.final_output,
        tripwire_triggered=not result.final_output.is_safe,
    )

# Guardrail을 Agent에 부착
main_agent = Agent(
    name="Main Agent",
    instructions="사용자의 질문에 답변합니다.",
    input_guardrails=[
        InputGuardrail(guardrail_function=check_safety),
    ],
)

from agents import Agent, Runner, trace

# 기본 Tracing — Runner.run()을 호출하면 자동으로 트레이스 생성
result = await Runner.run(triage_agent, "Pro Plan 가격이 얼마인가요?")

# 커스텀 Trace 그룹핑 — 여러 실행을 하나의 워크플로우로 묶기
with trace("customer-support-workflow"):
    classification = await Runner.run(triage_agent, user_message)
    # 이후 실행도 같은 trace에 포함됨

from agents import Agent, Runner, handoff, InputGuardrail, trace
import asyncio

# ── Tool 정의 ──
def lookup_order(order_id: str) -> str:
    """주문 상태를 조회합니다."""
    # 실제로는 DB 호출
    return f"주문 {order_id}: 배송 중 (예상 도착일: 2026-04-02)"

def check_refund_policy(product: str) -> str:
    """환불 정책을 확인합니다."""
    return f"{product}의 환불 가능 기간은 구매 후 30일입니다."

# ── Specialist Agents ──
order_agent = Agent(
    name="Order Agent",
    instructions="주문 조회, 배송 상태 확인을 담당합니다.",
    tools=[lookup_order],
)

refund_agent = Agent(
    name="Refund Agent",
    instructions="환불, 반품 관련 문의를 처리합니다.",
    tools=[check_refund_policy],
)

# ── Triage Agent ──
triage_agent = Agent(
    name="Triage Agent",
    instructions="""고객 문의를 분류하여 적절한 에이전트로 전달하세요:
    - 주문, 배송 → Order Agent
    - 환불, 반품 → Refund Agent
    - 일반 문의는 직접 답변하세요.""",
    handoffs=[
        handoff(order_agent),
        handoff(refund_agent),
    ],
)

# ── 실행 ──
async def main():
    with trace("customer-support"):
        result = await Runner.run(
            triage_agent,
            "주문번호 ORD-2026-1234 배송이 언제 오나요?"
        )
        print(result.final_output)
        # → "주문 ORD-2026-1234: 배송 중 (예상 도착일: 2026-04-02)"

asyncio.run(main())

from openai import OpenAI

client = OpenAI()

# Responses API — 단일 호출로 에이전틱 작업 수행
response = client.responses.create(
    model="gpt-4.1",
    input="2026년 1분기 AI Agent 시장 트렌드를 분석해주세요.",
    tools=[
        {"type": "web_search"},          # 웹 검색 자동 사용
        {"type": "code_interpreter"},     # 데이터 분석 필요 시 코드 실행
    ],
)

# 단일 API 호출이지만, 내부적으로:
# 1) web_search로 최신 정보 수집
# 2) code_interpreter로 데이터 정리/분석
# 3) 최종 분석 보고서 생성
# → 이 모든 것이 하나의 response에 담김
print(response.output_text)

# 모델별 에이전트 — 비용 최적화 패턴
triage = Agent(name="Triage", model="gpt-4.1-nano", ...)      # 라우팅만
classifier = Agent(name="Classifier", model="gpt-4.1-mini", ...) # 분류
analyst = Agent(name="Analyst", model="gpt-4.1", ...)           # 심층 분석

[기존 모델]
User → LLM 추론 → Tool 호출 결정 → Tool 실행 → LLM 응답

[o3/o4-mini]
User → 추론 시작 → (추론 중) Tool 호출 → 결과 반영하며 추론 계속
       → (추론 중) 또 다른 Tool 호출 → 결과 반영 → 최종 응답

# Databricks Model Serving으로 GPT-4.1 호출 (External Model 설정 후)
import mlflow.deployments

client = mlflow.deployments.get_deploy_client("databricks")

response = client.predict(
    endpoint="openai-gpt-4-1",  # External Model endpoint 이름
    inputs={
        "messages": [
            {"role": "system", "content": "당신은 데이터 분석 전문가입니다."},
            {"role": "user", "content": "매출 트렌드를 분석해주세요."},
        ],
        "temperature": 0.1,
    },
)

# 하이브리드 패턴 예시: Agents SDK + Databricks 데이터
from agents import Agent, function_tool
import mlflow.deployments

db_client = mlflow.deployments.get_deploy_client("databricks")

@function_tool
def query_lakehouse(sql: str) -> str:
    """Databricks SQL Warehouse로 데이터를 조회합니다."""
    from databricks import sql as dbsql
    with dbsql.connect(server_hostname="...", http_path="...") as conn:
        cursor = conn.cursor()
        cursor.execute(sql)
        return str(cursor.fetchall())

data_agent = Agent(
    name="Data Analyst",
    instructions="Databricks Lakehouse의 데이터를 분석합니다.",
    model="gpt-4.1",
    tools=[query_lakehouse],
)