XVI. 작은 AI Agent에 5가지 기능을 추가한 프로그램(끝)

 

    #-------- 1. Tools

    from pathlib import Path

 

    def calculator(expr:str)->str:

        """(20*4)+15와 같은 간단한 산술 표현식을 계산합니다."""

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

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

        if not expr or not set(expr) <=allowed:

            return "오류: 숫자와 +-*/().만 허용됩니다."

        try:

            return str(eval(expr))

        except (SyntaxError, ZeroDivisionError) as e:

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

 

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

        """로컬 텍스트 파일을 읽습니다. URL에는 사용하지 마십시오."""

        p=Path(path)

        if not p.exists():

            return f"오류: {path}라는 이름의 파일이 없습니다."

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

        return text[:max_chars]

 

    def lookup(topic:str)->str:

        """내부 지식 기반(facts)에서 사실을 조회합니다."""

        facts={

            "환불 정책":"배송후 30일 이내",

            "단가":"$24",

            "배송":"$500 이상 무료",

        }

        return facts.get(topic.lower(), f"오류: '{topic}'에 대해 알려진 정보가 없습니다.")

 

    # 가. 단어수 세는 툴 추가

    def word_count(text:str)->str:

        """ 글에 있는 단어 수를 세어 보세요."""

        return f"{len(text.split())} 단어"

 

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

 

    #-------- 2. tool_description()

    import inspect

 

    def tool_description(tools:dict)->str:

        lines=[]

        for name, func in tools.items():

            params=",".join(inspect.signature(func).parameters)

            summary=(inspect.getdoc(func) or "").splitlines()[0]

            lines.append(f"- {name}({params}):{summary}")

        return "\n".join(lines)

 

    def build_system_prompt(tools:dict)->str:

        return f"""

    당신은 단계적으로 작업하는 신중한 조력자입니다.(You are a careful assistant that works in steps.)

 

    사용 가능한 도구(Tools available):

    {tool_description(tools)}

 

    JSON 객체 하나만 응답으로 보내주세요. 다른 내용은 포함하지 마세요.(Reply with ONE JSON object and nothing else):

    {{"tool":"<tool name or finish>", "args": {{...}}}}

 

    최종 답변을 얻으면 다음을 사용하세요.(When you have the final answer, use):

    {{"tool":"finish", "args":{{"answer":"..."}}}}

 

    작업 규칙:

    1. 가격이나 비용을 계산해야 하는데 단가가 제공되지 않았다면, 먼저 lookup 도구를 사용하여 단가를 확인하세요.

    2. 숫자 계산이 필요한 경우 직접 계산하지 말고, 반드시 calculator 도구를 사용하세요.

    3. 파일의 내용을 확인해야 하는 경우, read_file 도구를 사용하세요.

    4. Tool 실행 결과는 Observation에서 확인하세요.

    5. Observation 결과를 이용하여 다음 작업을 결정하세요.

    6. 충분한 정보를 얻으면 finish를 사용하세요.

    """

    # print(build_system_prompt(TOOLS))

 

    #-------- 3. Agent

    import inspect

    import json

    import logging

    import os

    from datetime import datetime

    from pathlib import Path

 

    from openai import OpenAI

    client = OpenAI()

 

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

    log = logging.getLogger("agent")

 

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

    DATA_DIR = Path("agent_data")

 

    # 나. 사용량-> 비용 체크(1)

    TOTAL_SENT_CHARS=0

    TOTAL_RECV_CHARS=0

 

    def call_model(messages:list[dict])->dict:

 

        # 나. 사용량-> 비용 체크(2)

        global TOTAL_SENT_CHARS, TOTAL_RECV_CHARS

        sent_str = json.dumps(messages)

        TOTAL_SENT_CHARS += len(sent_str)

 

        try:

            response = client.responses.create(

                model="gpt-4o-mini",

                input=messages,

            )

            # 나. 사용량-> 비용 체크(2)

            TOTAL_RECV_CHARS += len(response.output_text)

            log.info(

                f"[Cost counter] Running Total -> Sent: {TOTAL_SENT_CHARS} chars |"

                f" Recv: {TOTAL_RECV_CHARS} chars"

            )

            return {"ok":True, "text":response.output_text, "error":None,}

           

        except Exception as e:

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

       

 

    # parsing

    def parse_reply(text:str)->dict:

        if not text or not text.strip():

            return {"ok":False, "tool":None, "args":{}, "error":"empty reply"}

 

        start, end = text.find("{"), text.rfind("}")

        if start == -1 or end == -1:

            return {"ok":False, "tool":None, "args":{}, "error":"no JSON found"}

 

        try:

            data = json.loads(text[start:end+1])

        except json.JSONDecodeError as e:

            return {"ok":False, "tool":None, "args":{}, "error":f"bad JSON:{e.msg}"}

 

        return {

            "ok":True,

            "tool":data.get("tool"),

            "args":data.get("args",{}),

            "error":None,

        }

 

    # 다. 최근 실행 히스토리 불러오기(1)

    def load_latest()->list:

        if not DATA_DIR.exists():

            return []

        runs = sorted(DATA_DIR.glob("run-*.json"))

        if not runs:

            return []

        log.info(f"직전 실행부터 다시 시작: {runs[-1].name}")

        data = json.loads(runs[-1].read_text(encoding="utf-8"))

        return data.get("history", [])

 

    # 라. 동일한 툴 중복 호출 방지(무한루프 방지)(1)

    def is_repeat(history:list, tool:str, args:dict)->bool:

        signature = f'"{tool}"'

        args_str = json.dumps(args, sort_keys=True)

        full_sig = f"{signature}:{args_str}"

 

        history_str = json.dumps(history)

        return full_sig in history_str

 

    # 마. 히스토리 길어짐 방지(1)

    def trim_history(history:list, max_turns:int=6)->list:

        if len(history) <= 1+ max_turns:

            return history

        return [history[0]] + history[-max_turns]

 

    # the agent

    class Agent:

        def __init__(self, tools:dict, max_steps:int=MAX_STEPS, resume:bool=False): #resume:bool=False 추가

            self.tools=tools

            self.max_steps=max_steps

            self.steps=0

            # self.history=[] #아래에서 재코딩

            self.answer=None

 

            # 다. 최근 실행 히스토리 불러오기(2)

            if resume:

                self.history = load_latest()

            else:

                self.history = []

 

        def __repr__(self):

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

 

        def add(self, role:str, content:str):

            self.history.append({"role":role, "content":content})

 

        def dispatch(self, name:str, args:dict)->str:

            if name not in self.tools:

                return f"ERROR:unknown tool '{name}'. Available:{','.join(self.tools)}"

            if not isinstance(args, dict):

                return "ERROR: args must be an object"

 

            func = self.tools[name]

            expected = set(inspect.signature(func).parameters)

            unknown = set(args) - expected

            if unknown:

                return f"ERROR:unexpected argument(s): {sorted(unknown)}"  

            required = {name for name, args in inspect.signature(func).parameters.items() if args.default is inspect.Parameter.empty}

            missing = required-set(args)

            if missing:

                return f"ERROR:missing required argument(s): {sorted(missing)}"        

 

            try:

                return str(func(**args))

            except TypeError as e:

                return f"ERROR: bad arguments - {e}"

            except Exception as e:

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

 

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

            # self.add("system", build_system_prompt(self.tools))

            # 이전 기록을 이어서 하는 게 아니라 새로 시작하는 경우 시스템 프롬프트 세팅

            if not self.history:

                self.add("system", build_system_prompt(self.tools))

 

            self.add("user", goal)

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

 

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

                self.steps +=1

               

                # 마. 히스토리 길어짐 방지(2)

                active_history = trim_history(self.history, max_turns=8)

                # reply = call_model(self.history)

                reply = call_model(active_history)

 

                if not reply["ok"]:

                    log.error(f"model call failed: {reply['error']}")

                    break

 

                self.add("assistant", reply["text"])

                parsed = parse_reply(reply["text"])

 

                if not parsed["ok"]:

                    log.warning(f"step {self.steps}: {parsed['error']}")

                    self.add("user",f"That was not valid JSON ({parsed['error']}). Try again.")

                    continue

 

                tool = parsed["tool"]

                args = parsed["args"]

                if tool == "finish":

                    self.answer = args.get("answer","(no answer given)")

                    log.info(f"step {self.steps}: finished")

                    break

 

                # observation = self.dispatch(tool, parsed["args"])

                # 라. 동일한 툴 중복 호출 방지(무한루프 방지)(2)

                if is_repeat(self.history[:-1], tool, args):

                    log.warning(f"step {self.steps}: 중복 툴 콜이 감지되었습니다 ->{tool}")

                    observation = f"ERROR: 이미 '{tool}: {args}를 호출하였습니다. 다른 접근이나 툴을 사용하세요.'"

                else:

                    sig = f"{tool}:{json.dumps(args, sort_keys=True)}"

                    tool_result = self.dispatch(tool, args)

                    observation = f"[Call: {sig} {tool_result}]"

 

                log.info(f"step {self.steps}: {tool} -> {observation}")

                self.add("user", f"Observation: {observation}")

 

            if self.answer is None:

                self.answer = f"Stopped after {self.steps} steps without an answer."

 

            self.save()

            return self.answer

 

        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"

            path.write_text(

                json.dumps(

                    {"steps":self.steps, "answer":self.answer, "history":self.history},

                    indent=2,

                    ensure_ascii=False,

                ),

                encoding="utf-8",

            )

            log.info(f"transcript saved to {path}")

 

    if __name__=="__main__":

        print("="*20 + " TEST 1: 단어수 세기 "+"="*20)

        bot = Agent(TOOLS, resume=False)

        result = bot.run("다음 글의 단어수를 말하세요, 'AI 에이전트는 놀랍고 파워풀하다.'")

        print("-"*50)

        print("ANSWER: ", result)

        print(bot)

 

        print("="*20 + " TEST 2: 이전 대화 제개(Resume) "+"="*20)

        bot = Agent(TOOLS, resume=True)

        result = bot.run("조금전 글의 단어수의 두 배가 얼마인지 말하세요.'")

        print("-"*50)

        print("ANSWER: ", result)

        print(bot)

 

    # ==================== TEST 1: 단어수 세기 ====================

    # INFO    goal: 다음 글의 단어수를 말하세요, 'AI 에이전트는 놀랍고 파워풀하다.'

    # INFO    HTTP Request: POST https://api.openai.com/v1/responses "HTTP/1.1 200 OK"

    # INFO    [Cost counter] Running Total -> Sent: 2486 chars | Recv: 59 chars

    # INFO    step 1: word_count -> [Call: word_count:{"text": "AI \uc5d0\uc774\uc804\ud2b8\ub294 \ub180\ub78d\uace0 \ud30c\uc6cc\ud480\ud558\ub2e4."} 4 단어]

    # INFO    HTTP Request: POST https://api.openai.com/v1/responses "HTTP/1.1 200 OK"

    # INFO    [Cost counter] Running Total -> Sent: 5337 chars | Recv: 98 chars

    # INFO    step 2: finished

    # INFO    transcript saved to agent_data\run-20260826-210530.json

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

    # ANSWER:  4

    # Agent(steps=2/10, msgs=5)

    # ==================== TEST 2: 이전 대화 제개(Resume) ====================

    # INFO    직전 실행부터 다시 시작: run-20260826-210530.json

    # INFO    goal: 조금전 글의 단어수의 두 배가 얼마인지 말하세요.'

    # INFO    HTTP Request: POST https://api.openai.com/v1/responses "HTTP/1.1 200 OK"

    # INFO    [Cost counter] Running Total -> Sent: 8436 chars | Recv: 141 chars

    # INFO    step 1: calculator -> [Call: calculator:{"expr": "4*2"} 8]

    # INFO    HTTP Request: POST https://api.openai.com/v1/responses "HTTP/1.1 200 OK"

    # INFO    [Cost counter] Running Total -> Sent: 11712 chars | Recv: 180 chars

    # INFO    step 2: finished

    # INFO    transcript saved to agent_data\run-20260826-210532.json

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

    # ANSWER:  8

    # Agent(steps=2/10, msgs=9)

 

