1️⃣ 도구 정의(Tool Definition)

AI에게 "이런 도구가 있어!"라고 알려주는 단계야. 각 도구의 이름, 설명, 필요한 입력값(파라미터)을 명확하게 정의해줘야 해.

{
  "name": "get_weather",
  "description": "특정 도시의 현재 날씨 정보를 가져옵니다",
  "parameters": {
    "city": {
      "type": "string",
      "description": "날씨를 확인할 도시 이름"
    }
  }
}

기본 도구 클래스 정의

from typing import Dict, Any, Callable
import json

class Tool:
    def __init__(self, name: str, description: str, 
                 function: Callable, parameters: Dict):
        self.name = name
        self.description = description
        self.function = function
        self.parameters = parameters
    
    def to_schema(self) -> Dict:
        """AI가 이해할 수 있는 스키마 형식으로 변환"""
        return {
            "name": self.name,
            "description": self.description,
            "parameters": self.parameters
        }
    
    def execute(self, **kwargs) -> Any:
        """도구 실행"""
        return self.function(**kwargs)

구체적인 도구 구현

# 날씨 조회 도구
def get_weather_impl(city: str) -> Dict:
    # 실제로는 API 호출하겠지만, 여기서는 시뮬레이션
    weather_data = {
        "서울": {"temp": 18, "condition": "맑음"},
        "부산": {"temp": 20, "condition": "흐림"}
    }
    return weather_data.get(city, {"temp": 0, "condition": "정보없음"})

weather_tool = Tool(
    name="get_weather",
    description="특정 도시의 현재 날씨를 조회합니다",
    function=get_weather_impl,
    parameters={
        "type": "object",
        "properties": {
            "city": {
                "type": "string",
                "description": "날씨를 확인할 도시 이름"
            }
        },
        "required": ["city"]
    }
)

# 계산기 도구
def calculator_impl(expression: str) -> float:
    # 안전한 계산을 위해 eval 대신 ast 사용 권장
    try:
        return eval(expression)
    except:
        return "계산 오류"

calculator_tool = Tool(
    name="calculator",
    description="수학 계산을 수행합니다",
    function=calculator_impl,
    parameters={
        "type": "object",
        "properties": {
            "expression": {
                "type": "string",
                "description": "계산할 수식 (예: '2 + 2', '10 * 5')"
            }
        },
        "required": ["expression"]
    }
)

도구 관리자 (Tool Manager)

class ToolManager:
    def __init__(self):
        self.tools: Dict[str, Tool] = {}
    
    def register_tool(self, tool: Tool):
        """도구 등록"""
        self.tools[tool.name] = tool
        print(f"✅ 도구 등록됨: {tool.name}")
    
    def get_tools_schema(self) -> list:
        """모든 도구의 스키마를 AI에게 전달할 형식으로 반환"""
        return [tool.to_schema() for tool in self.tools.values()]
    
    def execute_tool(self, tool_name: str, arguments: Dict) -> Any:
        """도구 실행"""
        if tool_name not in self.tools:
            return {"error": f"도구 '{tool_name}'를 찾을 수 없습니다"}
        
        tool = self.tools[tool_name]
        try:
            result = tool.execute(**arguments)
            return {"success": True, "result": result}
        except Exception as e:
            return {"success": False, "error": str(e)}

# 사용 예시
manager = ToolManager()
manager.register_tool(weather_tool)
manager.register_tool(calculator_tool)

# 도구 실행
result = manager.execute_tool("get_weather", {"city": "서울"})
print(result)  # {'success': True, 'result': {'temp': 18, 'condition': '맑음'}}

from langchain.agents import initialize_agent, Tool
from langchain.agents import AgentType
from langchain.llms import OpenAI

# 도구 정의
tools = [
    Tool(
        name="Calculator",
        func=lambda x: eval(x),
        description="수학 계산에 유용합니다. 입력은 수식이어야 합니다."
    ),
    Tool(
        name="Search",
        func=search_function,  # 실제 검색 함수
        description="최신 정보를 검색할 때 사용합니다."
    )
]

