AI 에이전트의 실체 파헤치기: 순수 Python으로 에이전틱 파이프라인(Agentic Pipeline) 처음부터 구축하기

대부분의 AI 데모는 간단한 질문을 던지기 전까지는 매우 인상적으로 보입니다. 즉, 내부적으로 실제로 어떤 일이 일어나고 있는가 하는 질문 말입니다. LangChain, CrewAI, Microsoft AutoGen과 같은 프레임워크들은 단 몇 줄의 코드로 "AI 에이전트 (AI agent)"를 매우 쉽게 만들어냅니다. 하지만 추상화에는 대가가 따릅니다. 많은 개발자가 에이전트를 구동하는 런타임 아키텍처 (runtime architecture)를 완전히 이해하지 못한 채 프레임워크를 사용하여 에이전트를 구축하곤 합니다. 본질적으로 대부분의 에이전트 프레임워크는 놀라울 정도로 단순한 기본 요소들을 중심으로 구축되어 있습니다: 프롬프트 오케스트레이션 (Prompt orchestration), 상태 유지 메모리 (Stateful memory), 도구 실행 (Tool execution), 제어 루프 (Control loops), 구조화된 출력 (Structured outputs). 이번 주에 저는 AI 에이전트가 실제로 내부에서 어떻게 작동하는지 알고 싶어 하는 한 친구와 이야기를 나누었습니다. 그 대화 도중 저는 한 가지를 깨달았습니다. 대부분의 튜토리얼이 AI 에이전트를 실제보다 훨씬 더 신비로운 것처럼 느끼게 만든다는 사실입니다. 프레임워크는 빠르게 움직이는 데는 훌륭하지만, 추상화 계층 뒤에 많은 핵심 메커니즘을 숨기기도 합니다. 라이브러리를 임포트(import)하고, "에이전트 (agent)"를 초기화하고, 도구를 연결하면 갑자기 모든 것이 자율적이고 지능적으로 보입니다. 하지만 이러한 추상화 아래에서 대부분의 에이전트 시스템은 놀라울 정도로 적은 개념의 집합 위에 구축되어 있습니다: 프롬프트 (Prompts), 메모리 (Memory), 도구 실행 (Tool execution), 구조화된 출력 (Structured outputs), 제어 루프 (Control loops). 그래서 저는 제가 에이전틱 시스템 (agentic systems)을 처음 탐구하기 시작했을 때 찾았더라면 좋았을 법한 글을 쓰기로 결심했습니다. 무거운 프레임워크는 없습니다. 오케스트레이션 라이브러리도 없습니다. 숨겨진 런타임 마법도 없습니다. 오직 순수 Python (pure Python)으로 처음부터 단계별로 구축한 핵심 아이디어뿐입니다. 이 글에서 우리는 추상화를 걷어내고 다음과 같은 것들을 사용하여 프로덕션에서 영감을 받은 에이전틱 파이프라인 (agentic pipeline)을 완전히 처음부터 구축할 것입니다: 순수 Python, 표준 라이브러리 (standard library)만 사용, 네이티브 HTTP 요청 (Native HTTP requests), SDK 미사용, 오케스트레이션 프레임워크 미사용. 이 글을 마칠 때쯤 여러분은 현대 AI 에이전트 뒤에 숨겨진 핵심 메커니즘과 왜 대부분의 프레임워크가 본질적으로 결정론적 실행 루프 (deterministic execution loop) 위에 계층화된 편의용 추상화인지 이해하게 될 것입니다. 에이전틱 파이프라인 (Agentic Pipeline)이란 무엇인가?

표준적인 LLM 상호작용은 대개 단발성 트랜잭션(single-shot transaction)입니다: 사용자 프롬프트 (User Prompt) ──> 모델 응답 (Model Response). 모델은 컨텍스트 (context)를 한 번 수신하고 정적인 응답을 생성합니다. 반면, 에이전트 (agent)는 다르게 동작합니다. 단일 응답을 생성하는 대신, 다음과 같은 연속적인 실행 사이클 (continuous execution cycle) 내에서 작동합니다:

┌───────────────────────────────────────┐
│ │ ▼ │ [ THINK ] ───> (Decision) ───> [ ACT ] ───> [ OBSERVE ]
│ (Tool Call) (Tool Result) │
└───────────────────────────────────────┘