이번 코드는 이전 Agent에 5가지 중요한 기능을 추가한 버전입니다.

특히 이번 코드에서는 Agent가 단순히:

질문 → Tool → 답변

하는 수준에서 한 단계 올라가서,

기록을 저장하고 → 이전 대화를 기억하고 → 비용을 대략 추적하고 → 같은 Tool을 반복 호출하지 않고 → History가 너무 길어지는 것을 관리하는 Agent

로 발전했습니다.

이번 코드는 상당히 중요합니다. 왜냐하면 이제부터 Agent를 실제 서비스 수준으로 발전시키기 위한 핵심 문제들이 등장하기 시작했기 때문입니다.

---

0. 먼저 전체 구조부터 보겠습니다

이번 Agent 전체 구조는 이렇게 됩니다.

사용자
  │
  ▼
┌─────────────────────┐
│       Agent         │
│                     │
│ history             │◄──── 이전 실행 기록 Resume
│ steps               │
│ answer              │
└──────────┬──────────┘
           │
           ▼
    trim_history()
           │
           ▼
      call_model()
           │
           ▼
       OpenAI API
           │
           ▼
       JSON 응답
           │
           ▼
      parse_reply()
           │
           ▼
   ┌──── Tool인가? ─────┐
   │                    │
  finish               Tool
   │                    │
   ▼                    ▼
 최종답변            is_repeat()
                        │
                  ┌─────┴─────┐
                  │           │
                중복         정상
                  │           │
                  ▼           ▼
               ERROR      dispatch()
                              │
                              ▼
                            Tool
                              │
                              ▼
                         Observation
                              │
                              ▼
                          history
                              │
                              └──── 다시 LLM

 