# 에이전트 초기화
llm = OpenAI(temperature=0)
agent = initialize_agent(
    tools, 
    llm, 
    agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION,
    verbose=True
)

# 실행
response = agent.run("2024년 올림픽은 어디서 열리나요? 그리고 123 * 456을 계산해주세요")
print(response)

🎯 LangChain의 장점:
✅ 다양한 LLM 지원 (OpenAI, Anthropic, Hugging Face 등)
✅ 풍부한 내장 도구들
✅ 체인(Chain) 구성이 직관적
✅ 커뮤니티가 활발하고 문서가 잘 되어있음

import openai

# 함수 정의
functions = [
    {
        "name": "get_current_weather",
        "description": "특정 위치의 현재 날씨를 가져옵니다",
        "parameters": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "도시 이름, 예: Seoul"
                },
                "unit": {
                    "type": "string",
                    "enum": ["celsius", "fahrenheit"]
                }
            },
            "required": ["location"]
        }
    }
]

# API 호출
response = openai.ChatCompletion.create(
    model="gpt-4",
    messages=[
        {"role": "user", "content": "서울 날씨 어때?"}
    ],
    functions=functions,
    function_call="auto"
)

# 응답 처리
message = response["choices"][0]["message"]

if message.get("function_call"):
    function_name = message["function_call"]["name"]
    arguments = json.loads(message["function_call"]["arguments"])
    
    # 실제 함수 실행
    if function_name == "get_current_weather":
        weather_result = get_current_weather(**arguments)
        
        # 결과를 다시 AI에게 전달
        second_response = openai.ChatCompletion.create(
            model="gpt-4",
            messages=[
                {"role": "user", "content": "서울 날씨 어때?"},
                message,
                {
                    "role": "function",
                    "name": function_name,
                    "content": json.dumps(weather_result)
                }
            ]
        )
        
        print(second_response["choices"][0]["message"]["content"])

💡 핵심 포인트:
- AI가 function_call 필드로 어떤 함수를 호출할지 결정
- 우리가 실제 함수를 실행하고 결과를 다시 전달
- AI가 결과를 자연어로 변환해서 답변

import anthropic

client = anthropic.Anthropic(api_key="your-api-key")

tools = [
    {
        "name": "get_weather",
        "description": "날씨 정보를 가져옵니다",
        "input_schema": {
            "type": "object",
            "properties": {
                "location": {
                    "type": "string",
                    "description": "도시 이름"
                }
            },
            "required": ["location"]
        }
    }
]

message = client.messages.create(
    model="claude-3-opus-20240229",
    max_tokens=1024,
    tools=tools,
    messages=[
        {"role": "user", "content": "서울 날씨 알려줘"}
    ]
)

# Tool use 처리
if message.stop_reason == "tool_use":
    tool_use = next(block for block in message.content if block.type == "tool_use")
    tool_name = tool_use.name
    tool_input = tool_use.input
    
    # 도구 실행 및 결과 반환...

시나리오: 사용자가 "지난 3개월 매출 데이터를 분석하고 시각화해줘"라고 요청

에이전트의 작업 흐름:

1️⃣ query_database 도구로 매출 데이터 조회
2️⃣ pandas_analysis 도구로 통계 분석 (평균, 추세 등)
3️⃣ create_chart 도구로 그래프 생성
4️⃣ generate_report 도구로 분석 리포트 작성

# 실제 구현 예시
class DataAnalysisAgent:
    def __init__(self):
        self.tools = {
            "query_database": self.query_db,
            "analyze_data": self.analyze,
            "create_chart": self.visualize,
            "generate_report": self.report
        }
    
    def query_db(self, query: str):
        # SQL 실행
        return pd.read_sql(query, connection)
    
    def analyze(self, data: pd.DataFrame):
        return {
            "mean": data.mean(),
            "trend": calculate_trend(data),
            "insights": generate_insights(data)
        }
    
    def visualize(self, data: pd.DataFrame, chart_type: str):
        # 차트 생성
        plt.figure(figsize=(10, 6))
        # ... 차트 그리기
        return "chart.png"
    
    def report(self, analysis: dict, chart_path: str):
        # 리포트 생성
        return f"""
        ## 매출 분석 리포트
        
        평균 매출: {analysis['mean']}
        추세: {analysis['trend']}
        
        ![차트]({chart_path})
        
        주요 인사이트:
        {analysis['insights']}
        """

