CrewAI

import weave
from crewai import Agent, Task, Crew, LLM, Process

# 프로젝트 이름으로 Weave 초기화
weave.init(project_name="crewai_demo")

# 결정론적 출력을 보장하기 위해 temperature를 0으로 설정하여 LLM 생성
llm = LLM(model="gpt-4o-mini", temperature=0)

# 에이전트 생성
researcher = Agent(
    role='Research Analyst',
    goal='Find and analyze the best investment opportunities',
    backstory='Expert in financial analysis and market research',
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

writer = Agent(
    role='Report Writer',
    goal='Write clear and concise investment reports',
    backstory='Experienced in creating detailed financial reports',
    llm=llm,
    verbose=True,
    allow_delegation=False,
)

# 태스크 생성
research_task = Task(
    description='Deep research on the {topic}',
    expected_output='Comprehensive market data including key players, market size, and growth trends.',
    agent=researcher
)

writing_task = Task(
    description='Write a detailed report based on the research',
    expected_output='The report should be easy to read and understand. Use bullet points where applicable.',
    agent=writer
)

# 크루 생성
crew = Crew(
    agents=[researcher, writer],
    tasks=[research_task, writing_task],
    verbose=True,
    process=Process.sequential,
)

# 크루 실행
result = crew.kickoff(inputs={"topic": "AI in material science"})
print(result)

# .... 기존 import 구문 ....
from crewai_tools import SerperDevTool

# 에이전트에게 툴 제공
researcher = Agent(
    role='Research Analyst',
    goal='Find and analyze the best investment opportunities',
    backstory='Expert in financial analysis and market research',
    llm=llm,
    verbose=True,
    allow_delegation=False,
    tools=[SerperDevTool()],
)

# .... 기존 코드 ....

import weave
# 프로젝트 이름으로 Weave 초기화
weave.init("crewai_demo")

from crewai.flow.flow import Flow, listen, router, start
from litellm import completion


class CustomerFeedbackFlow(Flow):
    model = "gpt-4o-mini"

    @start()
    def fetch_feedback(self):
        print("Fetching customer feedback")
        # 실제 시나리오에서는 이를 API 호출로 대체할 수 있습니다.
        # 이 예제에서는 고객 피드백을 시뮬레이션합니다.
        feedback = (
            "I had a terrible experience with the product. "
            "It broke after one use and customer service was unhelpful."
        )
        self.state["feedback"] = feedback
        return feedback

    @router(fetch_feedback)
    def analyze_feedback(self, feedback):
        # 언어 모델을 사용하여 감성 분석
        prompt = (
            f"Analyze the sentiment of this customer feedback and "
            "return only 'positive' or 'negative':\n\n"
            f"Feedback: {feedback}"
        )
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        sentiment = response["choices"][0]["message"]["content"].strip().lower()
        # 응답이 모호한 경우 기본값을 negative로 설정
        if sentiment not in ["positive", "negative"]:
            sentiment = "negative"
        return sentiment

    @listen("positive")
    def handle_positive_feedback(self):
        # 긍정적인 피드백에 대한 감사 메시지 생성
        prompt = "Generate a thank you message for a customer who provided positive feedback."
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        thank_you_message = response["choices"][0]["message"]["content"].strip()
        self.state["response"] = thank_you_message
        return thank_you_message

    @listen("negative")
    def handle_negative_feedback(self):
        # 부정적인 피드백에 대한 사과 메시지와 서비스 개선 약속 생성
        prompt = (
            "Generate an apology message to a customer who provided negative feedback and offer assistance or a solution."
        )
        response = completion(
            model=self.model,
            messages=[{"role": "user", "content": prompt}],
        )
        apology_message = response["choices"][0]["message"]["content"].strip()
        self.state["response"] = apology_message
        return apology_message

# Flow 인스턴스화 및 실행
flow = CustomerFeedbackFlow()
result = flow.kickoff()
print(result)

# .... 기존 import 및 weave 초기화 ....

# 가드레일 함수에 `@weave.op()` 데코레이터 적용
@weave.op(name="guardrail-validate_blog_content")
def validate_blog_content(result: TaskOutput) -> Tuple[bool, Any]:
    # raw 문자열 결과 가져오기
    result = result.raw

    """Blog 콘텐츠가 요구 사항을 충족하는지 검증합니다."""
    try:
        # 단어 수 확인
        word_count = len(result.split())

        if word_count > 200:
            return (False, {
                "error": "Blog content exceeds 200 words",
                "code": "WORD_COUNT_ERROR",
                "context": {"word_count": word_count}
            })

        # 추가 검증 로직
        return (True, result.strip())
    except Exception as e:
        return (False, {
            "error": "Unexpected error during validation",
            "code": "SYSTEM_ERROR"
        })


# .... 기존 에이전트 및 research analyst 태스크 ....

writing_task = Task(
    description='Write a detailed report based on the research under 200 words',
    expected_output='The report should be easy to read and understand. Use bullet points where applicable.',
    agent=writer,
    guardrail=validate_blog_content,
)

# .... 기존 크루 실행 코드 ....