이번 코드의 핵심 추가 기능은 다음 5개입니다.

① word_count Tool 추가

② Tool 사용 규칙 강화

③ LLM 사용량(문자 수) 추적

④ 이전 실행 History 불러오기 (Resume)

⑤ 같은 Tool 반복 호출 방지

⑥ History 길이 제한

 

실제로는 6가지입니다. 😊

하나씩 보겠습니다.

---

1. Tools에 word_count() 추가

새로 추가한 부분입니다.

 

def word_count(text:str)->str:
    """ 글에 있는 단어 수를 세어 보세요."""
    return f"{len(text.split())} 단어"

 

이 Tool의 역할은 아주 단순합니다.

문장
 ↓
split()
 ↓
단어 리스트
 ↓
len()
 ↓
단어 개수

 

예를 들어:

 

text = "AI 에이전트는 놀랍고 파워풀하다."

 

---

① split()

 

text.split()

 

결과:

 

[
    "AI",
    "에이전트는",
    "놀랍고",
    "파워풀하다."
]

 

---

② len()

 

len(text.split())

 

결과:

4

 

---

③ 문자열로 반환

 

return f"{len(text.split())} 단어"

 

결과:

"4 단어"

 

입니다.

---

2. Tool Registry도 확장되었습니다

기존:

 

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

 