시나리오: "중요한 이메일만 요약해서 슬랙으로 보내줘"

작업 흐름:

1️⃣ fetch_emails: Gmail API로 새 이메일 가져오기
2️⃣ classify_importance: AI로 중요도 판단
3️⃣ summarize_content: 이메일 내용 요약
4️⃣ send_slack_message: 슬랙으로 알림 전송

async def email_automation_agent(user_id):
    # 1. 이메일 가져오기
    emails = await fetch_emails(user_id, unread=True)
    
    for email in emails:
        # 2. 중요도 판단
        importance = await classify_importance(email)
        
        if importance > 0.7:  # 중요한 이메일만
            # 3. 요약
            summary = await summarize_content(email.body)
            
            # 4. 슬랙 전송
            await send_slack_message(
                channel=user_id,
                message=f"""
                📧 중요 이메일 도착!
                
                발신: {email.sender}
                제목: {email.subject}
                요약: {summary}
                
                [이메일 보기]({email.link})
                """
            )

안전한 도구 실행 패턴

class SafeToolExecutor:
    def __init__(self, max_iterations=10, timeout=30):
        self.max_iterations = max_iterations
        self.timeout = timeout
        self.execution_count = 0
    
    def execute_with_safety(self, tool, arguments):
        # 1. 실행 횟수 체크
        if self.execution_count >= self.max_iterations:
            raise Exception("최대 실행 횟수 초과")
        
        # 2. 입력 검증
        validated_args = self.validate_arguments(arguments)
        
        # 3. 타임아웃 설정
        with timeout_context(self.timeout):
            try:
                result = tool.execute(**validated_args)
                self.execution_count += 1
                
                # 4. 결과 검증
                return self.validate_result(result)
            except Exception as e:
                # 5. 에러 처리
                return self.handle_error(e)
    
    def validate_arguments(self, args):
        # SQL Injection 방지
        # XSS 방지
        # 파일 경로 검증 등
        return sanitized_args
    
    def validate_result(self, result):
        # 민감한 정보 필터링
        # 크기 제한
        return filtered_result

import logging
from datetime import datetime

class ToolLogger:
    def __init__(self):
        self.logger = logging.getLogger("AgentTools")
        self.execution_history = []
    
    def log_tool_call(self, tool_name, arguments, result, duration):
        log_entry = {
            "timestamp": datetime.now().isoformat(),
            "tool": tool_name,
            "arguments": arguments,
            "result": result,
            "duration_ms": duration,
            "success": result.get("success", False)
        }
        
        self.execution_history.append(log_entry)
        
        # 콘솔 출력
        self.logger.info(f"""
        🔧 도구 실행: {tool_name}
        📥 입력: {arguments}
        📤 출력: {result}
        ⏱️ 소요시간: {duration}ms
        """)
        
        # 실패 시 알림
        if not result.get("success"):
            self.logger.error(f"도구 실행 실패: {tool_name}")
            self.send_alert(log_entry)
    
    def get_statistics(self):
        """도구 사용 통계"""
        return {
            "total_calls": len(self.execution_history),
            "success_rate": self.calculate_success_rate(),
            "avg_duration": self.calculate_avg_duration(),
            "most_used_tool": self.get_most_used_tool()
        }

🧠 ReAct 사고 과정 예시

사용자: "2024년 파리 올림픽 금메달 1위 국가와 그 나라의 수도를 알려줘"

Thought 1: 먼저 2024 파리 올림픽 금메달 순위를 검색해야겠다.
Action 1: search("2024 파리 올림픽 금메달 순위")
Observation 1: 미국이 40개로 1위입니다.

