AI Agent 아키텍처: 자율 에이전트와 LangChain 프레임워크

AI Agents는 독립적으로 의사결정을 내리고, 도구를 사용하며, 복잡한 작업을 단계별로 해결할 수 있는 인공지능 시스템이다. 이 가이드에서는 에이전트 아키텍처와 구현 방식을 살펴본다.

What is an Agent?

챗봇과 에이전트의 간단한 비교:

1Chatbot:
2User: "How is the weather today?"
3Bot: "I don't know the weather." (static response)
4
5Agent:
6User: "How is the weather today?"
7Agent: 
8  1. "I need to find out the weather."
9  2. Call weather_api.get_weather()
10  3. Interpret the result
11  4. "It's 15°C in Istanbul, partly cloudy."

Agent Components

1. LLM (Brain)

의사결정 및 추론 기능.

2. Tools

외부 세계와의 상호작용:

• Web search
• API calls
• Database queries
• Code execution

3. Memory

대화 기록 및 정보 저장.

4. Planning

작업 분해 및 전략 수립.

ReAct Pattern

Reasoning + Acting 사이클:

1Thought: What do I need to do?
2Action: Which tool should I use?
3Action Input: Tool parameters
4Observation: Tool result
5... (repeat)
6Final Answer: Final response

ReAct Example

1Question: "What is Tesla's current stock price and how much has it changed in the last month?"
2
3Thought: I need to find Tesla's stock price.
4Action: stock_price
5Action Input: {"symbol": "TSLA"}
6Observation: Current price: $248.50
7
8Thought: Now I need to calculate the 1-month change.
9Action: stock_history  
10Action Input: {"symbol": "TSLA", "period": "1mo"}
11Observation: 1 month ago: $235.20, Change: +5.65%
12
13Thought: I can now answer the question.
14Final Answer: Tesla's current stock price is $248.50. 
15It has shown a 5.65% increase in the last month.

Agent Development with LangChain

Basic Agent

1from langchain.agents import create_react_agent, AgentExecutor
2from langchain.tools import Tool
3from langchain_openai import ChatOpenAI
4from langchain import hub
5
6# LLM
7llm = ChatOpenAI(model="gpt-4-turbo", temperature=0)
8
9# Tools
10def search_web(query: str) -> str:
11    """Searches the web"""
12    # Web search implementation
13    return f"Search results for: {query}"
14
15def calculate(expression: str) -> str:
16    """Performs mathematical calculation"""
17    return str(eval(expression))
18
19tools = [
20    Tool(name="search", func=search_web, description="Searches the web"),
21    Tool(name="calculator", func=calculate, description="Performs math calculations"),
22]
23
24# Prompt
25prompt = hub.pull("hwchase17/react")
26
27# Agent
28agent = create_react_agent(llm, tools, prompt)
29agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)
30
31# Run
32result = agent_executor.invoke({"input": "What is the world population in 2024?"})

Custom Tools

