16. 실전: 데이터사이언스 크루

🎯 이 장의 목표

커스텀 도구(BaseTool)를 만들어 에이전트에 코드 실행 능력을 준다
에이전트가 코드를 실행해 데이터를 다루는 워크플로를 이해한다
ML 회귀 파이프라인을 크루로 자동화하는 관점을 잡는다
구조화 출력과 task 컨텍스트 전달로 단계를 잇는 법을 본다

비유로 시작하기: 분석 도구를 든 데이터 팀

지금까지의 에이전트는 글을 쓰고 정보를 정리했습니다. 하지만 데이터 분석은 다릅니다. 실제로 코드를 돌려 CSV를 읽고, 결측치를 처리하고, 모델을 학습시키고, 성능을 평가해야 합니다. 즉 에이전트에게 파이썬 실행 환경이라는 도구를 쥐여줘야 합니다.

강의의 CrewAI 데이터사이언스 섹션이 바로 이것입니다. 회귀 모델(regression)을 학습·평가하는 ML 파이프라인을, 코드를 실행할 줄 아는 에이전트 팀으로 자동화합니다.

📌 핵심

참고: 이 장의 목표는 ML 회귀 이론 자체가 아니라 "CrewAI로 ML 워크플로를 자동화하는 방법"입니다. 회귀·랜덤포레스트·특성 중요도 같은 ML 개념, 그리고 뒤에 나오는 scikit-learn(파이썬의 대표적 머신러닝 라이브러리)은 강의 영상과 부록의 추가 학습 링크를 참고하세요. 여기서는 그 작업들을 에이전트 팀에 어떻게 맡기는지에 집중합니다.

커스텀 도구 만들기: BaseTool

5장과 7장에서 도구의 원리를 배웠습니다. CrewAI에서 커스텀 도구는 BaseTool을 상속해 만듭니다(구버전과 달라진 점이니 주의).

PYTHON

from crewai.tools import BaseTool
from pydantic import BaseModel, Field
from typing import Type

# 도구의 입력 스키마
class CodeInput(BaseModel):
    code: str = Field(..., description="실행할 파이썬 코드")

class PythonExecutorTool(BaseTool):
    name: str = "python_executor"
    description: str = "파이썬 코드를 실행하고 결과를 반환한다."
    args_schema: Type[BaseModel] = CodeInput

    def _run(self, code: str) -> str:
        # 실전에서는 반드시 샌드박스에서 실행 (보안!)
        local_vars = {}
        try:
            exec(code, {}, local_vars)
            return str(local_vars.get("result", "실행 완료"))
        except Exception as e:
            return f"에러: {e}"

name·description·args_schema·_run이 핵심입니다. description은 5장에서 강조했듯 모델이 "언제 이 도구를 쓸지" 판단하는 근거입니다.

⚠️ 보안 경고: 위 예시의 exec는 학습용 단순화입니다. (exec는 문자열로 된 파이썬 코드를 그 자리에서 실행하는 내장 함수입니다.) 실제로 LLM이 생성한 코드를 그대로 실행하면 매우 위험합니다. 프로덕션에서는 반드시 격리된 샌드박스/컨테이너에서 실행해야 합니다. 여기서 샌드박스는 바깥 시스템과 차단된 안전한 실행 공간을 뜻하고, 컨테이너(예: Docker)는 프로그램을 독립된 환경에 격리해 실행하는 기술로 그런 샌드박스를 만드는 대표적 수단입니다. CrewAI는 이를 위한 코드 실행 도구를 별도로 제공하니, 직접 exec를 쓰지 마세요. (보안은 8부에서 더 다룹니다.)

📌 핵심: 강의의 NotebookCodeExecutor 같은 도구가 바로 이 역할입니다. 에이전트가 직접 코드를 실행해 데이터를 다루게 해주되, 안전한 환경에서 돌리는 것이 관건입니다.

데이터사이언스 크루 설계

ML 파이프라인을 역할별로 나눕니다. 강의의 회귀 워크플로를 팀으로 옮기면 이렇게 됩니다.

flowchart TB
    classDef agent fill:#80DEEA,stroke:#00838F,color:#000
    classDef tool fill:#90CAF9,stroke:#1565C0,color:#000
    classDef result fill:#A5D6A7,stroke:#2E7D32,color:#000

    DE[데이터 엔지니어 에이전트<br/>로드·정제·결측치 처리]:::agent
    ML[ML 엔지니어 에이전트<br/>모델 학습·튜닝]:::agent
    EV[평가 에이전트<br/>성능 측정·해석]:::agent
    EX[(코드 실행 도구)]:::tool
    R[최종 리포트]:::result

    DE -.코드 실행.-> EX
    ML -.코드 실행.-> EX
    EV -.코드 실행.-> EX
    DE --> ML --> EV --> R

PYTHON

from crewai import Agent, Task, Crew, Process

executor = PythonExecutorTool()  # 위에서 만든 도구 (실전은 샌드박스)