Thought 2: 미국의 수도를 확인해야겠다.
Action 2: search("미국 수도")
Observation 2: 워싱턴 D.C.입니다.

Thought 3: 이제 답변할 수 있다!
Final Answer: 2024년 파리 올림픽 금메달 1위는 미국(40개)이며, 
미국의 수도는 워싱턴 D.C.입니다.

이 패턴의 핵심은 AI가 각 단계에서 왜 그 도구를 사용하는지 설명한다는 거야. 디버깅할 때 엄청 유용해! 🔍

예시: 콘텐츠 제작 팀

🎨 Designer Agent: 이미지 생성, 레이아웃 디자인
✍️ Writer Agent: 텍스트 작성, 편집
📊 Analyst Agent: 데이터 분석, 트렌드 파악
🎯 Manager Agent: 전체 조율, 최종 결정

class MultiAgentSystem:
    def __init__(self):
        self.agents = {
            "designer": DesignerAgent(),
            "writer": WriterAgent(),
            "analyst": AnalystAgent(),
            "manager": ManagerAgent()
        }
    
    async def create_content(self, topic):
        # 1. 분석가가 트렌드 분석
        trends = await self.agents["analyst"].analyze_trends(topic)
        
        # 2. 작가가 초안 작성
        draft = await self.agents["writer"].write_draft(topic, trends)
        
        # 3. 디자이너가 이미지 생성
        images = await self.agents["designer"].create_images(draft)
        
        # 4. 매니저가 최종 검토 및 조합
        final_content = await self.agents["manager"].finalize(
            draft, images, trends
        )
        
        return final_content

class ComposedTool:
    def __init__(self, name, steps):
        self.name = name
        self.steps = steps  # 실행할 도구들의 순서
    
    async def execute(self, initial_input):
        result = initial_input
        
        for step in self.steps:
            tool = step["tool"]
            transform = step.get("transform", lambda x: x)
            
            # 이전 결과를 변환해서 다음 도구의 입력으로
            tool_input = transform(result)
            result = await tool.execute(tool_input)
        
        return result

# 사용 예시: "웹페이지 요약" 도구
summarize_webpage = ComposedTool(
    name="summarize_webpage",
    steps=[
        {
            "tool": fetch_webpage_tool,
            "transform": lambda url: {"url": url}
        },
        {
            "tool": extract_text_tool,
            "transform": lambda html: {"html": html}
        },
        {
            "tool": summarize_text_tool,
            "transform": lambda text: {"text": text, "max_length": 200}
        }
    ]
)

# 실행
summary = await summarize_webpage.execute("https://example.com")

class ConditionalAgent:
    async def process_request(self, user_input):
        # 의도 파악
        intent = await self.classify_intent(user_input)
        
        if intent == "calculation":
            return await self.calculator_tool.execute(user_input)
        
        elif intent == "search":
            # 검색 종류도 세분화
            search_type = await self.classify_search_type(user_input)
            
            if search_type == "recent_news":
                return await self.news_search_tool.execute(user_input)
            elif search_type == "product":
                return await self.product_search_tool.execute(user_input)
            else:
                return await self.web_search_tool.execute(user_input)
        
        elif intent == "data_analysis":
            # 데이터 크기에 따라 다른 도구 사용
            data_size = await self.estimate_data_size(user_input)
            
            if data_size > 1000000:
                return await self.big_data_tool.execute(user_input)
            else:
                return await self.simple_analysis_tool.execute(user_input)
        
        else:
            return "죄송합니다. 요청을 이해하지 못했습니다."

1. 기본 서버 구조

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import List, Dict, Optional
import asyncio

app = FastAPI(title="AI Agent API")

class ChatRequest(BaseModel):
    message: str
    user_id: str
    session_id: Optional[str] = None

class ChatResponse(BaseModel):
    response: str
    tools_used: List[str]
    execution_time: float

# 전역 에이전트 인스턴스
agent = AIAgent()