1from langchain.tools import BaseTool
2from pydantic import BaseModel, Field
3
4class WeatherInput(BaseModel):
5    city: str = Field(description="City name")
6    unit: str = Field(default="celsius", description="Temperature unit")
7
8class WeatherTool(BaseTool):
9    name = "weather"
10    description = "Fetches weather information"
11    args_schema = WeatherInput
12    
13    def _run(self, city: str, unit: str = "celsius") -> str:
14        # API call
15        weather_data = fetch_weather(city)
16        return f"{city}: {weather_data['temp']}°{'C' if unit == 'celsius' else 'F'}"
17    
18    async def _arun(self, city: str, unit: str = "celsius") -> str:
19        return await async_fetch_weather(city)
20## 메모리 시스템
21
22### Conversation Buffer Memory
23
24```python
25from langchain.memory import ConversationBufferMemory
26
27memory = ConversationBufferMemory(
28    memory_key="chat_history",
29    return_messages=True
30)
31
32# Add to Agent
33agent_executor = AgentExecutor(
34    agent=agent,
35    tools=tools,
36    memory=memory,
37    verbose=True
38)

Summary Memory

긴 대화를 위한 요약 메모리:

1from langchain.memory import ConversationSummaryMemory
2
3memory = ConversationSummaryMemory(
4    llm=llm,
5    memory_key="chat_history"
6)

Vector Store Memory

시맨틱 검색을 활용한 장기 메모리:

1from langchain.memory import VectorStoreRetrieverMemory
2from langchain.vectorstores import Chroma
3
4vectorstore = Chroma(embedding_function=embeddings)
5retriever = vectorstore.as_retriever(search_kwargs={"k": 5})
6
7memory = VectorStoreRetrieverMemory(retriever=retriever)

멀티 에이전트 시스템

Supervisor 패턴

1from langgraph.graph import StateGraph, END
2
3def supervisor_agent(state):
4    """Supervisor distributing tasks"""
5    task = state["task"]
6    
7    if "research" in task:
8        return "researcher"
9    elif "write" in task:
10        return "writer"
11    else:
12        return "general"
13
14def researcher_agent(state):
15    """Agent conducting research"""
16    # Research logic
17    return {"research_results": results}
18
19def writer_agent(state):
20    """Agent writing content"""
21    # Writing logic
22    return {"content": content}
23
24# Create Graph
25workflow = StateGraph(AgentState)
26workflow.add_node("supervisor", supervisor_agent)
27workflow.add_node("researcher", researcher_agent)
28workflow.add_node("writer", writer_agent)
29
30workflow.add_conditional_edges("supervisor", route_to_agent)
31workflow.add_edge("researcher", "supervisor")
32workflow.add_edge("writer", END)
33
34app = workflow.compile()

협업 에이전트

1┌─────────────────────────────────────────────────────┐
2│                    Orchestrator                      │
3└───────────────────────┬─────────────────────────────┘
4                        │
5        ┌───────────────┼───────────────┐
6        │               │               │
7┌───────▼───────┐ ┌─────▼─────┐ ┌──────▼──────┐
8│   Researcher  │ │  Analyst  │ │   Writer    │
9│    Agent      │ │   Agent   │ │   Agent     │
10└───────────────┘ └───────────┘ └─────────────┘

계획 전략

Plan-and-Execute

1from langchain_experimental.plan_and_execute import (
2    PlanAndExecute,
3    load_agent_executor,
4    load_chat_planner
5)
6
7planner = load_chat_planner(llm)
8executor = load_agent_executor(llm, tools, verbose=True)
9
10agent = PlanAndExecute(planner=planner, executor=executor)
11
12result = agent.run("Research and create a draft for a blog post")

작업 분해

1Main Task: "Analyze products for an e-commerce site"
2
3Sub-Tasks:
41. Research competitor products
52. Perform price comparison
63. Analyze customer reviews
74. Generate report
8
9Each sub-task is handled by an agent.
10## 오류 처리 및 재시도
11
12```python
13from langchain.callbacks import get_openai_callback
14
15class RobustAgentExecutor:
16    def __init__(self, agent_executor, max_retries=3):
17        self.executor = agent_executor
18        self.max_retries = max_retries
19    
20    def invoke(self, input_dict):
21        for attempt in range(self.max_retries):
22            try:
23                with get_openai_callback() as cb:
24                    result = self.executor.invoke(input_dict)
25                    
26                    # Token limit check
27                    if cb.total_tokens > 10000:
28                        self.summarize_memory()
29                    
30                    return result
31                    
32            except Exception as e:
33                if attempt == self.max_retries - 1:
34                    raise
35                
36                # Strategy based on Error
37                if "rate_limit" in str(e):
38                    time.sleep(60)
39                elif "context_length" in str(e):
40                    self.truncate_context()
41                else:
42                    time.sleep(2 ** attempt)

스트리밍 및 실시간 처리

1from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler
2
3llm = ChatOpenAI(
4    streaming=True,
5    callbacks=[StreamingStdOutCallbackHandler()]
6)
7
8async def stream_agent_response(query):
9    async for event in agent_executor.astream_events(
10        {"input": query},
11        version="v1"
12    ):
13        kind = event["event"]
14        if kind == "on_chat_model_stream":
15            print(event["data"]["chunk"].content, end="")
16        elif kind == "on_tool_start":
17            print(f"\n🔧 Using tool: {event['name']}")

프로덕션 고려사항

로깅 및 모니터링

1import logging
2
3class AgentLogger:
4    def __init__(self):
5        self.logger = logging.getLogger("agent")
6    
7    def log_action(self, action, input_data, output):
8        self.logger.info({
9            "action": action,
10            "input": input_data,
11            "output_length": len(str(output)),
12            "timestamp": datetime.now().isoformat()
13        })
14    
15    def log_error(self, error, context):
16        self.logger.error({
17            "error": str(error),
18            "context": context
19        })

비용 관리

1def estimate_cost(agent_run):
2    # Cost per token (GPT)
3    input_cost = agent_run.input_tokens * 0.00003
4    output_cost = agent_run.output_tokens * 0.00006
5    
6    return input_cost + output_cost
7
8# Budget limit
9MAX_BUDGET_PER_REQUEST = 0.50  # $0.50
10
11if estimate_cost(run) > MAX_BUDGET_PER_REQUEST:
12    raise BudgetExceededError()

결론

AI Agents는 인공지능의 자율적이고 선제적인 활용을 가능하게 합니다. LangChain과 LangGraph 같은 프레임워크는 프로덕션 수준의 에이전트 시스템 개발을 쉽게 만들어 줍니다.

Veni AI는 엔터프라이즈 에이전트 솔루션을 개발하고 있습니다.