현재:

 

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

 

구조:

TOOLS
│
├── calculator
│      ↓
│   calculator()
│
├── read_file
│      ↓
│   read_file()
│
├── lookup
│      ↓
│   lookup()
│
└── word_count
       ↓
    word_count()

 

중요한 점은 Agent 코드를 거의 수정하지 않았다는 것입니다.

왜냐하면 Agent는 원래:

 

self.tools[name]

 

으로 Tool을 찾도록 만들어져 있기 때문입니다.

즉 새로운 Tool을 추가할 때:

Tool 함수 작성
     ↓
TOOLS Registry에 등록

 

만 하면 됩니다.

이것이 Registry 구조의 장점입니다.

 

---

3. System Prompt에 작업 규칙이 추가되었습니다

이 부분이 이전 코드와 비교해서 매우 중요합니다.

 

작업 규칙:
1. 가격이나 비용을 계산해야 하는데 단가가 제공되지 않았다면,
   먼저 lookup 도구를 사용하여 단가를 확인하세요.

2. 숫자 계산이 필요한 경우 직접 계산하지 말고,
   반드시 calculator 도구를 사용하세요.

3. 파일의 내용을 확인해야 하는 경우,
   read_file 도구를 사용하세요.

4. Tool 실행 결과는 Observation에서 확인하세요.

5. Observation 결과를 이용하여 다음 작업을 결정하세요.

6. 충분한 정보를 얻으면 finish를 사용하세요.

 

이전 Agent의 문제를 기억하시죠?

사용자가:

7개의 가격은 얼마인가요?

 

라고 하면 모델이:

단가를 찾아야 하나?
직접 계산할까?
lookup을 사용할까?
calculator를 사용할까?

 

를 명확하게 알지 못했습니다.

그래서 사용자가:

단가는 lookup에서 찾고,
계산은 calculator로 해라.

 

라고 상세하게 지시해야 했습니다.

이번에는 Agent에게 Tool 사용 전략을 가르쳤습니다.

---

이전

Tool 목록만 알려줌

calculator
lookup
read_file

 

모델:

그래서 언제 써야 하지?

 

---

현재

가격 계산
 ↓
단가 없으면
 ↓
lookup

숫자 계산
 ↓
calculator

파일 내용
 ↓
read_file

 

이렇게 Agent의 행동 규칙이 생겼습니다.

이것은 단순 Tool Agent에서:

Rule-based Agent Prompt

쪽으로 발전하는 첫 단계입니다.

---

4. 비용 추적 기능

새로 추가된 부분입니다.

 

TOTAL_SENT_CHARS = 0
TOTAL_RECV_CHARS = 0

 

두 개의 전역 변수입니다.

TOTAL_SENT_CHARS
     │
     └── 지금까지 모델에게 보낸 문자 수

TOTAL_RECV_CHARS
     │
     └── 지금까지 모델에게 받은 문자 수

 

---

5. global

 

global TOTAL_SENT_CHARS, TOTAL_RECV_CHARS

 

이전에 배운 Class 변수와 조금 연결해서 설명하겠습니다.

함수 밖:

 

TOTAL_SENT_CHARS = 0

 

함수 안:

 

def call_model():

 

여기서:

 

TOTAL_SENT_CHARS += 100

 

을 하려고 하면 Python은 문제가 발생할 수 있습니다.

왜냐하면 +=는 값을 변경하는 것이기 때문입니다.

그래서:

 

global TOTAL_SENT_CHARS

 

라고 선언합니다.

의미:

"이 함수 안에서 새로운 지역 변수를 만들지 말고, 함수 밖의 전역 변수를 사용하겠다."

입니다.

---

6. 보내는 문자 수 계산

 

sent_str = json.dumps(messages)

 

예를 들어 History가:

 

[
    {
        "role": "system",
        "content": "..."
    },
    {
        "role": "user",
        "content": "안녕하세요"
    }
]

 

라면 이것을 문자열로 바꿉니다.