@app.post("/chat", response_model=ChatResponse)
async def chat(request: ChatRequest):
    """메인 채팅 엔드포인트"""
    try:
        start_time = time.time()
        
        # 에이전트 실행
        result = await agent.process(
            message=request.message,
            user_id=request.user_id,
            session_id=request.session_id
        )
        
        execution_time = time.time() - start_time
        
        return ChatResponse(
            response=result["response"],
            tools_used=result["tools_used"],
            execution_time=execution_time
        )
    
    except Exception as e:
        raise HTTPException(status_code=500, detail=str(e))

@app.get("/tools")
async def list_tools():
    """사용 가능한 도구 목록"""
    return {
        "tools": agent.get_available_tools()
    }

@app.post("/tools/{tool_name}/execute")
async def execute_tool(tool_name: str, arguments: Dict):
    """특정 도구 직접 실행 (테스트용)"""
    try:
        result = await agent.execute_tool(tool_name, arguments)
        return {"success": True, "result": result}
    except Exception as e:
        return {"success": False, "error": str(e)}

2. 에이전트 코어 구현

import openai
from typing import List, Dict, Any

class AIAgent:
    def __init__(self):
        self.tool_manager = ToolManager()
        self.memory = ConversationMemory()
        self.logger = AgentLogger()
        
        # 도구 등록
        self._register_tools()
    
    def _register_tools(self):
        """모든 도구 등록"""
        self.tool_manager.register(WeatherTool())
        self.tool_manager.register(CalculatorTool())
        self.tool_manager.register(SearchTool())
        self.tool_manager.register(DatabaseTool())
        # ... 더 많은 도구들
    
    async def process(self, message: str, user_id: str, 
                     session_id: str = None) -> Dict:
        """메인 처리 로직"""
        
        # 1. 대화 기록 로드
        history = await self.memory.get_history(session_id)
        
        # 2. 도구 스키마 준비
        tools_schema = self.tool_manager.get_schemas()
        
        # 3. LLM 호출
        messages = self._build_messages(message, history)
        
        response = await openai.ChatCompletion.acreate(
            model="gpt-4",
            messages=messages,
            functions=tools_schema,
            function_call="auto"
        )
        
        # 4. 도구 호출이 필요한지 확인
        message_obj = response["choices"][0]["message"]
        
        if message_obj.get("function_call"):
            # 도구 실행
            result = await self._execute_tool_call(message_obj)
            
            # 결과를 다시 LLM에 전달
            final_response = await self._get_final_response(
                messages, message_obj, result
            )
            
            tools_used = [message_obj["function_call"]["name"]]
        else:
            final_response = message_obj["content"]
            tools_used = []
        
        # 5. 대화 기록 저장
        await self.memory.save_turn(
            session_id, message, final_response
        )
        
        # 6. 로깅
        self.logger.log_interaction(
            user_id, message, final_response, tools_used
        )
        
        return {
            "response": final_response,
            "tools_used": tools_used
        }
    
    async def _execute_tool_call(self, message_obj):
        """도구 실행"""
        function_name = message_obj["function_call"]["name"]
        arguments = json.loads(
            message_obj["function_call"]["arguments"]
        )
        
        return await self.tool_manager.execute(
            function_name, arguments
        )
    
    def _build_messages(self, current_message, history):
        """메시지 구성"""
        messages = [
            {
                "role": "system",
                "content": """당신은 도움이 되는 AI 어시스턴트입니다.
                필요할 때 제공된 도구들을 사용하세요."""
            }
        ]
        
        # 이전 대화 추가
        messages.extend(history)
        
        # 현재 메시지 추가
        messages.append({
            "role": "user",
            "content": current_message
        })
        
        return messages

import redis
import json
from datetime import timedelta

