Crews

from crewai import Agent, Crew, Task, Process
from crewai.project import CrewBase, agent, task, crew, before_kickoff, after_kickoff


@CrewBase
class YourCrewName:
    """여러분의 크루에 대한 설명"""

    # YAML 설정 파일 경로
    # YAML로 정의된 에이전트와 태스크 예제는 다음 링크를 참고하세요:
    # - 태스크: https://docs.crewai.com/concepts/tasks#yaml-configuration-recommended
    # - 에이전트: https://docs.crewai.com/concepts/agents#yaml-configuration-recommended
    agents_config = 'config/agents.yaml' 
    tasks_config = 'config/tasks.yaml' 

    @before_kickoff
    def prepare_inputs(self, inputs):
        # 크루 시작 전 입력값 수정
        inputs['additional_data'] = "추가 정보"
        return inputs

    @after_kickoff
    def process_output(self, output):
        # 크루 완료 후 출력값 처리
        output.raw += "\n크루 완료 후 처리됨."
        return output

    @agent
    def agent_one(self) -> Agent:
        return Agent(
            config=self.agents_config['agent_one'],
            verbose=True
        )

    @agent
    def agent_two(self) -> Agent:
        return Agent(
            config=self.agents_config['agent_two'],
            verbose=True
        )

    @task
    def task_one(self) -> Task:
        return Task(
            config=self.tasks_config['task_one']
        )

    @task
    def task_two(self) -> Task:
        return Task(
            config=self.tasks_config['task_two']
        )

    @crew
    def crew(self) -> Crew:
        return Crew(
            agents=self.agents,  # @agent 데코레이터로 자동 수집
            tasks=self.tasks,    # @task 데코레이터로 자동 수집
            process=Process.sequential,
            verbose=True,
        )

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

class YourCrewName:
    def agent_one(self) -> Agent:
        return Agent(
            role="데이터 분석가",
            goal="시장 내 데이터 트렌드 분석",
            backstory="경제학 배경을 가진 경험 많은 데이터 분석가",
            verbose=True,
            tools=[YourCustomTool()]
        )

    def agent_two(self) -> Agent:
        return Agent(
            role="시장 조사원",
            goal="시장 동향에 대한 정보 수집",
            backstory="세부 사항에 민감한 성실한 연구원",
            verbose=True
        )

    def task_one(self) -> Task:
        return Task(
            description="최근 시장 데이터를 수집하고 트렌드를 파악한다.",
            expected_output="시장의 주요 트렌드를 요약한 보고서",
            agent=self.agent_one()
        )

    def task_two(self) -> Task:
        return Task(
            description="시장 동향에 영향을 미치는 요인을 조사한다.",
            expected_output="시장에 영향을 미치는 요인 분석",
            agent=self.agent_two()
        )

    def crew(self) -> Crew:
        return Crew(
            agents=[self.agent_one(), self.agent_two()],
            tasks=[self.task_one(), self.task_two()],
            process=Process.sequential,
            verbose=True
        )

# 크루 실행 예제
crew = Crew(
    agents=[research_agent, writer_agent],
    tasks=[research_task, write_article_task],
    verbose=True
)

crew_output = crew.kickoff()

# 크루 출력에 접근
print(f"원본 출력: {crew_output.raw}")
if crew_output.json_dict:
    print(f"JSON 출력: {json.dumps(crew_output.json_dict, indent=2)}")
if crew_output.pydantic:
    print(f"Pydantic 출력: {crew_output.pydantic}")
print(f"작업 출력: {crew_output.tasks_output}")
print(f"토큰 사용량: {crew_output.token_usage}")

# 크루 로그 저장
crew = Crew(output_log_file = True)  # 로그가 logs.txt로 저장됨
crew = Crew(output_log_file = file_name)  # 로그가 file_name.txt로 저장됨
crew = Crew(output_log_file = file_name.txt)  # 로그가 file_name.txt로 저장됨
crew = Crew(output_log_file = file_name.json)  # 로그가 file_name.json로 저장됨

# 크루의 사용 메트릭에 접근
crew = Crew(agents=[agent1, agent2], tasks=[task1, task2])
crew.kickoff()
print(crew.usage_metrics)

# 크루의 작업 실행 시작
result = my_crew.kickoff()
print(result)

# 크루의 작업 실행 시작
result = my_crew.kickoff()
print(result)

# kickoff_for_each 사용 예제
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
results = my_crew.kickoff_for_each(inputs=inputs_array)
for result in results:
    print(result)

# kickoff_async 사용 예제
inputs = {'topic': 'AI in healthcare'}
async_result = my_crew.kickoff_async(inputs=inputs)
print(async_result)

# kickoff_for_each_async 사용 예제
inputs_array = [{'topic': 'AI in healthcare'}, {'topic': 'AI in finance'}]
async_results = my_crew.kickoff_for_each_async(inputs=inputs_array)
for async_result in async_results:
    print(async_result)

crewai log-tasks-outputs

crewai replay -t <task_id>