data_engineer = Agent(
    role="데이터 엔지니어",
    goal="CSV를 로드하고 결측치를 처리해 분석 가능한 데이터로 만든다",
    backstory="당신은 지저분한 데이터를 깔끔하게 정리하는 전문가입니다.",
    tools=[executor],
    verbose=True,
)

ml_engineer = Agent(
    role="ML 엔지니어",
    goal="정제된 데이터로 회귀 모델을 학습한다",
    backstory="당신은 scikit-learn으로 모델을 학습·튜닝해 왔습니다.",
    tools=[executor],
    verbose=True,
)

evaluator = Agent(
    role="평가 분석가",
    goal="모델 성능을 측정하고 결과를 해석해 보고한다",
    backstory="당신은 지표를 비즈니스 언어로 풀어내는 전문가입니다.",
    tools=[executor],
    verbose=True,
)

Task 연결: 컨텍스트 전달

각 단계의 결과가 다음 단계로 전달되어야 합니다. CrewAI는 context로 task 간 의존성을 명시합니다.

PYTHON

prep_task = Task(
    description="data.csv를 로드하고 결측치를 처리한 뒤 요약 통계를 출력하라.",
    expected_output="정제된 데이터의 형태와 요약 통계",
    agent=data_engineer,
)

train_task = Task(
    description="정제된 데이터로 회귀 모델을 학습하라.",
    expected_output="학습된 모델과 학습 과정 요약",
    agent=ml_engineer,
    context=[prep_task],   # 앞 task 결과를 입력으로
)

eval_task = Task(
    description="학습된 모델의 성능(R², RMSE 등)을 평가하고 해석하라.",
    expected_output="성능 지표와 비즈니스 관점의 해석 보고서",
    agent=evaluator,
    context=[train_task],
)

crew = Crew(
    agents=[data_engineer, ml_engineer, evaluator],
    tasks=[prep_task, train_task, eval_task],
    process=Process.sequential,
)
result = crew.kickoff()
print(result)

context=[prep_task]는 "이 task는 prep_task의 결과를 입력으로 받는다"는 뜻입니다. 14장에서 Sequential이 앞 결과를 자동 전달한다고 했는데, context로 명시적 의존성을 지정하면 더 분명하고 안정적입니다.

📌 핵심: 데이터사이언스 크루의 본질은 "코드 실행 도구를 가진 에이전트들이 단계별로 협업하는 것"입니다. 사람 데이터 팀의 분업(엔지니어 → 모델러 → 평가자)을 그대로 옮긴 구조입니다.

구조화 출력으로 안정성 높이기

평가 결과처럼 후속 처리가 필요한 값은 구조화하면 좋습니다. CrewAI도 Pydantic 출력을 지원합니다.

PYTHON

from pydantic import BaseModel

class EvalResult(BaseModel):
    r2_score: float
    rmse: float
    summary: str

eval_task = Task(
    description="모델 성능을 평가하라.",
    expected_output="R², RMSE, 요약",
    agent=evaluator,
    output_pydantic=EvalResult,   # 구조화 출력
    context=[train_task],
)

7장(OpenAI SDK의 output_type)에서 본 것과 같은 아이디어입니다. 결과를 .r2_score처럼 안전하게 다룰 수 있습니다.

💡 팁: 데이터사이언스 크루는 토큰을 많이 씁니다(코드+데이터+결과가 오감). 단순 정제 같은 단계는 빠르고 싼 모델로, 해석·보고 같은 단계는 강한 모델로 모델을 차등 배치하면 비용을 아낍니다(8부에서 상세히).

💡 실습 아이디어(강의의 의사결정나무 확장 대응): train_task의 description을 "회귀 대신 의사결정나무를 학습하라"로 바꿔 보세요. 같은 크루 구조에서 task 지시만 바꾸면 다른 모델 파이프라인이 됩니다. 에이전트 팀의 재사용성을 체감할 수 있습니다.

이 장에서 배운 것

CrewAI 커스텀 도구는 BaseTool을 상속해 만들며, _run에 실제 로직을 둔다.
LLM이 생성한 코드는 반드시 격리된 샌드박스에서 실행해야 한다. 직접 exec 금지.
데이터사이언스 크루는 코드 실행 도구를 가진 에이전트들이 엔지니어→모델러→평가자로 분업한다.
context로 task 간 의존성을 명시하고, output_pydantic으로 결과를 구조화해 안정성을 높인다.

✍️ 확인 문제

CrewAI에서 커스텀 도구를 만들 때 반드시 상속해야 하는 클래스는 무엇이며, 어떤 메서드에 실제 로직을 두는가?
LLM이 생성한 파이썬 코드를 exec로 바로 실행하는 것이 위험한 이유와, 올바른 대안은?
모델 학습 task가 데이터 정제 task의 결과를 확실히 입력받게 하려면 Task에 무엇을 지정하는가?

이전 장: 15. CrewAI Flows
다음 부: 6부 · MCP — 도구를 표준 방식으로 연결하는 프로토콜을 배웁니다.

← 이전15. Flows: 이벤트 기반 프로덕션 오케스트레이션다음 →17. MCP 입문: AI를 위한 USB-C