class ConversationMemory:
    def __init__(self):
        self.redis_client = redis.Redis(
            host='localhost',
            port=6379,
            decode_responses=True
        )
        self.ttl = timedelta(hours=24)  # 24시간 후 자동 삭제
    
    async def get_history(self, session_id: str, 
                         max_turns: int = 10) -> List[Dict]:
        """대화 기록 조회"""
        key = f"conversation:{session_id}"
        history_json = self.redis_client.get(key)
        
        if not history_json:
            return []
        
        history = json.loads(history_json)
        
        # 최근 N개만 반환 (토큰 절약)
        return history[-max_turns * 2:]  # user + assistant 쌍
    
    async def save_turn(self, session_id: str, 
                       user_message: str, 
                       assistant_message: str):
        """대화 턴 저장"""
        key = f"conversation:{session_id}"
        
        # 기존 기록 가져오기
        history = await self.get_history(session_id)
        
        # 새 턴 추가
        history.extend([
            {"role": "user", "content": user_message},
            {"role": "assistant", "content": assistant_message}
        ])
        
        # 저장 (TTL 설정)
        self.redis_client.setex(
            key,
            self.ttl,
            json.dumps(history)
        )
    
    async def clear_history(self, session_id: str):
        """대화 기록 삭제"""
        key = f"conversation:{session_id}"
        self.redis_client.delete(key)

import asyncio

class ParallelAgent:
    async def process_parallel_tools(self, tool_calls: List[Dict]):
        """여러 도구를 동시에 실행"""
        
        # 각 도구 호출을 비동기 태스크로 생성
        tasks = [
            self.execute_tool(call["name"], call["arguments"])
            for call in tool_calls
        ]
        
        # 모든 태스크를 동시에 실행
        results = await asyncio.gather(*tasks, return_exceptions=True)
        
        # 결과 처리
        processed_results = []
        for i, result in enumerate(results):
            if isinstance(result, Exception):
                processed_results.append({
                    "tool": tool_calls[i]["name"],
                    "success": False,
                    "error": str(result)
                })
            else:
                processed_results.append({
                    "tool": tool_calls[i]["name"],
                    "success": True,
                    "result": result
                })
        
        return processed_results

# 사용 예시
tool_calls = [
    {"name": "get_weather", "arguments": {"city": "서울"}},
    {"name": "get_weather", "arguments": {"city": "부산"}},
    {"name": "get_weather", "arguments": {"city": "대구"}}
]

# 순차 실행: 3초 (각 1초씩)
# 병렬 실행: 1초 (동시에!)
results = await agent.process_parallel_tools(tool_calls)

from functools import wraps
import hashlib

def cached_tool(ttl_seconds=300):
    """도구 결과를 캐싱하는 데코레이터"""
    def decorator(func):
        cache = {}
        
        @wraps(func)
        async def wrapper(*args, **kwargs):
            # 캐시 키 생성
            cache_key = hashlib.md5(
                f"{func.__name__}:{args}:{kwargs}".encode()
            ).hexdigest()
            
            # 캐시 확인
            if cache_key in cache:
                cached_value, timestamp = cache[cache_key]
                if time.time() - timestamp < ttl_seconds:
                    print(f"✅ 캐시 히트: {func.__name__}")
                    return cached_value
            
            # 캐시 미스 - 실제 실행
            result = await func(*args, **kwargs)
            cache[cache_key] = (result, time.time())
            
            return result
        
        return wrapper
    return decorator

# 사용 예시
@cached_tool(ttl_seconds=600)  # 10분 캐싱
async def get_weather(city: str):
    # API 호출 (비용 발생)
    return await weather_api.fetch(city)

class LLMLoadBalancer:
    def __init__(self):
        self.providers = [
            {"name": "openai", "client": openai_client, "weight": 0.6},
            {"name": "anthropic", "client": anthropic_client, "weight": 0.3},
            {"name": "cohere", "client": cohere_client, "weight": 0.1}
        ]
        self.current_loads = {p["name"]: 0 for p in self.providers}
    
    async def get_completion(self, messages, **kwargs):
        """가장 부하가 적은 제공자 선택"""
        
        # 가중치와 현재 부하를 고려해서 선택
        selected = min(
            self.providers,
            key=lambda p: self.current_loads[p["name"]] / p["weight"]
        )
        
        self.current_loads[selected["name"]] += 1
        
        try:
            result = await selected["client"].complete(messages, **kwargs)
            return result
        finally:
            self.current_loads[selected["name"]] -= 1