CrewAI, OpenAI Agents SDK & AutoGen

from crewai import Agent, Task, Crew, Process
from crewai_tools import SerperDevTool

# 도구 정의
search_tool = SerperDevTool()

# Agent 1: 리서처
researcher = Agent(
    role="시니어 데이터 리서처",
    goal="Databricks의 최신 기술 동향을 조사하여 핵심 인사이트를 도출한다",
    backstory=(
        "당신은 10년 경력의 데이터 플랫폼 분석가입니다. "
        "기술 트렌드를 빠르게 파악하고 비즈니스 임팩트를 연결하는 데 탁월합니다."
    ),
    tools=[search_tool],
    verbose=True,
)

# Agent 2: 보고서 작성자
writer = Agent(
    role="테크 라이터",
    goal="리서처의 조사 결과를 기반으로 경영진용 보고서를 작성한다",
    backstory=(
        "당신은 기술 콘텐츠를 비기술 임원이 이해할 수 있도록 "
        "변환하는 전문 테크 라이터입니다."
    ),
    verbose=True,
)

# Task 1: 조사
research_task = Task(
    description="2025년 Databricks의 주요 제품 업데이트와 시장 동향을 조사하세요.",
    expected_output="핵심 업데이트 5개와 각각의 비즈니스 임팩트 분석",
    agent=researcher,
)

# Task 2: 보고서 작성
writing_task = Task(
    description="리서치 결과를 기반으로 경영진용 1페이지 브리핑을 작성하세요.",
    expected_output="제목, 핵심 요약 3줄, 상세 분석 5개 항목의 구조화된 보고서",
    agent=writer,
)

# Crew 구성 및 실행
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    process=Process.sequential,  # 순차 실행: 리서치 -> 작성
    verbose=True,
)

result = crew.kickoff()
print(result)

from openai import agents

# 전문 Agent 정의
billing_agent = agents.Agent(
    name="billing_agent",
    instructions=(
        "당신은 청구/결제 전문 상담원입니다. "
        "청구 관련 질문에만 답변하세요. "
        "기술 지원 질문은 tech_agent로 핸드오프하세요."
    ),
    model="gpt-4o",
)

tech_agent = agents.Agent(
    name="tech_agent",
    instructions=(
        "당신은 기술 지원 전문 상담원입니다. "
        "기술적인 문제에 대해 답변하세요. "
        "청구 관련 질문은 billing_agent로 핸드오프하세요."
    ),
    model="gpt-4o",
)

# Triage Agent -- 첫 진입점
triage_agent = agents.Agent(
    name="triage_agent",
    instructions=(
        "당신은 고객 문의를 분류하는 접수 담당입니다. "
        "고객의 질문을 파악하여 적절한 전문 상담원에게 연결하세요."
    ),
    model="gpt-4o",
    handoffs=[billing_agent, tech_agent],  # 핸드오프 대상 지정
)

# 실행
result = agents.run(
    triage_agent,
    messages=[{"role": "user", "content": "지난달 청구서가 이상합니다"}],
)

# triage_agent -> billing_agent로 자동 핸드오프됨
print(result.final_output)

from openai.agents import guardrail, GuardrailFunctionOutput

@guardrail
def no_pii_guardrail(text: str) -> GuardrailFunctionOutput:
    """개인정보(PII)가 포함된 입력을 차단"""
    import re
    # 주민등록번호 패턴 체크
    if re.search(r'\d{6}-\d{7}', text):
        return GuardrailFunctionOutput(
            output_info="주민등록번호가 감지되었습니다.",
            tripwire_triggered=True,  # 실행 중단
        )
    return GuardrailFunctionOutput(
        output_info="안전한 입력입니다.",
        tripwire_triggered=False,
    )

agent = agents.Agent(
    name="safe_agent",
    instructions="고객 문의에 답변하세요.",
    model="gpt-4o",
    input_guardrails=[no_pii_guardrail],  # 입력 가드레일 적용
)

from autogen import AssistantAgent, UserProxyAgent, config_list_from_json

# LLM 설정
config_list = [{"model": "gpt-4o", "api_key": "YOUR_API_KEY"}]

# Agent 1: 코드 작성자
coder = AssistantAgent(
    name="coder",
    system_message=(
        "당신은 시니어 Python 개발자입니다. "
        "요청받은 기능을 깔끔한 Python 코드로 구현하세요. "
        "코드 블록으로 감싸서 응답하세요."
    ),
    llm_config={"config_list": config_list},
)

# Agent 2: 코드 리뷰어
reviewer = AssistantAgent(
    name="reviewer",
    system_message=(
        "당신은 코드 리뷰 전문가입니다. "
        "coder가 작성한 코드를 리뷰하고 개선점을 제안하세요. "
        "보안, 성능, 가독성 관점에서 평가하세요."
    ),
    llm_config={"config_list": config_list},
)

# Agent 3: 사용자 대리 (코드 실행 가능)
user_proxy = UserProxyAgent(
    name="user_proxy",
    human_input_mode="NEVER",  # 자동 실행 (사람 개입 없음)
    code_execution_config={"work_dir": "coding_output"},
    max_consecutive_auto_reply=3,
)

# 대화 시작: user_proxy가 coder에게 요청
user_proxy.initiate_chat(
    coder,
    message="Delta Lake 테이블의 변경 이력을 조회하는 Python 함수를 작성해주세요.",
)