# 1. import

import os

import json

import time

import logging

import inspect

from pathlib import Path

from datetime import datetime

from dataclasses import dataclass, field

from typing import Any

from dotenv import load_dotenv

from openai import OpenAI

# ============================================================

# 2. Environment settings

# ============================================================

# .env 파일을 환경변수로 읽기

load_dotenv()

MODEL_NAME = os.environ.get("MODEL_NAME", "gpt-4o-mini",)

MAX_STEPS = int(os.environ.get("MAX_STEPS", "10",))

DATA_DIR = Path("agent_data")

# API에 유지할 최근 대화 블록 수: block 하나 = 모델응답 + function call + function_call_output

MAX_CONTEXT_BLOCKS = int(os.environ.get("MAX_CONTEXT_BLOCKS", "8",))

# Tool 결과가 너무 길어지는 것을 방지

MAX_TOOL_OUTPUT_CHARS = int(os.environ.get("MAX_TOOL_OUTPUT_CHARS", "4000",))

# Retry 대기시간

RETRY_DELAYS = [1, 2, 4]

# ============================================================

# 3. Logging

# ============================================================

logging.basicConfig(level=logging.INFO, format="%(levelname)-7s %(message)s")

log = logging.getLogger("agent")

# ============================================================

# 4. OpenAI Client

# ============================================================

client = OpenAI()

# ============================================================

# 5. Tool Result

# ============================================================

@dataclass

class ToolResult:

"""모든 Tool 실행 결과를, 동일한/일정한 구조로 관리하기 위한 클래스"""

ok:bool

output:str=""

error:str | None=None

meta:dict=field(default_factory=dict)

# ============================================================

# 6. TOOLS

# ============================================================

# ------------------------------------------------------------

# 6.1 calculator

# ------------------------------------------------------------

def calculator(expr:str)->str:

"""

간단한 산술 계산을 수행합니다.

예:20*5, (24*7)+100

숫자와 +-*/(). 만 사용할 수 있습니다.

"""

allowed = set("0123456789+-*/().")

#공백제거

expr = expr.replace(" ","")

if not expr:

return "ERROR: 빈 계산식입니다."

if not set(expr) <= allowed:

return ("ERROR: 숫자와 +-*/(). 만 사용할 수 있습니다.")

try:

#builtins 제거 #eval은 문자열을 코드로 실행하므로, 입력이 신뢰할 수 없으면 보안 취약점이 될 수 있슴.

result = eval(expr, {"__builtins__":{}}, {},) #eval(expr, globals, locals)

return str(result)

except ZeroDivisionError:

return "ERROR: ZeroDivisionError"

except SyntaxError:

return "ERROR: SyntaxError"

except Exception as e:

return ("ERROR: {type(e).__name__}: {e}")

# ------------------------------------------------------------

# 6.2 read_file

# ------------------------------------------------------------

def read_file(path:str, max_chars:int=2000)->str:

"""

로컬 텍스트 파일을 읽습니다.

URL에는 사용하지 마십시오.

path: 읽을 파일 경로

max_chars: 최대 읽을 문자 수

"""

try:

p = Path(path)

if not p.exists():

return (f"ERROR: 파일이 없습니다: {path}")

if not p.is_file():

return (f"ERROR: 파일이 아닙니다: {path}")

text=p.read_text(encoding="utf-8")

return text[:max_chars]

except UnicodeDecodeError:

return ("ERROR: UTF-8 텍스트 파일이 아닙니다.")

except Exception as e:

return (f"ERROR: {type(e).__name__}: {e}")

# ------------------------------------------------------------

# 6.3 lookup

# ------------------------------------------------------------

def lookup(topic:str)->str:

"""

내부 업무 지식 기반으로 정보를 조회합니다.

사용 가능한 정보:

- 단가

- 환불 정책

- 배송

"""

facts = {

"환불 정책":"배송 후 30일 이내 환불 가능합니다.",

"단가":"$24",

"배송":"$500 이상 구매 시 무료 배송",

}

topic = topic.strip()

return facts.get(topic, f"ERROR: '{topic}'에 대한 정보가 없습니다.")

# ------------------------------------------------------------

# 6.4 word_count

# ------------------------------------------------------------

def word_count(text:str)->str:

""" 입력된 문장의 단어 수를 계산합니다."""

count = len(text.split())

return str(count)

# ------------------------------------------------------------

# 6.5 search_documents

# ------------------------------------------------------------

# search_documents: 실제 업무용 Agent를 위한 예시 Tool. 현재는 샘플 데이터. 나중에:

# - Vector DB

# - Qdrant

# - PostgreSQL

# - 사내 문서 검색 ... 등으로 교체 가능

# ------------------------------------------------------------

def search_documents(query:str, limit:int=3,)->str:

"""

사내 문서 또는 지식 기반에서 관련 정보를 검색합니다.

query: 검색할 내용

limit: 최대 결과 수

"""