생각하기 (Think): 모델은 사용자의 목표, 사용 가능한 도구 (tools), 이전의 관찰 결과 (observations), 그리고 현재의 메모리 상태 (memory state)를 평가합니다. 그런 다음 다음에 무엇을 할지 결정합니다.

행동하기 (Act): 에이전트는 행동을 실행합니다. 이는 함수 호출 (function calling), 데이터베이스 쿼리 (querying a database), 웹 검색 (searching the web), 파일 읽기 (reading files), 또는 최종 답변 반환이 될 수 있습니다.

관찰하기 (Observe): 시스템은 행동의 결과를 포착하여 이를 컨텍스트 윈도우 (context window)에 다시 입력합니다. 이 사이클은 목표가 완료될 때까지 반복됩니다.

유용한 사고 모델 (A Helpful Mental Model)

에이전트를 운영 환경의 문제를 디버깅하는 개발자처럼 생각해보세요:

에러 로그 관찰 (Observe error logs) │ ▼
가설 수립 (Form a hypothesis) │ ▼
명령어 실행 (Run a command) │ ▼
출력 검사 (Inspect output) │ ▼
반복 (Repeat)

이러한 반복적인 피드백 루프 (iterative feedback loop)가 바로 에이전틱 시스템 (agentic systems)이 작동하는 방식입니다.

프로젝트 구조 (Project Structure)

우리는 코드베이스를 작고 집중된 모듈 (modules)로 구성할 것입니다.

agentic-pipeline/
├── config.json # 런타임 설정 (Runtime configuration)
├── llm_client.py # 저수준 HTTP 클라이언트 (Low-level HTTP client)
├── memory.py # 컨텍스트/상태 관리자 (Context/state manager)
├── agent.py # 에이전트 오케스트레이션 엔진 (Agent orchestration engine)
└── main.py # 런타임 실행 루프 (Runtime execution loop)

이러한 분리는 실제 운영 시스템 (production systems)이 일반적으로 구조화되는 방식을 반영합니다.

1단계 — 설정 관리 (Configuration Management)

런타임 변수를 코드에 직접 하드코딩 (hardcoding)하지 마세요. 이번 데모를 위해, 오직 시연 목적으로만 config.json 파일을 생성하겠습니다:

{
  "llm" : {
    "provider" : "openai",
    "model" : "gpt-4o",
    "api_key" : "sk-your-api-key",
    "temperature" : 0.2,
    "max_tokens" : 1024
  }
}

⚠️ 참고: 운영 시스템 (production systems)에서는 자격 증명 (credentials)을 정적 설정 파일이 아닌 환경 변수 (environment variables)나 시크릿 매니저 (secrets manager)로부터 가져와야 합니다.

2단계 — 인프라 계층 (Infrastructure Layer) 구축하기

대부분의 SDK는 모든 LLM 상호작용이 단순한 HTTP 요청일 뿐이라는 현실을 숨깁니다. 추상화 아래에서 프로세스는 매우 간단합니다: 페이로드 (payload) 직렬화 ──> HTTPS POST 요청 전송 ──> JSON 응답 수신 ──> 출력 파싱 (Parse).

llm_client.py에서 이를 수동으로 구현해 보겠습니다.

import json
import urllib.request
import urllib.error
from typing import Dict, List

class LLMClient:
    def __init__(self, config: Dict):
        self.config = config["llm"]
        self.api_key = self.config["api_key"]

    def chat_completion(self, messages: List[Dict], temperature: float = None) -> str:
        payload = {
            "model": self.config["model"],
            "messages": messages,
            "temperature": temperature or self.config.get("temperature", 0.2),
            "max_tokens": self.config.get("max_tokens", 1024)
        }
        data = json.dumps(payload).encode("utf-8")
        req = urllib.request.Request("https://api.openai.com/v1/chat/completions", data=data, method="POST")
        req.add_header("Content-Type", "application/json")
        req.add_header("Authorization", f"Bearer {self.api_key}")

        try:
            with urllib.request.urlopen(req) as response:
                result = json.loads(response.read().decode())
                return result["choices"][0]["message"]["content"].strip()
        except urllib.error.HTTPError as e:
            error_body = e.read().decode()
            raise Exception(f"LLM API error: {e.code} - {error_body}")