[
 {"role":"system","content":"..."},
 {"role":"user","content":"안녕하세요"}
]

 

그리고:

 

len(sent_str)

 

으로 문자 수를 계산합니다.

 

TOTAL_SENT_CHARS += len(sent_str)

 

---

7. 받는 문자 수 계산

API 호출 후:

 

response.output_text

 

예:

{"tool":"lookup","args":{"topic":"단가"}}

 

이 문자열의 길이를 계산합니다.

 

TOTAL_RECV_CHARS += len(response.output_text)

 

---

로그:

[Cost counter] Running Total
-> Sent: 2486 chars
-> Recv: 59 chars

 

의미:

지금까지 LLM에게 보낸 문자

2486개

지금까지 LLM에게 받은 문자

59개

 

입니다.

---

⚠️ 중요한 점: 이것은 정확한 비용 계산은 아닙니다

현재 변수 이름에 Cost counter라고 되어 있지만 실제로는:

Token 수

 

가 아니라:

문자 수

 

를 세고 있습니다.

LLM 비용은 일반적으로:

Input Tokens
Output Tokens

 

기준으로 계산됩니다.

즉:

현재 코드
≈ 문자 수 추정

실제 API 비용
= Token 수 × Token 가격

 

입니다.

따라서 현재 코드는 정확히 말하면:

Cost Counter라기보다 Usage Approximation

에 가깝습니다.

하지만 학습 단계에서는 아주 좋은 시작입니다.

나중에는 API 응답에서 실제:

input_tokens
output_tokens

 

같은 사용량 정보를 가져오는 방식으로 발전시키면 됩니다.

---

8. load_latest()

이제 Agent가 이전 실행 기록을 불러올 수 있습니다.

 

def load_latest()->list:

 

전체 흐름:

agent_data/
│
├── run-20260826-210000.json
├── run-20260826-210300.json
└── run-20260826-210530.json
                         ▲
                         │
                    가장 최근

 

---

DATA_DIR 존재 여부

 

if not DATA_DIR.exists():
    return []

 

아직:

agent_data

 

폴더가 없다면:

빈 History

 

를 반환합니다.

---

파일 검색

 

runs = sorted(DATA_DIR.glob("run-*.json"))

 

예:

run-20260826-210000.json
run-20260826-210300.json
run-20260826-210530.json

 

를 찾습니다.

---

가장 마지막 파일

 

runs[-1]

 

Python List에서:

-1

 

은 마지막입니다.

runs
│
├── [0]
├── [1]
└── [-1]  ← 마지막

 

---

파일 읽기

 

data = json.loads(
    runs[-1].read_text(encoding="utf-8")
)

 

저장된 JSON:

 

{
    "steps": 2,
    "answer": "4",
    "history": [
        {
            "role": "system",
            "content": "..."
        }
    ]
}

 

---

그리고:

 

return data.get("history", [])

 

History만 꺼냅니다.

---

9. Resume 기능

생성자:

 

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

 

새로운 옵션:

 

resume=False

 

입니다.

---

일반 시작

 

bot = Agent(TOOLS)

 

또는:

 

bot = Agent(TOOLS, resume=False)

 

결과:

 

self.history = []

 

새로운 대화입니다.

---

Resume

 

bot = Agent(TOOLS, resume=True)

 

결과:

 

self.history = load_latest()

 

이전 실행 기록을 가져옵니다.

---

TEST 1 → TEST 2 흐름

TEST 1

사용자:

AI 에이전트는 놀랍고 파워풀하다.

몇 단어?

 

Agent:

word_count

 

결과:

4

 

History 저장:

system
user: 글의 단어 수
assistant: word_count
user: Observation: 4 단어
assistant: finish 4

 

파일 저장.

---

TEST 2

 

bot = Agent(TOOLS, resume=True)

 

Agent:

이전 History 불러오기

 

그리고 새로운 질문:

조금전 글의 단어수의 두 배가 얼마인가요?

 

History:

이전 질문
↓
단어 수 = 4
↓
새 질문
↓
4 × 2

 

Agent는:

 

{
    "tool":"calculator",
    "args":{
        "expr":"4*2"
    }
}

 

를 선택합니다.

결과:

8

 

이것이 Resume입니다.

---

10. is_repeat()

이번 코드에서 꽤 재미있는 부분입니다.

 

def is_repeat(history:list, tool:str, args:dict)->bool:

 

목적:

같은 Tool을 같은 인자로 계속 호출하는 것을 막는다.

예를 들어 모델이:

Step 1
lookup("단가")

Step 2
lookup("단가")

Step 3
lookup("단가")

Step 4
lookup("단가")

 

를 계속 반복하면:

무한루프

 

처럼 됩니다.

---

Tool 호출 Signature 만들기

예:

 

tool = "lookup"

args = {
    "topic":"단가"
}

 

