데이터 엔지니어링

Framework 별 간단한 Agent 구현

1) Langchain

"""
LangChain 커스텀 툴 - 에러 수정 버전

설치:
pip install langchain langchain-ollama
"""

from langchain.tools import tool
from langchain_ollama import OllamaLLM  # 새로운 import
from langchain.agents import AgentExecutor, create_react_agent
from langchain.prompts import PromptTemplate
from langchain import hub


# 간단한 덧셈 툴
@tool
def add_numbers(a: int, b: int) -> int:
    """두 숫자를 더합니다. a와 b는 정수여야 합니다."""
    result = a + b
    print(f"[툴 실행] {a} + {b} = {result}")
    return result


@tool
def multiply_numbers(a: int, b: int) -> int:
    """두 숫자를 곱합니다. a와 b는 정수여야 합니다."""
    result = a * b
    print(f"[툴 실행] {a} × {b} = {result}")
    return result


def main():
    print("=" * 70)
    print("LangChain 툴 연동 예제 (수정 버전)")
    print("=" * 70)
    
    # 툴 목록
    my_tools = [add_numbers, multiply_numbers]
    
    print("\n등록된 툴:")
    for tool in my_tools:
        print(f"  • {tool.name}: {tool.description}")
    
    # 1. 새로운 Ollama 클래스 사용
    llm = OllamaLLM(
        model="llama3.1:8b",
        temperature=0
    )
    
    # 2. PromptTemplate을 올바르게 생성
    # 방법 A: 직접 PromptTemplate 생성
    prompt = PromptTemplate.from_template("""
Answer the following questions as best you can. You have access to the following tools:

{tools}

Use the following format:

Question: the input question you must answer
Thought: you should always think about what to do
Action: the action to take, should be one of [{tool_names}]
Action Input: the input to the action in JSON format, e.g. {{"arg1": value1, "arg2": value2}}
Observation: the result of the action
... (this Thought/Action/Action Input/Observation can repeat N times)
Thought: I now know the final answer
Final Answer: the final answer to the original input question

IMPORTANT: Action Input must be valid JSON format with double quotes.
Example: {{"a": 5, "b": 3}}

Begin!

Question: {input}
Thought: {agent_scratchpad}
""")
    
    # 방법 B: LangChain Hub에서 기본 ReAct 프롬프트 가져오기 (권장)
    # prompt = hub.pull("hwchase17/react")
    
    # 3. Agent 생성
    agent = create_react_agent(
        llm=llm,
        tools=my_tools,
        prompt=prompt  # PromptTemplate 객체를 전달
    )
    
    # 4. AgentExecutor 생성
    agent_executor = AgentExecutor(
        agent=agent,
        tools=my_tools,
        verbose=True,
        handle_parsing_errors=True,
        max_iterations=5
    )
    
    # 5. 테스트 질문들
    questions = [
        "5와 3을 더하면?",
        "10과 7을 곱하면?",
        "12 더하기 8은 얼마인가요?",
    ]
    
    print("\n" + "=" * 70)
    print("Agent 실행 테스트")
    print("=" * 70)
    
    for i, question in enumerate(questions, 1):
        print(f"\n[질문 {i}] {question}")
        print("-" * 70)
        try:
            result = agent_executor.invoke({"input": question})
            print(f"\n✅ 최종 답변: {result['output']}")
        except Exception as e:
            print(f"❌ 오류: {e}")
        print("=" * 70)


# 간단한 툴 직접 테스트 (Agent 없이)
def simple_test():
    print("\n" + "=" * 70)
    print("툴 직접 테스트 (Agent 없이)")
    print("=" * 70)
    
    # 직접 호출
    result1 = add_numbers.invoke({"a": 10, "b": 20})
    print(f"\n결과 1: {result1}")
    
    result2 = multiply_numbers.invoke({"a": 5, "b": 6})
    print(f"결과 2: {result2}")
    
    print("\n" + "=" * 70)


