AI Agent 완전 가이드 - 개념, 아키텍처, 프레임워크, 실전 구축

# Agent Loop 의사 코드
def agent_loop(goal: str, tools: list, max_steps: int = 10):
    messages = [{"role": "user", "content": goal}]
    
    for step in range(max_steps):
        # 1. LLM에게 다음 행동 결정 요청
        response = llm.generate(messages, tools=tools)
        
        # 2. 응답이 최종 답변이면 종료
        if response.is_final_answer:
            return response.content
        
        # 3. 도구 호출이면 실행
        if response.tool_calls:
            for tool_call in response.tool_calls:
                result = execute_tool(tool_call)
                messages.append({
                    "role": "tool",
                    "content": result,
                    "tool_call_id": tool_call.id
                })
        
        # 4. 결과를 메시지에 추가하고 다음 반복
        messages.append(response)
    
    return "최대 단계 수에 도달했습니다."

# Anthropic Claude Tool Use 예시
import anthropic
 
client = anthropic.Anthropic()
 
tools = [
    {
        "name": "execute_sql",
        "description": "데이터베이스에 SQL 쿼리를 실행하고 결과를 반환합니다. SELECT 쿼리만 허용됩니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "query": {
                    "type": "string",
                    "description": "실행할 SQL SELECT 쿼리"
                },
                "database": {
                    "type": "string",
                    "enum": ["analytics", "production", "staging"],
                    "description": "쿼리를 실행할 데이터베이스"
                }
            },
            "required": ["query", "database"]
        }
    },
    {
        "name": "send_slack_message",
        "description": "Slack 채널에 메시지를 전송합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "channel": {
                    "type": "string",
                    "description": "Slack 채널명 (예: #engineering)"
                },
                "message": {
                    "type": "string",
                    "description": "전송할 메시지 내용"
                }
            },
            "required": ["channel", "message"]
        }
    }
]
 
response = client.messages.create(
    model="claude-sonnet-4-6",
    max_tokens=1024,
    tools=tools,
    messages=[{
        "role": "user",
        "content": "analytics DB에서 이번 달 매출 합계를 조회하고, 결과를 #sales 채널에 공유해줘."
    }]
)

# 단기 기억: 대화 메시지 리스트
messages = [
    {"role": "system", "content": "당신은 데이터 엔지니어링 어시스턴트입니다."},
    {"role": "user", "content": "Spark 클러스터 상태를 확인해줘"},
    {"role": "assistant", "content": "...", "tool_calls": [...]},
    {"role": "tool", "content": "클러스터 상태: 정상, 노드 5개 활성"},
    {"role": "assistant", "content": "Spark 클러스터는 정상입니다. 5개 노드가 활성 상태입니다."},
    {"role": "user", "content": "어제 실행한 배치 잡 상태도 확인해줘"},
    # ... 대화가 계속됨
]

# 컨텍스트 윈도우 관리: 요약 방식
def manage_context(messages, max_tokens=4096):
    if count_tokens(messages) > max_tokens:
        # 오래된 메시지를 요약으로 압축
        old_messages = messages[1:-5]  # 시스템 프롬프트, 최근 5개 제외
        summary = llm.summarize(old_messages)
        
        return [
            messages[0],  # 시스템 프롬프트 유지
            {"role": "system", "content": f"이전 대화 요약: {summary}"},
            *messages[-5:]  # 최근 5개 메시지 유지
        ]
    return messages

# 장기 기억: 벡터 DB에 경험 저장 및 검색
from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings
 
# 장기 기억 저장소
long_term_memory = Chroma(
    collection_name="agent_memory",
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./memory_db"
)
 
# 경험 저장
def save_experience(task: str, result: str, success: bool):
    long_term_memory.add_texts(
        texts=[f"태스크: {task}\n결과: {result}\n성공: {success}"],
        metadatas=[{
            "task_type": classify_task(task),
            "success": success,
            "timestamp": datetime.now().isoformat()
        }]
    )
 
# 과거 경험 검색 (유사한 태스크 수행 시 참고)
def recall_experience(current_task: str, k: int = 3):
    results = long_term_memory.similarity_search(current_task, k=k)
    return results