documents = [

{

"title": "환불 정책",

"content":"제품 배송 후 30일 이내 환불 가능합니다."

{

"title": "배송 정책",

"content":"500달러 이상 주문 시 무료 배송됩니다."

{

"title": "가격 정책",

"content":"기본 제품 단가는 개당 24달러입니다."

]

results =[]

query_lower = query.lower()

for doc in documents:

text = (doc["title"]+" "+doc["content"]).lower()

if query_lower in text:

results.append(doc)

results = results[:limit]

if not results:

return (f"검색 결과가 없습니다: {query}")

return json.dumps(results, ensure_ascii=False, indent=2)

# ============================================================

# 7. TOOL REGISTRY

# ============================================================

TOOLS={

"calculator":calculator,

"read_file":read_file,

"lookup":lookup,

"word_count":word_count,

"search_documents":search_documents,

}

# ============================================================

# 8. PYTHON TYPE -> JSON SCHEMA TYPE

# ============================================================

PYTHON_TO_JSON = {

str:"string",

int:"integer",

float:"number",

bool:"boolean",

}

# ============================================================

# 9. PYTHON FUNCTION -> OPENAI TOOL SCHEMA

# ============================================================

# 예시:#

# def calculator(expr: str)

# ↓

# {

# "type": "function",

# "name": "calculator",

# "description": "...",

# "parameters": {

# "type": "object",

# "properties": {

# "expr": {

# "type": "string"

# }

# },

# "required": ["expr"]

# }

# ============================================================

def describe_tool(func)->dict:

"""Python 함수를 분석하여, OpenAI Tool Schema로 변환합니다.

(Python 함수 → OpenAI Tool Schema 변환기)

"""

sig = inspect.signature(func)

#함수 docstring

doc = inspect.getdoc(func) or "No description provided"

#첫번째 줄을 Tool 설명으로 사용

description = (doc.strip().splitlines()[0])

properties = {}

required = []

#함수의 parameter 분석

for name, param in sig.parameters.items():

annotation = param.annotation

#Python type -> JSON type

json_type = PYTHON_TO_JSON.get(annotation, "string",)

properties[name] = {

"type":json_type,

"description":f"{name} parameter"

}

#기본값이 없으면 required

if (param.default is inspect.Parameter.empty):

required.append(name)

else:

properties[name]["default"] = (param.default)

# OpenAI Function Tool Schema

return {

"type":"function",

"name":func.__name__,

"description":description,

"parameters":{

"type":"object",

"properties":properties,

"required":required,

"additionalProperties":False,

}

# ============================================================

# 10. 모든 Tool Schema 생성

# ============================================================

OPENAI_TOOLS = [describe_tool(func) for func in TOOLS.values()]

# ============================================================

# 11. SYSTEM PROMPT

# ============================================================

# 이제 Tool 선택은 정식 Tool Calling이 담당. 따라서 이전처럼

# {"tool":"calculator"...}와 같이, JSON 출력을 강제로 요구할 필요가 없음

# ============================================================

SYSTEM_PROMPT ="""

당신은 신중하고 정확한 업무용 AI Agent입니다.

작업 규칙:

1. 필요한 정보가 없으면 적절한 Tool을 사용하세요.

2. 계산이 필요한 경우 calculator Tool을 사용하세요.

3. 파일 내용을 확인해야 하면 read_file Tool을 사용하세요.

4. 내부 정책이나 가격 정보를 확인해야 하면 lookup Tool을 사용하세요.

5. 문서 검색이 필요하면 search_documents Tool을 사용하세요.

6. Tool 결과를 확인한 후 다음 행동을 결정하세요.

7. 충분한 정보를 얻으면 사용자에게 자연스럽고 명확하게 답변하세요.

8. Tool 결과에 ERROR가 포함되어 있으면, 다른 방법을 시도하거나 사용자에게 문제를 설명하세요.

9. 추측하지 마세요.

10. 필요한 정보를 얻을 수 없으면, 부족한 정보를 명확하게 설명하세요.

""".strip()

# ============================================================

# 12. INTERNAL HISTORY TRIM

# ============================================================

def trim_history(history:list[dict], max_messages:int=30)->list[dict]:

"""

내부 History를 최근 메시지 중심으로 제한합니다.

system message는 항상 유지합니다.

주의:

이 함수는 저장용 / 로그용 History 관리 목적입니다.

"""

if len(history) <= max_messages:

return history

system_messages = [m for m in history if m.get("role")=="system"]

remaining = (max_messages - len(system_messages))

recent_messages = history[-remaining:]

return (system_messages + recent_messages)

# ============================================================

# 13. API CONTEXT BLOCK TRIM

# ============================================================

def build_api_input(initial_user_message:dict, context_blocks:list[list], max_blocks:int,)->list:

"""

실제 OpenAI API 호출용 Context 생성.

구조: initial user message + 최근 context blocks

block 하나는 반드시 함께 유지: response.output + function_call_output

이렇게 하면 Tool Call과 Tool Output이 서로 분리되는 것을 방지할 수 있습니다.

"""

#최근 block만 선택

recent_blocks = context_blocks[-max_blocks:]

api_input = [initial_user_message]

#block들을 순서대로 연결

for block in recent_blocks:

api_input.extend(block)

return api_input

# ============================================================

# 14. 중복 Tool 호출 확인

# ============================================================

# 같은 Tool을...같은 arguments로...반복 실행하는 것을 방지

# ============================================================

def is_repeat_tool_call(tool_name:str, args:dict, recent_calls:set,)->bool:

"""같은 Tool을 같은 arguments로 반복 호출하는 것을 방지"""

signature = (tool_name + ":" + json.dumps(args, sort_keys=True, ensure_ascii=False))

if signature in recent_calls:

return True #방금 만든 지문이 recent_calls 집합에 들어있는지 확인 후, 들어있다면 True (중복됨)를 반환.

recent_calls.add(signature) #처음 보는 지문(signature)이라면 recent_calls 집합에 새롭게 기록해 둡니다.

return False

# ============================================================

# 15. TOOL DISPATCHER

# ============================================================

# dispatch_tool() - Tool 실행 담당자

# ============================================================

def dispatch_tool(tools:dict, name:str, args:dict)->ToolResult:

"""

Tool 실행 전 검증

1. Tool 존재 여부

2. args가 dict인지

3. 예상하지 못한 argument

4. 필수 argument 누락

5. 실제 Tool 실행

"""

# Tool 존재 여부

if name not in tools:

return ToolResult(ok=False, error=(f"Unknown tool:{name}"),)

#args type 검사

if not isinstance(args, dict,):

return ToolResult(ok=False, error=("Tool arguments must be an object."),)

func = tools[name]

sig = inspect.signature(func)

expected = set(sig.parameters)

#에상하지 못한 argument

unknown = (set(args) - expected)

if unknown:

return ToolResult(ok=False, error=(f"Unexpected arguments: {sorted(unknown)}"),)

#필수 argument

required = {

param_name for param_name, param in sig.parameters.items()

if(param.default is inspect.Parameter.empty) # 기본값이 없으면 required

}

missing = (required - set(args))

if missing:

return ToolResult(ok=False, error=(f"Missing arguments: {sorted(missing)}"),)

#실제 Tool 실행

try:

output = func(**args)

return ToolResult(ok=True, output=str(output))

except TypeError as e:

return ToolResult(ok=False, error=(f"Bad arguments:{e}"),)

except Exception as e:

log.exception("Tool execution failed")

return ToolResult(ok=False, error=(f"{type(e).__name__}: {e}"),)

# ============================================================

# 16. RETRY 가능한 ERROR 확인

# ============================================================

def is_retryable_error(error:Exception,)->bool:

"""일시적인 API 오류인지 확인"""

error_name = (type(error).__name__)

retryable_names = {

"RateLimitError",

"APIConnectionError",

"APITimeoutError",

"InternalServerError",

}

return (error_name in retryable_names)

# ============================================================

# 17. MODEL CALL

# Retry + Exponential Backoff

# ============================================================

def call_model(api_input:list,)->Any:

"""

OpenAI Responses API 호출

Retry 가능한 오류 발생 시:

1초

2초

4초

대기 후 재시도

"""

last_error = None

total_attempts = (len(RETRY_DELAYS)+1) #3+1=>4

for attempt in range(1, total_attempts+1): #1,2,3,4. 최대 4번 시도.

try:

log.info(f"model call attempt: {attempt}/{total_attempts}")

response = (client.responses.create(

model=MODEL_NAME,

instructions=SYSTEM_PROMPT,

input=api_input,

tools=OPENAI_TOOLS,

))

return response

except Exception as e:

last_error = e

#재시도 불가능한 오류

if not is_retryable_error(e):

log.error(f"non-retryable error: {type(e).__name__}: {e}")

raise

#마지막 시도

if attempt==total_attempts:

break

wait = RETRY_DELAYS[attempt-1]

log.warning("retryable error: {type(e).__name__} | waiting {wait}s")

#모든 retry 실패 후

raise RuntimeError("Model failed after {total_attempts} attempts: {last_error}")

# ============================================================

# 18. AGENT

# ============================================================

class Agent:

# --------------------------------------------------------

# 18.1 __init__

# --------------------------------------------------------

def __init__(self, tools:dict, max_steps:int=MAX_STEPS):

self.tools = tools

self.max_steps = max_steps

self.steps = 0

# 전체 History: 저장/로그/Transcript용

self.history = []

self.answer = None

# 중복 Tool 호출 확인용

self.recent_tool_calls = set()

# 실제 API Context: [block1, block2, block3, ...]

self.api_context_blocks = []

# --------------------------------------------------------

# 18.2 __repr__

# --------------------------------------------------------

def __repr__(self):

return (f"Agent(steps={self.steps}/{self.max_steps}, msgs={len(self.history)}, api_blocks={len(self.api_context_blocks)})")

# --------------------------------------------------------

# 18.3 add_history

# --------------------------------------------------------

def add_history(self, role:str, content:str, **extra,):

message = {"role":role, "content":content,}

message.update(extra)

self.history.append(message)

# --------------------------------------------------------

# 18.4 run

# --------------------------------------------------------

def run(self, goal:str,)->str:

# ====================================================

# 18.4.1 Agent 상태 초기화

# ====================================================

self.history = []

self.steps = 0

self.answer = None

self.recent_tool_calls.clear()

self.api_context_blocks = []

# 내부 History

self.add_history("system",SYSTEM_PROMPT,)

self.add_history("user", goal,)

#최초 사용자 메시지

initial_user_message = {

"role":"user",

"content":goal,

}

log.info(f"goal: {goal}")

# ====================================================

# 18.4.2 Agent Loop

# ====================================================

while (self.answer is None and self.steps < self.max_steps):

# ------------------------------------------------

# STEP 증가

# ------------------------------------------------

self.steps += 1

log.info(

f"=========="

f"STEP {self.steps}"

f"=========="

)

# ------------------------------------------------

# Internal History Trim

# ------------------------------------------------

active_history = trim_history(self.history)

log.info(f"internal history: {len(active_history)}/{len(self.history)}")

# ------------------------------------------------

# 실제 API Input 생성

# ★ 핵심: 최근 API Context Block만 사용. 즉 History Trim이 실제 API 호출에 적용됨

# ------------------------------------------------

api_input = build_api_input(

initial_user_message,

self.api_context_blocks,

MAX_CONTEXT_BLOCKS,

)

log.info(

f"API context blocks: "

f"{min(len(self.api_context_blocks),MAX_CONTEXT_BLOCKS)}"

f"/"

f"{len(self.api_context_blocks)}"

)

log.info(f"API input items: {len(api_input)}")

# ------------------------------------------------

# MODEL CALL

# ------------------------------------------------

try:

response = call_model(api_input)

except Exception as e:

log.error(f"model failed: {type(e).__name__}: {e}")

self.answer = ("모델 호출 중 오류가 발생했습니다.")

break

# ------------------------------------------------

# Function Call 찾기

# ------------------------------------------------

# 모델이 "이 Tool을 실행해 주세요"라고 요청

# ------------------------------------------------

function_calls = [item for item in response.output if item.type=="function_call"]

# ====================================================

# Tool Call 없음 → 일반 텍스트 최종 답변

# ====================================================

if not function_calls:

self.answer = (response.output_text)

self.add_history("assistant", self.answer)

log.info("finished: normal response")

break

# ====================================================

# Tool Call 있음

# ====================================================

tool_outputs = []

# ------------------------------------------------

# 여러 Tool Call 처리 가능

# ------------------------------------------------

for call in function_calls:

tool_name = (call.name)

# ============================================

# Tool Arguments JSON Parsing

# ============================================

try:

args = json.loads(call.arguments)

except json.JSONDecodeError:

args = {}

result = ToolResult(ok=False, error=("Tool arguments were not valid JSON"),)

# ============================================

# 중복 Tool 검사

# ============================================

else:

if is_repeat_tool_call(tool_name, args, self.recent_tool_calls):

result = ToolResult(

ok=False,

error=("Duplicate tool call detected.Try another approach."),

)

else:

result = dispatch_tool(self.tools, tool_name, args,)

# ============================================

# Tool Result -> Text

# ============================================

if result.ok:

output_text = (result.output)

else:

output_text = ("ERROR: " + str(result.error))

# ============================================

# Tool 결과 길이 제한

# ============================================

if len(output_text) > MAX_TOOL_OUTPUT_CHARS:

output_text = (output_text[:MAX_TOOL_OUTPUT_CHARS] + "\n[TRUNCATED]")

# ============================================

# Logging

# ============================================

log.info(f"tool: {tool_name} | args={args} | result={output_text}")

# ============================================

# 내부 History 저장

# ============================================

# 우리가 배우고 있는 Agent 구조에서는 명확하게 role="tool" 저장

# ========================================

self.add_history(

"assistant", f"Function call:{tool_name}({args})", tool_name=tool_name,

)

self.add_history(

"tool", output_text, tool_name=tool_name, args=args, ok=result.ok,

)

# ============================================

# OpenAI API Tool Output

# ============================================

tool_outputs.append({

"type":"function_call_output",

"call_id":call.call_id,

"output":output_text,

})

# ====================================================

# API Context Block 생성 ★ 매우 중요

# 모델 응답과 Tool 결과를 하나의 Block으로 묶음 .

# 이렇게 해야 Trim할 때 function_call과 function_call_output 이 분리되지 않음

# ====================================================

context_block = []

# 모델이 만든 response output

context_block.extend(response.output)

# Tool 실행 결과

context_block.extend(tool_outputs)

# API Context에 하나의 Block 추가

self.api_context_blocks.append(context_block)

# ====================================================

# 18.4.3 Step Limit

# ====================================================

if self.answer is None:

self.answer = (f"최대단계 {self.max_steps}회에 도달하여 작업을 종료했습니다.")

self.add_history("assistant", self.answer)

# ====================================================

# 18.4.4 Save

# ====================================================

self.save()

return self.answer

# --------------------------------------------------------

# 18.5 save

# --------------------------------------------------------

def save(self):

DATA_DIR.mkdir(parents=True, exist_ok=True,)

stamp=(datetime.now().strftime("%Y%m%d-%H%M%S"))

path = (DATA_DIR/f"run-{stamp}.json")

data = {

"model":MODEL_NAME,

"steps":self.steps,

"answer":self.answer,

"history":self.history,

}

path.write_text(json.dumps(data, indent=2, ensure_ascii=False,),encoding="utf-8",)

log.info(f"saved: {path}")

# ============================================================

# 19. MAIN

# ============================================================

if __name__=="__main__":

print("="*60)

print("AI AGENT START")

print("="*60)

bot = Agent(TOOLS)

result = bot.run(

"""

단가를 확인해서 제품 7개의 총 가격을 계산해 주세요.

반드시 필요한 Tool을 사용하세요.

"""

)

print()

print("-"*60)

print("ANSWER: ")

print(result)

print("-"*60)

print(bot)

이 코드는 지금까지 우리가 단계적으로 발전시켜 온 Agent 중에서 상당히 중요한 전환점입니다.

이전 Agent가:

모델에게 JSON 문자열을 직접 만들어 달라고 요청 → 우리가 파싱 → Tool 실행

이었다면, 현재 코드는:

OpenAI의 정식 Function/Tool Calling → Tool 실행 → Tool 결과를 다시 모델에게 전달 → 다음 판단

구조로 발전했습니다.

특히 이번 코드의 핵심은 단순히 Tool Calling이 아니라 다음 5가지입니다.

① Python 함수 → 자동 Tool Schema 생성
② 모델이 function_call 생성
③ Python이 실제 Tool 실행
④ function_call_output으로 결과를 모델에게 반환
⑤ Context Block 단위로 관리하여 안전하게 History Trim

전체 구조 먼저 보기

전체 프로그램을 아주 크게 보면 이렇게 흘러갑니다.

사용자 질문
    ↓
Agent.run()
    ↓
API Context 생성
    ↓
OpenAI 모델 호출
    ↓
┌──────────────────────────────┐
│ 모델의 판단                  │
├──────────────────────────────┤
│ Tool 필요 → function_call    │
│ Tool 불필요 → 일반 답변      │
└──────────────────────────────┘
        ↓
Python Tool 실행
        ↓
function_call_output 생성
        ↓
Context Block 저장
        ↓
다음 모델 호출
        ↓
최종 답변

이제 1번 import부터 19번 MAIN까지 작은 블록으로 나누어 설명하겠습니다.

1. import

import os
import json
import time
import logging
import inspect

이 다섯 개는 Agent의 핵심 운영 기능에 사용됩니다.

1-1. import os

import os

운영체제의 환경변수를 읽기 위해 사용합니다.

예:

MODEL_NAME = os.environ.get("MODEL_NAME", "gpt-4o-mini")

의미:

환경변수 MODEL_NAME이 있으면 → 그것 사용
없으면 → gpt-4o-mini 사용

이렇게 하면 코드 자체를 수정하지 않고도 실행 환경에서 설정을 바꿀 수 있습니다.

예:

MODEL_NAME=gpt-4o

또는:

MAX_STEPS=20

이런 식으로 운영 환경을 조정할 수 있습니다.

1-2. import json

import json

JSON을 처리합니다.

이번 Agent에서 JSON은 매우 중요합니다.

사용되는 곳:

Tool Arguments

모델이:

{
    "expr": "24*7"
}

을 보내면 Python은:

args = json.loads(call.arguments)

로 Python Dictionary로 변환합니다.

즉:

JSON 문자열
     ↓
json.loads()
     ↓
Python dict

예:

'{"expr":"24*7"}'

↓

{"expr": "24*7"}

1-3. import time

import time

Retry 대기 시간에 사용합니다.

예:

time.sleep(wait)

API 서버가 일시적으로 문제가 있을 때:

1초 대기
↓
재시도

2초 대기
↓
재시도

4초 대기
↓
재시도

이런 방식을 Backoff라고 합니다.

1-4. import logging

import logging

프로그램의 실행 상황을 기록합니다.

예:

log.info("goal: ...")

출력:

INFO    goal: 제품 7개의 가격을 계산해 주세요.

Agent는 일반 프로그램보다 로그가 중요합니다.

왜냐하면 Agent는 여러 단계를 거치기 때문입니다.

모델 호출
↓
Tool 선택
↓
Tool 실행
↓
결과 반환
↓
다시 모델 호출

문제가 생겼을 때 로그가 없으면 어디서 문제가 발생했는지 알기 어렵습니다.

1-5. import inspect

import inspect

이 코드에서 매우 중요한 라이브러리입니다.

Python 함수 자체를 분석합니다.

예:

def calculator(expr: str) -> str:

inspect를 사용하면 프로그램이 자동으로 알아낼 수 있습니다.

함수 이름 → calculator

파라미터 → expr

타입 → str

기본값 → 없음

즉:

inspect.signature(calculator)

를 통해 Python 함수 정보를 자동으로 읽습니다.

이것이 바로 뒤에서 설명할:

Python 함수 → OpenAI Tool Schema 자동 변환

의 핵심입니다.

추가 import

from pathlib import Path
from datetime import datetime
from dataclasses import dataclass, field
from typing import Any
from dotenv import load_dotenv
from openai import OpenAI

1-6. Path

from pathlib import Path

파일과 폴더를 다룹니다.

예:

DATA_DIR = Path("agent_data")

그리고:

DATA_DIR.mkdir(parents=True, exist_ok=True)

는:

agent_data 폴더 생성

입니다.

1-7. datetime

from datetime import datetime

실행 기록 파일에 시간을 붙이기 위해 사용합니다.

예:

stamp = datetime.now().strftime("%Y%m%d-%H%M%S")

결과:

20260830-103015

파일명:

run-20260830-103015.json

1-8. dataclass

from dataclasses import dataclass, field

Tool 실행 결과를 일정한 구조로 관리하기 위해 사용합니다.

뒤에서:

@dataclass
class ToolResult:

로 사용합니다.

1-9. Any

from typing import Any

여러 종류의 객체가 올 수 있을 때 사용합니다.

예:

def call_model(api_input: list) -> Any:

OpenAI API Response 객체의 구체적인 타입을 여기서는 제한하지 않는다는 의미입니다.

1-10. load_dotenv

from dotenv import load_dotenv

.env 파일의 환경변수를 읽습니다.

예를 들어 .env 파일:

OPENAI_API_KEY=sk-...
MODEL_NAME=gpt-4o-mini
MAX_STEPS=10

그리고:

load_dotenv()

하면 프로그램이 환경변수처럼 사용할 수 있습니다.

1-11. OpenAI

from openai import OpenAI

OpenAI API를 사용하기 위한 공식 Client입니다.

2. Environment settings

load_dotenv()

프로그램이 시작될 때 .env 파일을 읽습니다.

MODEL_NAME

MODEL_NAME = os.environ.get(
    "MODEL_NAME",
    "gpt-4o-mini",
)

구조:

환경변수 MODEL_NAME 존재?
        │
       Yes
        ↓
환경변수 값 사용

       No
        ↓
gpt-4o-mini 사용

MAX_STEPS

MAX_STEPS = int(os.environ.get("MAX_STEPS", "10"))

Agent가 최대 몇 번 생각하고 행동할지 결정합니다.

예:

Step 1 → lookup
Step 2 → calculator
Step 3 → finish

이라면:

3 steps 사용

입니다.

무한 루프 방지 장치입니다.

DATA_DIR

DATA_DIR = Path("agent_data")

실행 기록을 저장하는 폴더입니다.

agent_data/
    run-20260830-100000.json
    run-20260830-103000.json

MAX_CONTEXT_BLOCKS

MAX_CONTEXT_BLOCKS = int(
    os.environ.get("MAX_CONTEXT_BLOCKS", "8")
)

이 부분이 이번 Agent에서 가장 중요한 개선 중 하나입니다.

예전에는:

History 전체
↓
그대로 API 전달

이었습니다.

그러면 대화가 길어질수록:

Token 증가
비용 증가
속도 저하
Context overflow 위험

이 발생합니다.

그래서 이제:

최근 Context Block 8개만 API에 전달

합니다.

3. Logging

logging.basicConfig(
    level=logging.INFO,
    format="%(levelname)-7s %(message)s"
)

로그 형식을 지정합니다.

예:

INFO    STEP 1
INFO    tool: lookup
WARNING retryable error
ERROR   model failed

그리고:

log = logging.getLogger("agent")

Agent 전용 Logger를 만듭니다.

4. OpenAI Client

client = OpenAI()

이제 client가 OpenAI 서버와 통신합니다.

예:

client.responses.create(...)

5. ToolResult

@dataclass
class ToolResult:

모든 Tool 결과를 동일한 구조로 관리합니다.

@dataclass
class ToolResult:
    ok: bool
    output: str = ""
    error: str | None = None
    meta: dict = field(default_factory=dict)

예를 들어 성공:

ToolResult(
    ok=True,
    output="168"
)

실패:

ToolResult(
    ok=False,
    error="ZeroDivisionError"
)

이렇게 구조가 통일됩니다.

6. TOOLS

Agent가 사용할 수 있는 실제 기능들입니다.

현재:

calculator
read_file
lookup
word_count
search_documents

6.1 calculator

def calculator(expr: str) -> str:

문자열 형태의 계산식을 받아 계산합니다.

예:

calculator("24*7")

결과:

허용 문자

allowed = set("0123456789+-*/().")

허용:

0~9
+
-
*
/
(
)
.

공백 제거

expr = expr.replace(" ", "")

예:

"24 * 7"

↓

"24*7"

빈 문자열 검사

if not expr:

입력이:

""

이면 오류입니다.

허용되지 않은 문자 검사

if not set(expr) <= allowed:

예:

calculator("hello")

은 실패합니다.

eval

result = eval(
    expr,
    {"__builtins__": {}},
    {}
)

eval()은 문자열을 Python 코드처럼 실행합니다.

예:

eval("24*7")

↓

하지만 일반적인 eval()은 위험할 수 있습니다.

그래서:

{"__builtins__": {}}

로 Python 기본 기능 접근을 제한합니다.

게다가 앞에서:

allowed

검사도 합니다.

6.2 read_file

def read_file(path: str, max_chars: int = 2000) -> str:

로컬 파일을 읽습니다.

예:

read_file("contract.txt")

Path 객체 생성

p = Path(path)

파일 존재 확인

if not p.exists():

실제 파일인지 확인

if not p.is_file():

폴더를 읽으려고 하면 막습니다.

UTF-8로 읽기

text = p.read_text(encoding="utf-8")

길이 제한

return text[:max_chars]

예:

파일 전체: 100,000자
max_chars: 2,000

Agent에게는:

처음 2,000자만 전달

합니다.

6.3 lookup

def lookup(topic: str) -> str:

현재는 간단한 내부 지식 DB입니다.

facts = {
    "단가": "$24",
    "배송": "$500 이상 구매 시 무료 배송"
}

예:

lookup("단가")

결과:

$24

실제 업무에서는 이것을:

PostgreSQL
ERP
CRM
사내 DB
API
Vector DB

등으로 바꿀 수 있습니다.

6.4 word_count

def word_count(text: str) -> str:

입력 문장의 단어 수를 계산합니다.

text.split()

예:

AI Agent는 매우 강력하다

↓

["AI", "Agent는", "매우", "강력하다"]

↓

6.5 search_documents

이 Tool은 앞으로 실제 업무용 Agent로 발전할 때 매우 중요합니다.

현재:

documents = [
    {...},
    {...},
    {...}
]

라는 샘플 데이터입니다.

하지만 나중에는:

사용자 질문
    ↓
Embedding
    ↓
Vector DB
    ↓
Qdrant
    ↓
관련 문서 검색
    ↓
LLM

으로 발전할 수 있습니다.

7. TOOL REGISTRY

TOOLS = {
    "calculator": calculator,
    "read_file": read_file,
    "lookup": lookup,
    "word_count": word_count,
    "search_documents": search_documents,
}

이것은 Agent의 Tool 목록입니다.

구조:

Tool 이름
    ↓
실제 Python 함수

예:

TOOLS["calculator"]

↓

calculator 함수

8. Python Type → JSON Schema Type

PYTHON_TO_JSON = {
    str: "string",
    int: "integer",
    float: "number",
    bool: "boolean",
}

Python과 JSON Schema의 타입 표현이 다르기 때문입니다.

PythonJSON Schema

str	string
int	integer
float	number
bool	boolean

예:

def calculator(expr: str):

↓

{
    "expr": {
        "type": "string"
    }
}

9. Python Function → OpenAI Tool Schema

이 부분이 이번 코드의 핵심 중 하나입니다.

def describe_tool(func) -> dict:

역할:

Python 함수
      ↓
자동 분석
      ↓
OpenAI가 이해할 수 있는 Tool Schema 생성

실제 예제

Python 함수:

def calculator(expr: str) -> str:

이 함수는 사람에게는 이해됩니다.

하지만 OpenAI API에게는 다음처럼 설명해야 합니다.

{
  "type": "function",
  "name": "calculator",
  "description": "간단한 산술 계산을 수행합니다.",
  "parameters": {
    "type": "object",
    "properties": {
      "expr": {
        "type": "string"
      }
    },
    "required": [
      "expr"
    ]
  }
}

describe_tool()이 이것을 자동으로 만듭니다.

함수 Signature 분석

sig = inspect.signature(func)

예:

inspect.signature(calculator)

개념적으로:

(expr: str) -> str

을 얻습니다.

docstring

doc = inspect.getdoc(func)

calculator의 설명:

"""
간단한 산술 계산을 수행합니다.
예:20*5
"""

첫 번째 줄

description = doc.strip().splitlines()[0]

결과:

간단한 산술 계산을 수행합니다.

Parameter 분석

for name, param in sig.parameters.items():

calculator의 경우:

name = expr
param.annotation = str

Python → JSON Type 변환

json_type = PYTHON_TO_JSON.get(annotation, "string")

str
 ↓
"string"

Required 확인

if param.default is inspect.Parameter.empty:

기본값이 없다면:

def calculator(expr: str)

expr는 반드시 필요합니다.

↓

"required": ["expr"]

반면:

def read_file(path, max_chars=2000)

max_chars는 선택사항입니다.

10. 모든 Tool Schema 생성

OPENAI_TOOLS = [
    describe_tool(func)
    for func in TOOLS.values()
]

결과적으로:

calculator schema
read_file schema
lookup schema
word_count schema
search_documents schema

가 만들어집니다.

개념적으로:

OPENAI_TOOLS = [
    {...calculator schema...},
    {...read_file schema...},
    {...lookup schema...},
]

그리고 API 호출 시:

tools=OPENAI_TOOLS

로 모델에게 전달합니다.

11. SYSTEM PROMPT

SYSTEM_PROMPT = """
당신은 신중하고 정확한 업무용 AI Agent입니다.
...
"""

예전 코드와 가장 큰 차이점:

예전에는:

반드시 JSON으로 응답하세요.

{"tool":"calculator", "args": {...}}

를 강제했습니다.

하지만 지금은:

OpenAI 정식 Tool Calling

을 사용하므로 모델이 Tool을 호출할 때 별도의 구조화된 function_call을 생성합니다.

즉:

System Prompt → 행동 규칙
Tool Schema → 사용할 수 있는 능력

으로 역할이 분리되었습니다.

12. INTERNAL HISTORY TRIM

def trim_history(history, max_messages=30):

이것은 내부 기록 관리용입니다.

전체 History:

system
user
assistant
tool
assistant
tool
assistant
tool
...

너무 길어질 수 있습니다.

그래서 최근 것만 남깁니다.

다만:

system_messages = [
    m for m in history
    if m.get("role") == "system"
]

System Prompt는 유지합니다.

중요한 점:

이 함수는 현재 코드에서 실제 API 호출 Context를 자르는 핵심 함수가 아닙니다.

진짜 API Context 관리는 다음 13번이 담당합니다.

13. API CONTEXT BLOCK TRIM

def build_api_input(
    initial_user_message,
    context_blocks,
    max_blocks,
):

이것이 이번 코드의 가장 중요한 구조 개선입니다.

왜 Block이 필요한가?

단순하게 메시지를 자르면 위험합니다.

예:

Assistant:
calculator를 호출하겠습니다.

Function Call:
calculator({"expr":"24*7"})

Tool Result:
168

만약 중간에서 Trim하면:

Tool Result:
168

만 남고:

어떤 Tool의 결과인지
왜 실행했는지
어떤 계산이었는지

를 잃을 수 있습니다.

그래서 하나의 작업 단위를:

┌──────────────────────────────┐
│ Context Block               │
│                              │
│ response.output             │
│   └ function_call            │
│                              │
│ function_call_output         │
└──────────────────────────────┘

으로 묶습니다.

예:

Block 1
 ├─ 모델 응답
 │   └─ lookup("단가")
 │
 └─ Tool 결과
     └─ "$24"

Block 2
 ├─ 모델 응답
 │   └─ calculator("24*7")
 │
 └─ Tool 결과
     └─ "168"

이렇게 하면 오래된 Block 전체를 제거할 수 있습니다.

Block 1 삭제
Block 2 유지
Block 3 유지

Tool Call과 Tool Output이 분리되지 않습니다.

14. 중복 Tool 호출 확인

def is_repeat_tool_call(...)

예:

lookup("단가")

를 이미 호출했는데 모델이 또:

lookup("단가")

를 요청하면 중복입니다.

그래서:

signature = (
    tool_name
    + ":"
    + json.dumps(args)
)

예:

lookup:{"topic":"단가"}

를 만들어 Set에 저장합니다.

recent_calls.add(signature)

다음에 같은 호출이 들어오면:

중복 감지

합니다.

15. TOOL DISPATCHER

def dispatch_tool(
    tools,
    name,
    args
) -> ToolResult:

이 함수는 Tool 실행 관리자입니다.

실제 Tool을 바로 실행하지 않고 먼저 검증합니다.

1단계 Tool 존재 확인

if name not in tools:

예:

모델 요청:

delete_database()

하지만 등록된 Tool에 없으면 실행하지 않습니다.

2단계 args 타입 확인

if not isinstance(args, dict):

정상:

{
    "expr": "24*7"
}

비정상:

["24*7"]

3단계 예상하지 않은 Argument

예:

Tool:

def calculator(expr):

그런데 모델이:

{
    "expr": "24*7",
    "currency": "USD"
}

를 보내면:

currency는 예상하지 못한 argument

입니다.

4단계 필수 argument 확인

예:

calculator()

은 expr이 없으므로 실행하면 안 됩니다.

5단계 실제 실행

output = func(**args)

예:

func = calculator
args = {
    "expr": "24*7"
}

결과:

calculator(expr="24*7")

와 같습니다.

16. Retry 가능한 Error 확인

def is_retryable_error(error):

모든 오류를 Retry하면 안 됩니다.

예:

API Key 오류
잘못된 요청
모델 이름 오류

이런 것은 다시 해도 실패합니다.

하지만:

RateLimitError
APIConnectionError
APITimeoutError
InternalServerError

는 일시적인 문제일 수 있습니다.

그래서 Retry합니다.

17. MODEL CALL

def call_model(api_input):

Agent의 두뇌 호출 부분입니다.

Retry 횟수

RETRY_DELAYS = [1, 2, 4]

total_attempts = len(RETRY_DELAYS) + 1

즉:

첫 번째 시도
↓ 실패
1초 대기

두 번째 시도
↓ 실패
2초 대기

세 번째 시도
↓ 실패
4초 대기

네 번째 시도
↓ 실패
최종 실패

OpenAI 호출

response = client.responses.create(
    model=MODEL_NAME,
    instructions=SYSTEM_PROMPT,
    input=api_input,
    tools=OPENAI_TOOLS,
)

여기서 중요한 세 가지:

instructions

Agent의 행동 규칙

input

현재까지 필요한 Context

tools

모델이 사용할 수 있는 능력

18. AGENT

이제 전체 시스템을 실제로 운영하는 부분입니다.

18.1 init

def __init__(self, tools, max_steps=MAX_STEPS):

Agent가 생성될 때 필요한 상태를 준비합니다.

tools

self.tools = tools

Tool Registry 저장.

steps

self.steps = 0

현재 몇 단계 실행했는지 기록합니다.

history

self.history = []

전체 실행 기록입니다.

예:

system
user
assistant
tool
assistant
tool
assistant

recent_tool_calls

self.recent_tool_calls = set()

중복 Tool 호출 방지용입니다.

api_context_blocks

self.api_context_blocks = []

이것이 실제 API Context를 관리하는 핵심입니다.

구조:

[
    Block1,
    Block2,
    Block3
]

18.2 repr

def __repr__(self):

Agent 객체를 출력할 때 보기 쉽게 만듭니다.

예:

Agent(
    steps=3/10,
    msgs=7,
    api_blocks=2
)

18.3 add_history

def add_history(self, role, content, **extra):

내부 History에 기록합니다.

예:

self.add_history(
    "tool",
    "168",
    tool_name="calculator",
    args={"expr":"24*7"},
    ok=True
)

저장 결과:

{
    "role": "tool",
    "content": "168",
    "tool_name": "calculator",
    "args": {
        "expr": "24*7"
    },
    "ok": True
}

18.4 run

이제 실제 Agent가 움직입니다.

18.4.1 초기화

self.history = []
self.steps = 0
self.answer = None
self.recent_tool_calls.clear()
self.api_context_blocks = []

새로운 작업 시작입니다.

System History

self.add_history("system", SYSTEM_PROMPT)

내부 기록에 System Prompt 저장.

User Goal

self.add_history("user", goal)

사용자 질문 저장.

최초 API 메시지

initial_user_message = {
    "role": "user",
    "content": goal,
}

첫 번째 API 호출의 출발점입니다.

Agent Loop

while (
    self.answer is None
    and
    self.steps < self.max_steps
):

계속 반복합니다.

조건:

아직 답이 없음
AND
최대 단계 미만

STEP 증가

self.steps += 1

예:

STEP 1
STEP 2
STEP 3

실제 API Input 생성

api_input = build_api_input(
    initial_user_message,
    self.api_context_blocks,
    MAX_CONTEXT_BLOCKS,
)

예를 들어:

Initial User Message

Block 1
Block 2
Block 3
Block 4
Block 5
Block 6
Block 7
Block 8
Block 9
Block 10

인데 최대:

8 Blocks

라면 API에는:

Initial User Message
Block 3
Block 4
Block 5
Block 6
Block 7
Block 8
Block 9
Block 10

만 전달합니다.

이것이 진짜 API Context Trim입니다.

Model Call

response = call_model(api_input)

모델이 두 가지 중 하나를 합니다.

경우 1: Tool 필요 없음

예:

사용자:
안녕하세요

모델:

안녕하세요. 무엇을 도와드릴까요?

이 경우:

function_calls = []

그래서:

self.answer = response.output_text

으로 끝납니다.

경우 2: Tool 필요

사용자:

제품 7개의 가격을 계산해줘

모델:

먼저 단가를 알아야 한다.

그래서:

function_call

을 생성합니다.

예:

lookup
arguments:
{
    "topic": "단가"
}

1. response.output에는 무엇이 들어오는가?

예를 들어 모델이 Tool을 호출하면 개념적으로:

response.output
[
    reasoning item,
    function_call item
]

또는 일반적으로:

[
    {
        type: "function_call",
        name: "lookup",
        arguments: "{\"topic\":\"단가\"}",
        call_id: "call_abc123"
    }
]

핵심:

response.output

에는 모델이 이번 턴에 만든 구조화된 출력들이 들어 있습니다.

그중:

item.type == "function_call"

인 것을 찾습니다.

2. function_call과 function_call_output의 연결

매우 중요합니다.

모델:

function_call

생성:

call_id = "call_123"

name = "calculator"

arguments = {
    "expr": "24*7"
}

Python:

calculator("24*7")

↓

그리고 API에:

{
    "type": "function_call_output",
    "call_id": "call_123",
    "output": "168"
}

를 보냅니다.

즉:

function_call
       │
       │ call_id = call_123
       ↓
Python Tool 실행
       ↓
function_call_output
       │
       │ call_id = call_123
       ↓
모델이 "이 결과가 내가 요청한 Tool 결과"라고 이해

이 연결이 call_id입니다.

Tool Arguments Parsing

args = json.loads(call.arguments)

예:

'{"expr":"24*7"}'

↓

{
    "expr": "24*7"
}

중복 Tool 검사

if is_repeat_tool_call(...):

같은 Tool:

lookup("단가")

를 계속 반복하면 막습니다.

실제 Tool 실행

result = dispatch_tool(
    self.tools,
    tool_name,
    args
)

Tool Result → Text

성공:

output_text = result.output

실패:

output_text = "ERROR: ..."

Tool Output 길이 제한

if len(output_text) > MAX_TOOL_OUTPUT_CHARS:

긴 문서를 그대로 모델에 넣으면:

비용 증가
Context 증가
응답 속도 저하

가 발생합니다.

그래서:

4000자

로 제한합니다.

내부 History 저장

self.add_history(
    "assistant",
    f"Function call:{tool_name}({args})"
)

그리고:

self.add_history(
    "tool",
    output_text
)

이것은 사람이 나중에 실행 기록을 볼 수 있게 하기 위한 것입니다.

OpenAI API Tool Output 생성

tool_outputs.append({
    "type": "function_call_output",
    "call_id": call.call_id,
    "output": output_text,
})

이것이 정식 Tool Calling의 핵심입니다.

Context Block 생성

context_block = []

먼저:

context_block.extend(response.output)

모델의 Tool 요청을 넣습니다.

그리고:

context_block.extend(tool_outputs)

Tool 결과를 넣습니다.

결과:

Context Block

response.output
   ↓
function_call
   ↓
function_call_output

이것을 하나로 묶습니다.

왜 Context Block이 필요한가?

예를 들어:

Block 1

function_call
lookup("단가")

function_call_output
"$24"

그리고:

Block 2

function_call
calculator("24*7")

function_call_output
"168"

가 있습니다.

Context를 줄일 때:

Block 1 전체 삭제
Block 2 유지

할 수 있습니다.

하지만 메시지를 개별적으로 자르면:

function_call만 삭제
function_call_output만 남음

같은 문제가 생길 수 있습니다.

그래서 Block 구조가 필요합니다.

이것은 현재 Agent 구조에서 아주 좋은 설계입니다.

18.4.3 Step Limit

if self.answer is None:

최대 단계까지 갔는데 답을 못 얻으면:

최대 단계에 도달했습니다.

로 종료합니다.

무한 Agent Loop를 막는 중요한 안전장치입니다.

18.5 save

def save(self):

실행 기록을 파일로 저장합니다.

저장 데이터:

data = {
    "model": MODEL_NAME,
    "steps": self.steps,
    "answer": self.answer,
    "history": self.history,
}

예:

{
    "model": "gpt-4o-mini",
    "steps": 3,
    "answer": "7개의 가격은 168달러입니다.",
    "history": [...]
}

19. MAIN

if __name__ == "__main__":

이 파일을 직접 실행했을 때만 아래 코드가 실행됩니다.

Agent 생성

bot = Agent(TOOLS)

Agent에게 Tool들을 제공합니다.

Agent
 │
 ├── calculator
 ├── read_file
 ├── lookup
 ├── word_count
 └── search_documents

Agent 실행

result = bot.run(...)

질문:

단가를 확인해서 제품 7개의 총 가격을 계산해 주세요.

실제 전체 실행 흐름도

이 질문이 들어왔다고 가정해 보겠습니다.

"단가를 확인해서 제품 7개의 총 가격을 계산해 주세요."

STEP 0: Agent 시작

Agent 생성
    ↓
상태 초기화
    ↓
System Prompt 준비
    ↓
User Goal 준비

STEP 1

API Input:

User:
단가를 확인해서 제품 7개의 총 가격을 계산해 주세요.

↓

모델 판단:

단가 정보가 없다.
lookup Tool 사용해야 한다.

↓

모델 생성:

function_call

name:
lookup

arguments:
{"topic":"단가"}

call_id:
call_001

↓

Python:

lookup("단가")

↓

결과:

$24

↓

Python:

function_call_output

call_id:
call_001

output:
$24

↓

Context Block 1 저장:

Block 1
 ├─ function_call lookup
 └─ function_call_output $24

STEP 2

API Input:

User Goal

Block 1
 ├ lookup("단가")
 └ "$24"

↓

모델 판단:

단가 = $24
수량 = 7

계산 필요
calculator 사용

↓

function_call:

calculator

{
    "expr": "24*7"
}

↓

Python:

calculator("24*7")

↓

결과:

↓

function_call_output:

↓

Block 2 저장:

Block 2
 ├─ calculator("24*7")
 └─ 168

STEP 3

API Input:

User Goal

Block 1
lookup → $24

Block 2
calculator → 168

↓

모델 판단:

정보 충분
Tool 필요 없음
최종 답변

↓

모델:

제품 7개의 총 가격은 168달러입니다.

↓

Agent 종료

전체 그림

┌─────────────────────────────┐
│           USER              │
│ "제품 7개의 가격 계산"       │
└──────────────┬──────────────┘
               │
               ▼
┌─────────────────────────────┐
│         Agent.run()         │
└──────────────┬──────────────┘
               │
               ▼
       ┌───────────────┐
       │ API Context   │
       │ 최근 Blocks   │
       └───────┬───────┘
               │
               ▼
       ┌───────────────┐
       │ OpenAI Model  │
       └───────┬───────┘
               │
        ┌──────┴──────┐
        │             │
        ▼             ▼
   Tool 필요       Tool 불필요
        │             │
        ▼             ▼
 function_call    최종 답변
        │
        ▼
 Python Tool
        │
        ▼
 ToolResult
        │
        ▼
function_call_output
        │
        ▼
 Context Block 생성
        │
        └───────────────┐
                        │
                        ▼
                 다음 Model Call

이 코드에서 특히 중요한 구조 5가지

① Tool Schema 자동 생성

Python 함수
    ↓ inspect
함수명 / parameter / type / default
    ↓
JSON Schema
    ↓
OpenAI Tool 등록

② 정식 Tool Calling

예전:

모델에게 JSON 문자열을 잘 만들어 달라고 부탁

현재:

OpenAI API의 공식 function_call 구조 사용

훨씬 안정적입니다.

③ call_id 연결

function_call
    │
    │ call_id
    ▼
Python Tool 실행
    │
    ▼
function_call_output
    │
    │ 동일한 call_id
    ▼
모델

④ Context Block

Model Response
+
Function Call
+
Tool Output

을 하나의 작업 단위로 관리합니다.

그래서 Context Trim 시 관계가 깨지지 않습니다.

⑤ 실제 업무용 Agent의 기본 구조

현재 구조는 앞으로 다음과 같이 발전시키기 좋은 형태입니다.

현재
│
├── lookup
├── calculator
├── read_file
└── search_documents
        │
        ▼
다음 단계
│
├── PostgreSQL
├── Qdrant
├── 사내 문서
├── CRM
├── ERP
├── 이메일
├── 웹 검색
├── 한컴 문서 자동화
└── 업무 Workflow

마지막으로, 현재 코드에서 제가 가장 중요하게 보는 부분

지금 코드의 핵심적인 발전은 단순한 챗봇이 아니라 Agent Loop의 기본 골격을 제대로 갖추기 시작했다는 것입니다.

특히:

Model
  ↓ 판단
Tool Call
  ↓
Python 실행
  ↓
Tool Output
  ↓
Model 재판단

이라는 구조를 갖췄고,

여기에:

Retry
Error Handling
Tool Validation
Duplicate Tool Call 방지
Output 길이 제한
Context Block
Context Trim
Transcript 저장

까지 들어왔습니다.

이제 다음 단계에서는 이 구조를 더욱 실제 업무용으로 발전시키는 것이 좋습니다. 특히 현재 Agent에 가장 먼저 추가할 만한 것은 Resume(이전 작업 이어가기), Tool 권한/보안 정책, 작업 계획(Planning), 그리고 실제 사내 문서·DB 연결입니다.

저작자표시 비영리 변경금지 (새창열림)

'AI, 클라우드, 문서, 자동화 > AI_AGENT' 카테고리의 다른 글

XVI. 작은 AI Agent에 5가지 기능을 추가한 프로그램(끝) (0)	2026.08.26
XV. 작은 AI Agent 프로그램(실제 LLM -OpenAI 사용) (0)	2026.08.26
XIV. 작은 AI Agent 프로그램(가짜 LLM Model-script 사용) (0)	2026.08.26
XIII. Python 함수 자체를 분석하여, AI가 사용할 수 있는 Tool 설명서(schema)를 자동으로 만드는 방법 (0)	2026.08.25
XII. Python 프로그램이 인터넷 너머의 API 서버와 대화하는 방법 → 오류, 재시도 ...→ 결국 실제 LLM 호출 함수 call_model()로 발전시키는 과정 (2)	2026.08.25

XVII. AI Agent Engine 기본 완성: Python · Agent Loop · Tool Dispatcher · Function Call · Context Block · History Trim

전체 구조 먼저 보기

1. import

1-1. import os

1-2. import json

Tool Arguments

1-3. import time

1-4. import logging

1-5. import inspect

추가 import

1-6. Path

1-7. datetime

1-8. dataclass

1-9. Any

1-10. load_dotenv

1-11. OpenAI

2. Environment settings

MODEL_NAME

MAX_STEPS

DATA_DIR

MAX_CONTEXT_BLOCKS

3. Logging

4. OpenAI Client

5. ToolResult

6. TOOLS

6.1 calculator

허용 문자

공백 제거

빈 문자열 검사

허용되지 않은 문자 검사

eval

6.2 read_file

Path 객체 생성

파일 존재 확인

실제 파일인지 확인

UTF-8로 읽기

길이 제한

6.3 lookup

6.4 word_count

6.5 search_documents

7. TOOL REGISTRY

8. Python Type → JSON Schema Type

9. Python Function → OpenAI Tool Schema

실제 예제

함수 Signature 분석

docstring

첫 번째 줄

Parameter 분석

Python → JSON Type 변환

Required 확인

10. 모든 Tool Schema 생성

11. SYSTEM PROMPT

12. INTERNAL HISTORY TRIM

13. API CONTEXT BLOCK TRIM

왜 Block이 필요한가?

14. 중복 Tool 호출 확인

15. TOOL DISPATCHER

1단계 Tool 존재 확인

2단계 args 타입 확인

3단계 예상하지 않은 Argument

4단계 필수 argument 확인

5단계 실제 실행

16. Retry 가능한 Error 확인

17. MODEL CALL

Retry 횟수

OpenAI 호출

instructions

input

tools

18. AGENT

18.1 __init__

tools

steps

history

recent_tool_calls

api_context_blocks

18.2 __repr__

18.3 add_history

18.4 run

18.4.1 초기화

18.1 init

18.2 repr