if __name__ == "__main__":
    # 1. 먼저 툴만 간단히 테스트
    simple_test()
    
    print("\n\n")
    
    # 2. Agent와 함께 사용
    try:
        main()
    except Exception as e:
        print(f"\n❌ Agent 실행 오류: {e}")
        print("\n💡 Ollama가 실행 중인지 확인하세요: ollama serve")
        print("💡 모델이 설치되어 있는지 확인하세요: ollama pull llama3.1:8b")

2) Strands

from strands import Agent
from strands.models.ollama import OllamaModel
from strands import tool


@tool
def add_numbers(a:int, b:int) -> int:
    """
    add two numbers together
    Args:
        a: first number
        b: second number

    Returns:
        sum of a and b
    """
    return a+b

model = OllamaModel(
    model_id="llama3.1:8b",
    host="http://localhost:11434",
    temperature=0.0)

agent = Agent(model=model, tools=[add_numbers])

response = agent("1+2")
print(response)

3) crawAI

"""
CrewAI + Ollama를 사용한 커스텀 툴 예제

설치:
pip install crewai crewai-tools langchain-ollama
"""

from crewai import Agent, Task, Crew, Process
from crewai.tools import BaseTool  
from crewai import LLM
from typing import Type
from pydantic import BaseModel, Field


# 1. 덧셈 툴 생성
class AddNumbersInput(BaseModel):
    """덧셈 툴의 입력 스키마"""
    a: int = Field(..., description="첫 번째 숫자")
    b: int = Field(..., description="두 번째 숫자")


class AddNumbersTool(BaseTool):
    name: str = "add_numbers"
    description: str = "두 개의 숫자를 더합니다. 덧셈 계산이 필요할 때 사용하세요."
    args_schema: Type[BaseModel] = AddNumbersInput
    
    def _run(self, a: int, b: int) -> str:
        """덧셈 실행"""
        result = a + b
        return f"{a} + {b} = {result}"


def main():
    print("=" * 70)
    print("CrewAI + Ollama 3.1:8b + 커스텀 툴 예제")
    print("=" * 70)
    
    # Ollama 3.1:8b 모델 설정
    llm = LLM(
        model="ollama/llama3.1:8b",  # ⚠️ "ollama/" prefix 필수!
        base_url="http://localhost:11434"
    )
    # 툴 인스턴스 생성
    add_tool = AddNumbersTool()
    
    print("\n✅ 등록된 툴:")
    print(f"  • {add_tool.name}: {add_tool.description}")
    
    # Agent 생성 - 수학 전문가
    math_expert = Agent(
        role="수학 계산 전문가",
        goal="사용자의 계산 요청을 정확하게 처리합니다",
        backstory="""당신은 숫자 계산에 능숙한 수학 전문가입니다. 
        주어진 툴을 사용하여 정확한 계산 결과를 제공합니다.""",
        tools=[add_tool],
        llm=llm,
        verbose=True,
        allow_delegation=False
    )
    
    # Task 1: 덧셈 작업
    task1 = Task(
        description="15와 27을 더한 결과를 계산하세요.",
        expected_output="두 숫자의 덧셈 결과를 명확하게 제시",
        agent=math_expert
    )

    
    # Crew 생성 및 실행
        agents=[math_expert],
        tasks=[task1],
        process=Process.sequential,  # 순차적 실행
        verbose=True
    )
    
    print("\n" + "=" * 70)
    print("🚀 Crew 실행 시작")
    print("=" * 70)
    
    try:
        result = crew.kickoff()
        
        print("\n" + "=" * 70)
        print("✅ 최종 결과")
        print("=" * 70)
        print(result)
        
    except Exception as e:
        print(f"\n❌ 오류 발생: {e}")
        import traceback
        traceback.print_exc()


# 간단한 툴 테스트
def simple_tool_test():
    print("\n" + "=" * 70)
    print("툴 직접 테스트")
    print("=" * 70)
    
    add_tool = AddNumbersTool()
    
    # 덧셈 테스트
    result1 = add_tool._run(a=10, b=20)
    print(f"\n덧셈 결과: {result1}")

결론