# 작업 기억: Scratchpad
class AgentScratchpad:
    def __init__(self):
        self.plan = []           # 현재 계획
        self.completed = []      # 완료된 단계
        self.intermediate = {}   # 중간 결과물
        self.observations = []   # 관찰 기록
    
    def update_plan(self, plan: list):
        self.plan = plan
    
    def mark_complete(self, step: int, result: str):
        self.completed.append(step)
        self.intermediate[f"step_{step}"] = result
    
    def get_context(self) -> str:
        """현재 작업 상태를 텍스트로 변환"""
        return f"""
현재 계획: {self.plan}
완료 단계: {self.completed}
중간 결과: {self.intermediate}
남은 단계: {[s for s in self.plan if s not in self.completed]}
"""

# LangGraph를 이용한 ReAct Agent
from langgraph.prebuilt import create_react_agent
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
 
@tool
def search_database(query: str) -> str:
    """사내 데이터베이스에서 정보를 검색합니다."""
    # 실제 DB 검색 로직
    return f"검색 결과: {query}에 대한 데이터..."
 
@tool
def run_sql(sql: str) -> str:
    """SQL 쿼리를 실행하고 결과를 반환합니다."""
    # 실제 SQL 실행 로직
    return f"쿼리 결과: ..."
 
llm = ChatOpenAI(model="gpt-4o")
tools = [search_database, run_sql]
 
# ReAct Agent 생성
agent = create_react_agent(llm, tools)
 
# 실행
result = agent.invoke({
    "messages": [{"role": "user", "content": "이번 분기 매출을 조회해줘"}]
})

# LangGraph를 이용한 커스텀 Agent 워크플로
from langgraph.graph import StateGraph, MessagesState, START, END
 
def should_continue(state: MessagesState):
    """도구 호출이 있으면 계속, 없으면 종료"""
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"
    return END
 
def call_model(state: MessagesState):
    """LLM 호출"""
    response = llm.invoke(state["messages"])
    return {"messages": [response]}
 
def call_tools(state: MessagesState):
    """도구 실행"""
    # 도구 실행 로직
    ...
 
# 그래프 구성
workflow = StateGraph(MessagesState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)
workflow.add_edge(START, "agent")
workflow.add_conditional_edges("agent", should_continue)
workflow.add_edge("tools", "agent")
 
app = workflow.compile()

from crewai import Agent, Task, Crew
 
# 에이전트 정의
researcher = Agent(
    role="데이터 리서처",
    goal="정확한 시장 데이터와 트렌드를 조사합니다",
    backstory="10년 경력의 시장 분석 전문가입니다.",
    tools=[search_tool, web_scraper],
    llm="gpt-4o"
)
 
writer = Agent(
    role="보고서 작성자",
    goal="조사 결과를 명확한 보고서로 작성합니다",
    backstory="기술 문서 작성 경험이 풍부한 테크니컬 라이터입니다.",
    llm="gpt-4o"
)
 
reviewer = Agent(
    role="품질 검토자",
    goal="보고서의 정확성과 완성도를 검토합니다",
    backstory="데이터 검증과 QA 전문가입니다.",
    llm="gpt-4o"
)
 
# 태스크 정의
research_task = Task(
    description="2025년 AI 시장 동향을 조사하세요.",
    agent=researcher,
    expected_output="시장 규모, 성장률, 주요 트렌드 요약"
)
 
write_task = Task(
    description="조사 결과를 기반으로 보고서를 작성하세요.",
    agent=writer,
    expected_output="구조화된 시장 동향 보고서"
)
 
review_task = Task(
    description="보고서의 데이터 정확성과 논리를 검토하세요.",
    agent=reviewer,
    expected_output="검토 의견 및 수정 사항"
)
 
# Crew 구성 및 실행
crew = Crew(
    agents=[researcher, writer, reviewer],
    tasks=[research_task, write_task, review_task],
    verbose=True
)
 
result = crew.kickoff()

from autogen import AssistantAgent, UserProxyAgent
 
# 어시스턴트 에이전트
assistant = AssistantAgent(
    name="data_engineer",
    system_message="당신은 데이터 엔지니어입니다. Python 코드로 데이터 분석 작업을 수행합니다.",
    llm_config={"model": "gpt-4o"}
)
 
# 사용자 프록시 (코드 실행 담당)
user_proxy = UserProxyAgent(
    name="executor",
    human_input_mode="NEVER",  # 자동 실행
    code_execution_config={"work_dir": "workspace"}
)
 
