본문 바로가기

명사 美 비격식 (무리 중에서) 아주 뛰어난[눈에 띄는] 사람[것]

Personal/SK 네트웍스 AI 캠프

SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day48_AI Agent 구현 실습 1

Agent는 기억을 하지 않는다. Memory 시스템(Checkpointer, DB 등)이 Agent에게 기억을 제공한다.

사용자
   │
   ▼
API (FastAPI)

   │
thread_id = "user123"

   ▼
Agent.invoke()

   │
   ▼
Checkpointer
(InMemorySaver /
 SQLite /
 Postgres)

   │
State 저장

   ▼
LLM

 

 

 

Stateless

이전 요청을 아무것도 기억하지 않는 상태

 

History = Prompt에 붙여주는 것

messages = [
    HumanMessage("안녕"),
    AIMessage("안녕하세요."),
    HumanMessage("내 이름은 철수야"),
    AIMessage("알겠습니다."),
    HumanMessage("내 이름 뭐야?")
]

llm.invoke(messages)

 

 

 

 

 

 

 

단기기억

세션 동안만 기억한다.

LangGraph에서 가장 많이 사용한다.RAM에 있기 때문에 서버가 종료시 모두 삭제되며 서버가 여러개일경우 공유되지않는다. 수천명 접소시 RAM 사용량이 증가해 운영에서는 거의 사용하지않고 Redis, SQLite, Postgres 를 사용한다.
thread_id 누구인지 기억한다.
State History보다 큰 개념
Checkpointer State를 저장하는 인터페이스

InMemorySaver  Agent Memory History를 자동으로 관리하는 시스템

from typing import TypedDict
from langgraph.graph import StateGraph, START, END
from langgraph.checkpoint.memory import InMemorySaver
from langchain_openai import ChatOpenAI
from langchain_core.messages import (
    HumanMessage,
    AIMessage,
    SystemMessage
)

##################################################
# State
##################################################

class State(TypedDict):
    messages: list
    summary: str

##################################################
# LLM
##################################################

llm = ChatOpenAI(
    model="gpt-4.1-mini",
    streaming=True
)

##################################################
# Chatbot Node
##################################################

def chatbot(state: State):

    prompt = []

    # 이전 요약이 있으면 먼저 전달
    if state["summary"]:
        prompt.append(
            SystemMessage(
                content=f"""
이전 대화 요약

{state["summary"]}
"""
            )
        )

    # 최근 대화 추가
    prompt.extend(state["messages"])

    response = llm.invoke(prompt)

    return {
        "messages": state["messages"] + [response]
    }

##################################################
# Summary Node
##################################################

def summarize(state: State):

    # 최근 대화가 적으면 요약 안함
    if len(state["messages"]) < 8:
        return {
            "summary": state["summary"]
        }

    prompt = f"""
다음 대화를 5줄 이내로 요약해라.

{state["messages"]}

중요한 사용자 정보와 대화만 남겨라.
"""

    summary = llm.invoke(prompt)

    # 최근 4개만 남기고 이전 대화는 버림
    recent_messages = state["messages"][-4:]

    return {
        "summary": summary.content,
        "messages": recent_messages
    }

##################################################
# Graph
##################################################

builder = StateGraph(State)

builder.add_node("chatbot", chatbot)
builder.add_node("summary", summarize)

builder.add_edge(START, "chatbot")
builder.add_edge("chatbot", "summary")
builder.add_edge("summary", END)

##################################################
# Memory
##################################################

memory = InMemorySaver()

graph = builder.compile(
    checkpointer=memory
)

##################################################
# Config
##################################################

config = {
    "configurable": {
        "thread_id": "user123"
    }
}

##################################################
# 첫 번째 대화
##################################################

for event in graph.stream(
    {
        "messages": [
            HumanMessage("내 이름은 철수야.")
        ],
        "summary": ""
    },
    config=config,
    stream_mode="values"
):

    if event["messages"]:
        print(event["messages"][-1].content)

##################################################
# 두 번째 대화
##################################################

for event in graph.stream(
    {
        "messages": [
            HumanMessage("나는 커피를 좋아해.")
        ]
    },
    config=config,
    stream_mode="values"
):

    if event["messages"]:
        print(event["messages"][-1].content)

##################################################
# 세 번째 대화
##################################################

for event in graph.stream(
    {
        "messages": [
            HumanMessage("내 이름과 좋아하는 음료가 뭐야?")
        ]
    },
    config=config,
    stream_mode="values"
):

    if event["messages"]:
        print(event["messages"][-1].content)

 

 

 

장기기억

실제로는 PostgreSQL, MongoDB, VectorDB를 사용한다. 

 

from langchain_openai import ChatOpenAI

llm = ChatOpenAI(model="gpt-4.1-mini")

#############################################
# 장기기억 DB
#############################################

memory_db = {}

#############################################
# 저장
#############################################

def save_memory(user_id, fact):

    if user_id not in memory_db:
        memory_db[user_id] = []

    memory_db[user_id].append(fact)

#############################################
# 조회
#############################################

def load_memory(user_id):

    return memory_db.get(user_id, [])

#############################################
# 첫 번째 날
#############################################

user_id = "kim"

save_memory(
    user_id,
    "사용자는 커피를 좋아한다."
)

#############################################
# 일주일 후
#############################################

memory = load_memory(user_id)

prompt = f"""

사용자 정보

{memory}

사용자 질문

카페 추천해줘

"""

response = llm.invoke(prompt)

print(response.content)

 

 

 

 

 

단기기억과 장기기억

User

↓

FastAPI

↓

Agent

├── Short Memory
│      │
│      └── Checkpointer
│             │
│             └── InMemory/Postgres
│
└── Long Memory
       │
       └── PostgreSQL
           Redis
           MongoDB
           VectorDB
구분 단기기억 (Short-Term) 장기기억 (Long-Term)
저장 내용 현재 대화 상태(State, messages 등) 사용자 프로필, 선호도, 과거 사실 등
저장 위치 Checkpointer (InMemory, SQLite, Postgres 등) Postgres, MongoDB, Vector DB 등
유지 기간 같은 thread/session 동안 세션 종료 후에도 유지
조회 방식 thread_id로 현재 State 복원 검색(Query, Embedding, SQL 등)
대표 구현 InMemorySaver, PostgresSaver RAG + Vector DB, 사용자 프로필 DB

 

 

 

 

DuckDuckGoSearchResults
LangChain에서 DuckDuckGo 검색 엔진을 이용해 웹 검색 결과를 가져오는 Tool

from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchResults
from langchain.agents import initialize_agent

# 1. LLM 생성
llm = ChatOpenAI(
    model="gpt-4.1-mini",
    temperature=0
)

# 2. 검색 Tool 생성
search_tool = DuckDuckGoSearchResults(
    max_results=3
)

# 3. Agent 생성
agent = initialize_agent(
    tools=[search_tool],
    llm=llm,
    verbose=True
)

# 4. 질문
result = agent.invoke(
    "2026년 SQLD 시험 일정 알려줘"
)

print(result)