여기서 LLMClient가 무엇을 하고 있는지 이해하려면, 이를 구식 전신 기사 (telegraph operator)처럼 생각하는 것이 도움이 됩니다. 이 계층에는 추론 (reasoning), 계획 (planning), 또는 도구 실행 (executing tools)에 대한 개념이 전혀 없습니다. 메모리 (memory)조차 관리하지 않습니다. 이 계층의 유일한 임무는 텍스트 더미를 패키징하여 모델로 전송하고, 가공되지 않은 응답 (raw response)을 다시 전달하는 것입니다. 이 계층은 메시지 안에 적힌 단 한 마디도 이해할 필요 없이, 메시지를 안정적으로 주고받는 역할만 수행합니다.

Step 3 — 에이전트 메모리 관리 (Managing Agent Memory)

LLM은 상태가 없는 (stateless) 특성을 가집니다. 모든 요청 시 전체 이력을 다시 전달하지 않는 한, 이전의 상호작용을 기억하지 못합니다. 실행 루프가 진행됨에 따라 컨텍스트 윈도우 (context window)는 지속적으로 커집니다. 따라서 우리는 memory.py에 경량 메모리 관리자를 구현할 필요가 있습니다.

from typing import List, Dict

class AgentMemory:
    def __init__(self, max_messages: int = 20):
        self.messages: List[Dict] = []
        self.max_messages = max_messages

    def add(self, role: str, content: str):
        self.messages.append({"role": role, "content": content})
        if len(self.messages) > self.max_messages:
            # 시스템 프롬프트 (system prompt) 보존
            system_prompt = self.messages[0]
            # 대화창 슬라이딩 (Slide conversation window)
            active_history = self.messages[1:]
            self.messages = ([system_prompt] + active_history[-(self.max_messages - 1):])

    def get_messages(self) -> List[Dict]:
        return self.messages.copy()

    def clear(self):
        self.messages.clear()

LLM 클라이언트가 우리의 전신 기사라면, 이 메모리 관리자는 탐정의 수첩과 같다고 상상할 수 있습니다. 에이전트가 작업을 조사함에 따라, 사용자의 원래 요청, 내부 추론 (internal reasoning), 도구 선택 (tool choices), 그리고 그 과정에서 발견된 단서 등 모든 미세한 세부 사항이 기록됩니다. 수첩의 페이지가 무한할 수는 없기에, 탐정은 결국 핵심 조사 맥락을 중심에 두면서 오래된 세부 사항들을 아카이브(archive)해야 합니다. 이러한 슬라이딩 윈도우 (sliding window) 로직이 바로 우리가 컨텍스트를 관리 가능한 수준으로 유지하는 방법입니다.

Step 4 — 에이전트 엔진 구축 (Building the Agent Engine)

이곳은 오케스트레이션 (orchestration) 로직이 존재하는 곳입니다. 에이전트는 사용 가능한 도구를 이해하고, 언제 도구를 사용할지 결정하며, 구조화된 출력 (structured outputs)을 파싱하고, 함수를 실행하며, 관찰 결과 (observations)를 다시 메모리에 입력해야 합니다. agent.py를 작성해 봅시다:

from llm_client import LLMClient
from memory import AgentMemory
from typing import Dict, Callable
import json

class Agent:
    def __init__(self, system_prompt: str, config_path: str = "config.json"):
        with open(config_path) as f:
            self.

config = json.loads(f)
self.llm = LLMClient(self.config)
self.memory = AgentMemory()
self.system_prompt = system_prompt
self.tools: Dict[str, dict] = {}
self.memory.add("system", system_prompt)
def register_tool(self, name: str, func: Callable, description: str):
    self.tools[name] = {"func": func, "description": description}
def _get_tool_descriptions(self) -> str:
    if not self.tools:
        return "No tools available."
    return "\n".join([f"- {name}: {info['description']}" for name, info in self.tools.items()])