# 대화 시작
user_proxy.initiate_chat(
    assistant,
    message="sales.csv 파일을 읽어서 월별 매출 추이를 분석하고 차트를 그려주세요."
)

import anthropic
from anthropic.types import ToolUseBlock
 
client = anthropic.Anthropic()
 
tools = [
    {
        "name": "read_file",
        "description": "파일의 내용을 읽어 반환합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "파일 경로"}
            },
            "required": ["path"]
        }
    },
    {
        "name": "write_file",
        "description": "파일에 내용을 작성합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "path": {"type": "string", "description": "파일 경로"},
                "content": {"type": "string", "description": "작성할 내용"}
            },
            "required": ["path", "content"]
        }
    }
]
 
def agent_loop(goal: str):
    messages = [{"role": "user", "content": goal}]
    
    while True:
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            tools=tools,
            messages=messages
        )
        
        # 도구 호출이 없으면 최종 응답
        if response.stop_reason == "end_turn":
            return extract_text(response)
        
        # 도구 호출 처리
        messages.append({"role": "assistant", "content": response.content})
        
        tool_results = []
        for block in response.content:
            if isinstance(block, ToolUseBlock):
                result = execute_tool(block.name, block.input)
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })
        
        messages.append({"role": "user", "content": tool_results})
 
result = agent_loop("config.yaml 파일을 읽어서 포트 설정을 8080으로 변경해줘")

from openai import OpenAI
 
client = OpenAI()
 
# Agent를 Responses API로 구현
response = client.responses.create(
    model="gpt-4o",
    tools=[
        {"type": "web_search"},
        {"type": "code_interpreter"},
        {
            "type": "function",
            "function": {
                "name": "query_database",
                "description": "사내 데이터베이스를 조회합니다.",
                "parameters": {
                    "type": "object",
                    "properties": {
                        "sql": {"type": "string"}
                    }
                }
            }
        }
    ],
    input="이번 달 신규 가입자 수를 조회하고 추이를 분석해줘"
)

import anthropic
import json
 
client = anthropic.Anthropic()
 
# 도구 정의
tools = [
    {
        "name": "query_sales_db",
        "description": "매출 데이터베이스에서 데이터를 조회합니다. SQL 쿼리를 실행합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "sql": {"type": "string", "description": "실행할 SQL 쿼리"}
            },
            "required": ["sql"]
        }
    },
    {
        "name": "calculate",
        "description": "수학 계산을 수행합니다. Python 수식을 평가합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "expression": {"type": "string", "description": "Python 수식"}
            },
            "required": ["expression"]
        }
    },
    {
        "name": "create_report",
        "description": "분석 결과를 보고서로 작성합니다.",
        "input_schema": {
            "type": "object",
            "properties": {
                "title": {"type": "string"},
                "content": {"type": "string"},
                "format": {"type": "string", "enum": ["markdown", "html", "text"]}
            },
            "required": ["title", "content"]
        }
    }
]
 
# 도구 실행 함수
def execute_tool(name: str, input_data: dict) -> str:
    if name == "query_sales_db":
        # 실제로는 DB 연결하여 실행
        return json.dumps({"rows": [
            {"month": "2025-01", "revenue": 1200000},
            {"month": "2025-02", "revenue": 1350000},
            {"month": "2025-03", "revenue": 1180000}
        ]})
    elif name == "calculate":
        result = eval(input_data["expression"])
        return str(result)
    elif name == "create_report":
        return f"보고서 '{input_data['title']}' 생성 완료"
    return "알 수 없는 도구"
 
# Agent 루프
def run_agent(goal: str):
    messages = [{"role": "user", "content": goal}]
    
    print(f"목표: {goal}\n{'='*50}")
    
    for step in range(10):
        response = client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=4096,
            system="당신은 데이터 분석 Agent입니다. 도구를 활용하여 사용자의 요청을 수행하세요.",
            tools=tools,
            messages=messages
        )
        
        # 텍스트 응답 출력
        for block in response.content:
            if hasattr(block, "text"):
                print(f"\n[Agent 응답] {block.text}")
        
        # 최종 응답이면 종료
        if response.stop_reason == "end_turn":
            break
        
        # 도구 호출 처리
        messages.append({"role": "assistant", "content": response.content})
        tool_results = []
        
        for block in response.content:
            if block.type == "tool_use":
                print(f"\n[도구 호출] {block.name}({json.dumps(block.input, ensure_ascii=False)})")
                result = execute_tool(block.name, block.input)
                print(f"[도구 결과] {result[:200]}")
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": result
                })
        
        messages.append({"role": "user", "content": tool_results})
    
    print(f"\n{'='*50}\nAgent 실행 완료 (단계: {step + 1})")
 