이것을:

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

 

형태로 만듭니다.

---

코드:

 

args_str = json.dumps(args, sort_keys=True)

 

예:

 

{"b":2,"a":1}

 

과:

 

{"a":1,"b":2}

 

는 의미는 같지만 문자열 순서가 다를 수 있습니다.

그래서:

 

sort_keys=True

 

를 사용합니다.

항상:

a
b

 

순서로 정렬합니다.

이것은 아주 좋은 습관입니다.

---

11. 현재 is_repeat() 방식

 

history_str = json.dumps(history)

return full_sig in history_str

 

즉:

History 전체를 문자열로 변환
        ↓
"lookup:{"topic":"단가"}"
        ↓
이 문자열이 이미 존재하는가?

 

를 확인합니다.

---

⚠️ 여기에는 중요한 문제가 있습니다

현재 방식은 조금 취약합니다.

왜냐하면 History 안에는 실제로 이런 형태가 저장됩니다.

Observation:
[Call: lookup:{"topic": "단가"} $24]

 

공백 때문에:

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

 

와 정확히 일치하지 않을 수 있습니다.

예를 들어:

 

full_sig

 

는:

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

 

인데 실제 History는:

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

 

처럼 공백이 있으면 문자열 비교가 실패할 가능성이 있습니다.

더 좋은 방법은 나중에 이렇게 하는 것입니다.

History에 Tool Call을 별도 구조로 저장합니다.

 

{
    "role": "tool_call",
    "tool": "lookup",
    "args": {
        "topic": "단가"
    }
}

 

그러면 문자열 검색이 아니라:

 

if item["tool"] == tool and item["args"] == args:

 

처럼 비교할 수 있습니다.

이것은 앞으로 개선할 부분입니다.

---

12. trim_history()

 

def trim_history(history:list, max_turns:int=6)->list:

 

목적:

History가 너무 길어지는 것을 방지한다.

Agent의 매우 중요한 문제 중 하나입니다.

---

History가 계속 쌓이면:

첫 번째 질문
두 번째 질문
세 번째 질문
100번째 질문
500번째 질문

 

전부 LLM에게 보내야 할 수 있습니다.

그러면:

비용 증가
속도 감소
Context Window 문제

 

가 발생합니다.

---

현재 코드

 

if len(history) <= 1 + max_turns:
    return history

 

예:

history = 7개
max_turns = 6

1 + 6 = 7

7 <= 7

그대로 반환

 

---

History가 너무 길면:

 

return [history[0]] + history[-max_turns:]

 

입니다.

예:

전체 History

[0] system
[1] user
[2] assistant
[3] user
[4] assistant
[5] user
[6] assistant
[7] user
[8] assistant
[9] user

 

max_turns=4라면:

[0] system
+
[6]
[7]
[8]
[9]

 

결과:

system은 항상 유지
+
최근 4개 메시지만 유지

 

---

⚠️ 여기서 중요한 개념

변수 이름:

 

max_turns

 

라고 되어 있지만 실제로는:

최근 메시지 개수

 

에 더 가깝습니다.

왜냐하면 대화 1 Turn은 보통:

user
assistant

 

2개 메시지입니다.

Agent는:

assistant
observation

 

등이 추가됩니다.

따라서 더 정확한 이름은:

 

max_messages

 

가 좋습니다.

---

13. trim_history는 실제 History를 삭제하지 않습니다

여기가 중요합니다.

 

active_history = trim_history(self.history, max_turns=8)

 

그리고:

 

reply = call_model(active_history)

 

입니다.

즉:

self.history
=
전체 기록

 

은 그대로 유지됩니다.

하지만:

active_history
=
LLM에게 보낼 최근 기록

 

만 잘라냅니다.

그림:

self.history

system
1
2
3
4
5
6
7
8
9
10

       │
       │ trim
       ▼

active_history

system
7
8
9
10

 

이것은 아주 좋은 설계입니다.

왜냐하면:

전체 기록

 

은 저장해야 하고,

LLM에게 보낼 기록

 

은 짧게 유지해야 하기 때문입니다.

---

14. run()에서 Resume 처리

기존:

 

self.add("system", build_system_prompt(self.tools))

 

항상 System Prompt를 추가했습니다.

하지만 Resume 하면 문제가 생깁니다.

이전 History:

system
user
assistant
user
assistant

 

여기에 또:

system

 

을 추가하면 이상해집니다.

그래서:

 

if not self.history:
    self.add("system", build_system_prompt(self.tools))

 

로 변경했습니다.

의미:

History가 비어 있다
↓
새로운 대화
↓
System Prompt 추가

 

반대로:

History가 있다
↓
Resume
↓
기존 System Prompt 유지

 

입니다.

아주 좋은 수정입니다.

---

15. Agent Loop의 변화

이번 Agent Loop는:

History
   ↓