def think(self, user_input: str) -> str:
    self.memory.add("user", user_input)
    messages = self.memory.get_messages()
    tool_info = self._get_tool_descriptions()
    if self.tools:
        messages = messages.copy()
        enhanced_content = (f"{user_input}\n\n" f"AVAILABLE TOOLS:\n" f"{tool_info}\n\n" f"If you need a tool, respond ONLY with JSON:\n" f'{{"tool": "{tool_name}", "args":{{}}}}\n\n' f"If the task is complete, respond naturally and include 'FINAL ANSWER' .")
        messages[-1]["content"] = enhanced_content
    response = self.llm.chat_completion(messages)
    self.memory.add("assistant", response)
    return response
def act(self, response: str):
    if "{": in response and "}" in response:
        try:
            start = response.find("{`)
            end = response.rfind("}") + 1
            tool_json = json.loads(response[start:end])
            tool_name = tool_json.get("tool")
            args = tool_json.get("args", {})
            if tool_name in self.tools:
                result = self.tools[tool_name]"<a href="**args">func</a>"
                self.memory.add("system", f"Observation from '{tool_name}': {result}")
                return result
        except Exception as e:
            error_msg = f"Tool execution failed: {str(e)}"
            self.memory.

add("system", error_msg)
            return error_msg
        return None

이러한 구조적 핸드오프(handoff)는 현대 AI 에이전트에서 가장 오해받는 부분 중 하나를 시사합니다. 모델은 사용자의 Python 함수를 직접 실행하지 않습니다. 대신, 여러분은 프롬프트(prompt) 내에 로컬 코드에 대한 일반 텍ast 설명을 제공하는 것입니다. 모델이 이 설명을 읽고 도움이 필요하다고 판단하면, 단순히 도구 이름과 매개변수를 지정하는 가공되지 않은 JSON 블록 형태로 텍스트 출력을 형식화(format)합니다. 그러면 호스트 애플리케이션(host application)이 해당 JSON을 포착하여 읽고, 로컬에서 네이티브 Python 코드를 실행한 뒤, 그 결과를 다시 텍스트 히스토리(text history)로 전달합니다. LLM 자체는 완전히 격리된 상태로 유지되며, 여러분의 로컬 애플리케이션이 실제 실행 환경(execution environment) 역할을 수행합니다.

단계 5 — 런타임 제어 루프 (The Runtime Control Loop)
런타임 루프(runtime loop)가 없다면 에이전트는 다단계 추론(multi-step reasoning)을 수행할 수 없습니다. 호스트 애플리케이션이 실행을 지속적으로 추진해야 합니다. main.py를 살펴보겠습니다:

```python
from agent import Agent
import time

def web_search(query: str) -> str:
    print(f" 🔍 Searching index for: '{query}'")
    time.sleep(1)
    if "agentic ai" in query.lower():
        return (
            "Found: Modern agentic systems are moving away from rigid chains "
            "toward lightweight control loops and modular tools."
        )
    return (
        "Found: Building agents from scratch reveals implementation details "
        "often hidden by frameworks."
    )

if __name__ == "__main__":
    system_prompt = (
        "You are an autonomous operations assistant. "
        "Reason step-by-step. "
        "Use tools when necessary. "
        "When the task is fully complete, include the phrase FINAL ANSWER."
    )
    agent = Agent(system_prompt)
    agent.register_tool(
        name="search",
        func=web_search,
        description="Queries an index database. Input schema: { 'query': str}"
    )
    task = "Research trends in agentic AI and explain why building from scratch is valuable."

print(f" 🎯 Objective: {task} ")
max_steps = 5
for step in range(max_steps):
print(f"
[Cycle {step + 1}] ")
prompt = task if step == 0 else "Analyze previous observations and continue. "
response = agent.think(prompt)
print(f"
🤖 Agent:
{response}")
tool_output = agent.act(response)
if tool_output:
print(f"
🛠 Observation:
{tool_output}")
if "final answer" in response.lower():
print("
✅ Objective completed.")
break
Tracing the Runtime Execution
Here is a look at what happens internally during execution over two separate cycles:
Cycle 1 Think: The model receives the task, tool descriptions, and the initial system memory state. It realizes it lacks direct information about current trends. Act: The model emits structured JSON: { "tool" : "search" , "args" : { "query" : "latest trends in agentic AI" } } plaintext
The runtime parses this block and executes the local Python web_search function.

• Observe: The tool