# 실행
run_agent("Q1 매출 데이터를 조회하고, 월별 증감률을 계산한 뒤 보고서를 작성해주세요.")

from langgraph.graph import StateGraph, MessagesState, START, END
from langchain_openai import ChatOpenAI
from typing import Literal
 
llm = ChatOpenAI(model="gpt-4o")
 
# 각 Agent 노드 정의
def researcher(state: MessagesState):
    """조사 Agent: 데이터 수집 및 분석"""
    system = "당신은 데이터 리서처입니다. 주어진 주제에 대해 핵심 데이터를 조사하세요."
    messages = [{"role": "system", "content": system}] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}
 
def writer(state: MessagesState):
    """작성 Agent: 보고서 작성"""
    system = "당신은 테크니컬 라이터입니다. 조사 결과를 체계적인 보고서로 작성하세요."
    messages = [{"role": "system", "content": system}] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}
 
def reviewer(state: MessagesState):
    """검토 Agent: 품질 검토"""
    system = """당신은 품질 검토자입니다. 보고서의 정확성과 완성도를 검토하세요.
    검토 결과를 'APPROVED' 또는 'REVISION_NEEDED: [수정사항]'으로 반환하세요."""
    messages = [{"role": "system", "content": system}] + state["messages"]
    response = llm.invoke(messages)
    return {"messages": [response]}
 
def should_revise(state: MessagesState) -> Literal["writer", END]:
    """검토 결과에 따라 수정 또는 완료 결정"""
    last_message = state["messages"][-1].content
    if "REVISION_NEEDED" in last_message:
        return "writer"
    return END
 
# 워크플로 그래프 구성
workflow = StateGraph(MessagesState)
workflow.add_node("researcher", researcher)
workflow.add_node("writer", writer)
workflow.add_node("reviewer", reviewer)
 
workflow.add_edge(START, "researcher")
workflow.add_edge("researcher", "writer")
workflow.add_edge("writer", "reviewer")
workflow.add_conditional_edges("reviewer", should_revise)
 
app = workflow.compile()
 
# 실행
result = app.invoke({
    "messages": [{"role": "user", "content": "2025년 생성형 AI 시장 동향 보고서를 작성해주세요."}]
})

from langchain_chroma import Chroma
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent
 
# 벡터 DB (사전 구축된 사내 문서)
vectorstore = Chroma(
    persist_directory="./company_docs_db",
    embedding_function=OpenAIEmbeddings()
)
 
@tool
def search_internal_docs(query: str) -> str:
    """사내 기술 문서, 정책, 가이드를 검색합니다.
    Spark, Kafka, NiFi, Kudu 등의 기술 문서와 운영 가이드를 찾을 때 사용합니다."""
    results = vectorstore.similarity_search(query, k=3)
    return "\n\n---\n\n".join([
        f"[출처: {r.metadata.get('source', 'unknown')}]\n{r.page_content}"
        for r in results
    ])
 
@tool
def run_spark_query(sql: str) -> str:
    """Spark SQL 쿼리를 실행하고 결과를 반환합니다.
    데이터 분석, 집계, 조인 등의 쿼리를 수행할 때 사용합니다."""
    # 실제 Spark 연동 로직
    return f"쿼리 실행 결과: ..."
 
@tool
def create_jira_ticket(title: str, description: str, priority: str) -> str:
    """Jira 티켓을 생성합니다.
    버그 리포트, 작업 요청, 개선 사항 등록 시 사용합니다."""
    # 실제 Jira API 호출
    return f"Jira 티켓 생성 완료: PROJ-1234 '{title}'"
 
# RAG + Agent
llm = ChatOpenAI(model="gpt-4o")
tools = [search_internal_docs, run_spark_query, create_jira_ticket]
 
agent = create_react_agent(
    llm, tools,
    prompt="당신은 Data Dynamics의 시니어 데이터 엔지니어입니다. "
           "사내 문서를 검색하고, 데이터를 분석하며, 필요 시 Jira 티켓을 생성합니다."
)
 