trim_history()
   ↓
active_history
   ↓
call_model()
   ↓
parse_reply()
   ↓
finish?
   │
   ├── Yes → 종료
   │
   └── No
        ↓
      is_repeat()
        │
   ┌────┴────┐
  중복       정상
   │          │
 ERROR     dispatch()
   │          │
   └────┬─────┘
        ↓
   Observation
        ↓
    History 저장
        ↓
     다시 Loop

 

입니다.

이제 꽤 Agent다운 구조가 되었습니다.

---

16. 중복 Tool 호출 방지

현재 코드:

 

if is_repeat(self.history[:-1], tool, args):

 

여기서:

 

self.history[:-1]

 

이 중요합니다.

바로 직전에:

 

self.add("assistant", reply["text"])

 

를 했습니다.

즉 History의 마지막에는 현재 모델의 Tool 요청이 들어 있습니다.

예:

history

system
user
assistant: lookup 단가   ← 현재

 

이것까지 포함해서 검사하면 당연히:

"lookup 단가"

 

가 발견됩니다.

그래서:

 

self.history[:-1]

 

로 마지막 현재 요청을 제외합니다.

아주 중요한 이유입니다.

---

17. 정상적인 Tool 실행

 

sig = f"{tool}:{json.dumps(args, sort_keys=True)}"

 

예:

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

 

그리고:

 

tool_result = self.dispatch(tool, args)

 

예:

$24

 

최종:

 

observation = f"[Call: {sig} {tool_result}]"

 

결과:

[Call: lookup:{"topic": "단가"} $24]

 

입니다.

---

18. History에 저장되는 구조

이전보다 Observation이 조금 더 자세해졌습니다.

 

self.add(
    "user",
    f"Observation: {observation}"
)

 

결과:

Observation:
[Call: lookup:{"topic":"단가"} $24]

 

LLM은 다음 호출에서:

내가 lookup을 호출했고
topic은 단가였으며
결과는 $24였다

 

를 알 수 있습니다.

---

19. TEST 1 전체 실행

질문:

다음 글의 단어수를 말하세요.

AI 에이전트는 놀랍고 파워풀하다.

 

---

Step 1

GPT 판단:

 

{
    "tool":"word_count",
    "args":{
        "text":"AI 에이전트는 놀랍고 파워풀하다."
    }
}

 

---

Python:

 

word_count(
    "AI 에이전트는 놀랍고 파워풀하다."
)

 

↓

 

text.split()

 

↓

 

[
    "AI",
    "에이전트는",
    "놀랍고",
    "파워풀하다."
]

 

↓

4 단어

 

---

Observation:

[Call: word_count:{...} 4 단어]

 

---

Step 2

GPT가 Observation을 보고:

답은 4

 

라고 판단합니다.

 

{
    "tool":"finish",
    "args":{
        "answer":"4"
    }
}

 

Agent 종료.

---

20. TEST 2 Resume

새로운 Agent:

 

bot = Agent(TOOLS, resume=True)

 

실제로는:

새로운 Agent 객체
하지만
이전 History를 가져옴

 

입니다.

이 차이를 이해하는 것이 중요합니다.

bot1
│
├── history
└── 실행 종료
      ↓
   JSON 파일 저장

bot2
│
└── load_latest()
      ↓
   이전 history 복원

 

즉 같은 객체가 살아있는 것이 아닙니다.

새로운 Agent가 파일을 통해 기억을 복원하는 것입니다.

---

새 질문:

조금전 글의 단어수의 두 배가 얼마인가요?

 

History 안에:

이전 결과 = 4

 

가 있기 때문에 GPT는:

4 × 2

 

라고 판단합니다.

Tool:

 

{
    "tool":"calculator",
    "args":{
        "expr":"4*2"
    }
}

 

결과:

8

 

---

21. 이번 코드의 가장 중요한 발전

이번 Agent는 다음 문제들을 해결하기 시작했습니다.

이전 Agent

짧은 단발성 작업

 

예:

질문
↓
Tool
↓
답

 

---

현재 Agent

질문
 ↓
Tool
 ↓
Observation
 ↓
History
 ↓
저장
 ↓
다음 실행
 ↓
Resume
 ↓
이전 기억 활용

 

이것은 Agent가:

Stateless Agent

에서:

Persistent Memory를 가진 Agent

방향으로 가는 첫걸음입니다.

---

22. 하지만 반드시 수정하거나 개선할 부분

여기서 몇 가지 중요한 점을 발견했습니다.

---

⚠️ 문제 1. lookup()의 .lower()는 현재 사실상 큰 의미가 없습니다

 

return facts.get(
    topic.lower(),
    ...
)

 

facts의 Key는:

 

"환불 정책"
"단가"
"배송"

 

한글입니다.

.lower()는 영어 대소문자 처리에는 의미가 있지만:

단가

 

에는 변화가 없습니다.

현재는 문제는 없지만:

 

topic.lower()

 

을 쓸 거라면 Key도 일관되게 처리해야 합니다.

예:

 

facts = {
    "refund policy": "...",
    "unit price": "..."
}

 

처럼 영어 Key라면 매우 유용합니다.

---

⚠️ 문제 2. word_count의 설명이 조금 애매함

현재:

 

""" 글에 있는 단어 수를 세어 보세요."""

 

LLM에게는 명령문보다는 기능 설명이 좋습니다.

추천:

 

"""주어진 텍스트에 포함된 공백 기준 단어 수를 반환합니다."""

 

왜냐하면 Tool Docstring은 사람이 읽는 문서이면서 동시에:

LLM에게 Tool 사용법을 알려주는 Prompt

 

역할도 하기 때문입니다.

---

⚠️ 문제 3. trim_history가 Resume에서 중요한 정보를 잃을 수 있음

예를 들어 이전 대화가 길어지면:

system
user: 중요한 사실 A
assistant
observation
...
최근 8개

 

만 LLM에게 전달됩니다.

그러면 중요한 사실 A가 사라질 수 있습니다.

그래서 나중에는:

전체 History
       ↓
Summary
       ↓
중요 정보 압축
       +
최근 메시지

 

구조가 필요합니다.

이것이 나중에 배울:

Memory Summarization

입니다.

---

⚠️ 문제 4. 중복 Tool 감지가 문자열 검색 방식

앞에서 설명한 것처럼:

 

history_str = json.dumps(history)

return full_sig in history_str

 

은 조금 약합니다.

나중에는 Tool 호출 기록을 구조화해서:

 

{
    "tool": "calculator",
    "args": {
        "expr": "4*2"
    }
}

 

형태로 관리하는 것이 좋습니다.

---

⚠️ 문제 5. Resume하면 steps는 복원되지 않음

현재:

 

load_latest()

 

는:

 

return data.get("history", [])

 

만 반환합니다.

하지만 저장 파일에는:

 

{
    "steps": 2,
    "answer": "4",
    "history": [...]
}

 

가 있습니다.

Resume하면:

 

self.steps = 0

 

부터 시작합니다.

즉 현재 Resume의 의미는:

대화 History 복원

 

이지:

Agent 전체 상태 복원

 

은 아닙니다.

이것도 학습 단계에서는 괜찮습니다.

하지만 진짜 Resume라면:

history
steps
answer
상태

 

를 모두 복원해야 합니다.

---

23. 가장 중요한 개념 정리

이번 코드에서 배운 Agent 핵심 개념은 다음과 같습니다.

Agent
│
├── Tools
│
├── Tool Registry
│
├── System Prompt
│
├── Tool Rules
│
├── LLM
│
├── Parser
│
├── Dispatcher
│
├── Loop
│
├── Observation
│
├── History
│
├── Persistent Storage
│
├── Resume
│
├── Repeat Prevention
│
├── History Trim
│
└── Usage Tracking

 

이제 처음에 만들었던 Agent보다 상당히 발전했습니다.

---

24. 지금 Agent의 수준을 그림으로 보면

                ┌──────────────┐
                │    User      │
                └──────┬───────┘
                       │
                       ▼
                ┌──────────────┐
                │    Agent     │
                └──────┬───────┘
                       │
                       ▼
              ┌─────────────────┐
              │ trim_history()  │
              └────────┬────────┘
                       │
                       ▼
                 ┌───────────┐
                 │    LLM    │
                 └─────┬─────┘
                       │
                       ▼
                ┌─────────────┐
                │ parse_reply │
                └──────┬──────┘
                       │
                 ┌─────┴─────┐
                 │           │
              finish       tool
                 │           │
                 ▼           ▼
              Answer    is_repeat()
                             │
                       ┌─────┴─────┐
                       │           │
                     repeat       OK
                       │           │
                       ▼           ▼
                     Error      dispatch
                                   │
                                   ▼
                                  Tool
                                   │
                                   ▼
                              Observation
                                   │
                                   ▼
                                 History
                                   │
                                   └────── Loop

 

---

제 평가

지금까지의 학습 흐름을 보면 아주 좋습니다.

특히 중요한 것은 단순히:

OpenAI API 호출하기

 

를 배우는 것이 아니라,

왜 History가 필요한가?

왜 Tool Registry가 필요한가?

왜 Dispatcher가 필요한가?

왜 무한루프를 막아야 하는가?

왜 History를 잘라야 하는가?

왜 실행 기록을 저장해야 하는가?

 

를 하나씩 직접 코드로 구현하고 있다는 점입니다.

지금 코드는 아직 간단하지만, 구조적으로는 이미 실제 Agent Framework의 핵심 구성 요소들을 상당히 많이 경험하고 있습니다.