# 실행
result = agent.invoke({
    "messages": [{
        "role": "user",
        "content": "Kudu 테이블의 파티셔닝 전략을 사내 가이드에서 찾아보고, "
                   "현재 orders 테이블에 적용할 수 있는 최적화 방안을 분석해주세요. "
                   "개선 사항이 있으면 Jira 티켓도 만들어줘."
    }]
})

# 가드레일 구현 예시
class AgentGuardrails:
    def __init__(self):
        self.allowed_tools = {"search_docs", "run_sql", "calculate"}
        self.blocked_patterns = [
            r"DROP\s+TABLE",
            r"DELETE\s+FROM",
            r"UPDATE\s+.*SET",
            r"INSERT\s+INTO",
            r"rm\s+-rf",
        ]
        self.max_steps = 15
        self.max_cost_usd = 1.0
    
    def validate_tool_call(self, tool_name: str, tool_input: dict) -> tuple:
        """도구 호출 전 검증"""
        # 1. 허용된 도구인지 확인
        if tool_name not in self.allowed_tools:
            return False, f"허용되지 않은 도구: {tool_name}"
        
        # 2. 위험한 패턴 검사
        input_str = json.dumps(tool_input)
        for pattern in self.blocked_patterns:
            if re.search(pattern, input_str, re.IGNORECASE):
                return False, f"위험한 패턴 감지: {pattern}"
        
        return True, "OK"
    
    def check_budget(self, current_cost: float) -> bool:
        """비용 한도 확인"""
        return current_cost < self.max_cost_usd
 
guardrails = AgentGuardrails()
 
# Agent 루프에서 가드레일 적용
for block in response.content:
    if block.type == "tool_use":
        is_valid, reason = guardrails.validate_tool_call(block.name, block.input)
        if not is_valid:
            # 도구 실행 차단, 에이전트에 피드백
            tool_results.append({
                "type": "tool_result",
                "tool_use_id": block.id,
                "content": f"차단됨: {reason}",
                "is_error": True
            })

def human_approval_required(tool_name: str, tool_input: dict) -> bool:
    """사람의 승인이 필요한 작업인지 판단"""
    high_risk_tools = {"send_email", "create_jira_ticket", "deploy", "delete_file"}
    return tool_name in high_risk_tools
 
def request_human_approval(tool_name: str, tool_input: dict) -> bool:
    """사람에게 승인 요청"""
    print(f"\n{'='*50}")
    print(f"[승인 요청] Agent가 다음 작업을 수행하려 합니다:")
    print(f"  도구: {tool_name}")
    print(f"  입력: {json.dumps(tool_input, indent=2, ensure_ascii=False)}")
    print(f"{'='*50}")
    
    approval = input("승인하시겠습니까? (y/n): ")
    return approval.lower() == "y"
 
# Agent 루프에서 적용
for block in response.content:
    if block.type == "tool_use":
        if human_approval_required(block.name, block.input):
            if not request_human_approval(block.name, block.input):
                tool_results.append({
                    "type": "tool_result",
                    "tool_use_id": block.id,
                    "content": "사용자가 작업을 거부했습니다.",
                    "is_error": True
                })
                continue
        
        result = execute_tool(block.name, block.input)

class AgentErrorHandler:
    def __init__(self, max_retries: int = 3):
        self.max_retries = max_retries
        self.error_counts = {}
    
    def handle_tool_error(self, tool_name: str, error: Exception) -> str:
        """도구 실행 에러 처리"""
        self.error_counts[tool_name] = self.error_counts.get(tool_name, 0) + 1
        
        if self.error_counts[tool_name] >= self.max_retries:
            return f"도구 '{tool_name}'이 {self.max_retries}회 연속 실패했습니다. 다른 방법을 시도하세요."
        
        return f"에러 발생: {str(error)}. 재시도 가능합니다. ({self.error_counts[tool_name]}/{self.max_retries})"
    
    def handle_llm_error(self, error: Exception) -> str:
        """LLM 호출 에러 처리"""
        if "rate_limit" in str(error).lower():
            time.sleep(5)  # Rate limit 시 대기
            return "RETRY"
        elif "context_length" in str(error).lower():
            return "TRUNCATE"  # 컨텍스트 축소
        return "ABORT"