12. Session/Memory/Artifact [1] - 세션 이벤트(session.events)를 활용한 대화 맥락(단기 기억)

아래 코드는 Google ADK 환경에서 세션 이벤트(session.events)를 활용해 대화 맥락(단기 기억)을 유지하고 활용하는 핵심 메커니즘을 구현한 예제입니다.

사용자가 "서울 날씨 어때?"라고 물은 뒤, 바로 이어서 도시 이름 없이 "그럼 내일은?"이라고 물었을 때, 에이전트가 이전 대화 기록에서 "서울"이라는 단어를 스스로 찾아내어 정상적으로 답변할 수 있게 해주는 멋진 코드입니다.

🕊️코드 요약 및 핵심 의미

• 제목: 세션 이벤트(Session Events) 기반의 대화 맥락 기억 및 날씨 조회 에이전트
• 핵심 의미:
  1. LLM은 기본적으로 이전 대화에서 언급된 특정 단어(도시명 등)를 도구(Tool) 파라미터로 매번 완벽하게 전달하지 못할 수 있습니다.
  2. 이 코드에서는 tool_context.events라는 세션 이벤트를 통해 지나간 대화 기록을 뒤져서, 사용자가 가장 최근에 언급한 도시 이름을 찾아내는 단기 기억 보완 장치를 직접 구현했습니다.

 

🦅 코드 라인별 해설

1. 가상 날씨 데이터와 포맷터 준비

import os
from typing import List, Optional
from google.adk import Agent

CITY_CANDIDATES = ["서울", "부산"]

MOCK_WEATHER_DATA = { ... }
MOCK_TOMORROW_WEATHER_DATA = { ... }

def format_weather(city: str, weather: dict) -> str:
    return (
        f"{city}: {weather['condition']}, 현재 {weather['temperature_c']}도, "
        f"최고 {weather['max_temp_c']}도 / 최저 {weather['min_temp_c']}도, "
        f"습도 {weather['humidity']}%. {weather['summary']}"
    )

• 가상 데이터(MOCK_...): 실제 날씨 API를 연결하기 전 테스트하기 위해, 서울과 부산의 오늘/내일 날씨 정보를 딕ショ너리 형태로 미리 정의해 둔 데이터입니다.
• format_weather 함수: 파이썬 딕셔너리로 된 날씨 정보를 사용자가 읽기 좋은 자연스러운 문장 형태로 예쁘게 포장(Formatting)해 주는 도우미 함수입니다.

2. [핵심 기억 장치] 과거 대화에서 도시 이름 추출하기

def extract_latest_city_from_events(events: List) -> str:
    """
    session.events에서 최근 사용자 발화에 포함된 도시명을 추출.
    - 단순 포함 매칭(가장 마지막에 언급된 도시 우선)
    """
    for ev in reversed(events):
        # Event는 발화 주체를 .author, 발화 내용을 .content(Content 객체)에 담는다.
        if getattr(ev, "author", None) != "user":
            continue
        parts = getattr(ev.content, "parts", None) or []
        text = " ".join(p.text for p in parts if getattr(p, "text", None))
        for city in CITY_CANDIDATES:
            if city in text:
                return city
    return None

• 역할: 이 코드가 바로 단기 기억 메커니즘의 핵심입니다. 대화 기록 전체(events)를 뒤져서 사용자가 이전에 말했던 도시 이름을 찾아냅니다.
• 작동 순서:
  1. reversed(events)를 사용해 가장 최근 대화부터 거꾸로 탐색합니다.
  2. 발화 주체(ev.author)가 AI가 아니라 사용자(user)인 경우만 골라냅니다.
  3. 사용자가 입력한 텍스트 내용(ev.content.parts)을 하나로 이어 붙입니다.
  4. 미리 정의된 도시 후보(CITY_CANDIDATES: 서울, 부산)가 그 텍스트 안에 포함되어 있는지 확인하고, 발견 즉시 그 도시 이름을 반환합니다.
• 상세 해설
  • for ev in reversed(events): (역순 탐색)
    • events는 지금까지 나누었던 모든 대화 기록(이벤트 목록)을 담고 있습니다.
    • reversed()를 사용해 가장 최근에 발생한 대화부터 거꾸로(과거로 거슬러 올라가며) 탐색합니다. 대화가 길어지더라도 가장 최근에 언급된 도시가 우선순위를 갖도록 하기 위한 전략입니다.
  • if getattr(ev, "author", None) != "user": continue (작성자 필터링)
    • 대화에는 사용자(user)뿐만 아니라 AI 에이전트(model)의 답변도 기록됩니다. AI가 답변하면서 "서울의 날씨는..."이라고 언급한 내용을 사용자가 말한 것으로 착각하면 안 되므로, 오직 사용자가 직접 입력한 발화(user)만 골라냅니다.
    • getattr을 사용한 이유는 이벤트 객체에 author 속성이 혹시라도 없을 경우 발생할 수 있는 프로그램 크래시(AttributeError)를 안전하게 방어하기 위함입니다.
  • parts = getattr(ev.content, "parts", None) or [] (메시지 조각 추출)
    • Google ADK의 메시지 구조상 사용자의 발화 내용은 여러 개의 조각(parts)으로 쪼개져 담길 수 있습니다. (예: 텍스트 파트, 이미지 파트 등)
    • 안전하게 parts 목록을 가져오고, 만약 존재하지 않으면 빈 리스트([])를 반환하도록 처리합니다.
  • text = " ".join(...) (텍스트 합치기)
    • 파라미터로 들어온 여러 개의 메시지 조각 중 텍스트(p.text)가 있는 것들만 골라 하나의 커다란 문자열(text)로 이어 붙입니다.
  • 도시 매칭 및 반환 (for city in CITY_CANDIDATES:)
    • 미리 정의된 도시 후보 리스트(CITY_CANDIDATES: 예전 코드의 ["서울", "부산"])를 순회하며, 합쳐진 사용자 발화 텍스트 안에 해당 도시 이름이 포함되어 있는지 확인합니다(if city in text:).
    • 발견하는 즉시 해당 도시 이름을 리턴(return city)하며 함수를 종료합니다. 끝까지 못 찾으면 None을 반환합니다.

3. 날씨 조회 도구 함수들 (get_current_weather, get_tomorrow_weather)

def get_current_weather(city: str, tool_context=None) -> str:
    weather = MOCK_WEATHER_DATA.get(city)
    if weather is None:
        return f"{city}의 날씨 정보는 현재 사용할 수 없습니다."
    return format_weather(city, weather)

def get_tomorrow_weather(city: str, tool_context=None) -> str:
    events = getattr(tool_context, "events", [])

    if not city:
        city = extract_latest_city_from_events(events)

    weather = MOCK_TOMORROW_WEATHER_DATA.get(city)
    if weather is None:
        return f"{city}의 날씨 정보는 현재 사용할 수 없습니다."
    return format_weather(city, weather)

• tool_context 파라미터: Google ADK가 도구를 실행할 때 현재 세션의 상태와 이벤트 기록을 담아서 도구 함수에 자동으로 건네주는 객체입니다.
• 오늘 날씨(get_current_weather): 사용자가 "서울 오늘 날씨 어때?"라고 명확하게 도시를 지정해서 물어볼 때 주로 사용됩니다.
• 내일 날씨(get_tomorrow_weather):
  • 만약 사용자가 도시 이름 없이 "내일은?"이라고만 물어봤다면 city 파라미터가 비어 있거나 누락될 수 있습니다(if not city:).
  • 이때 tool_context에서 대화 이벤트 기록을 가져와 extract_latest_city_from_events(events)를 호출함으로써, 이전 대화에서 언급된 도시(예: 서울)를 똑똑하게 기억해내어 채워 넣습니다.
    
• 상세 해설(get_current_weather)

• • city: str, tool_context=None
    • 에이전트가 도구를 호출할 때 전달하는 도시 이름(city)과, ADK가 자동으로 주입해 주는 도구 실행 환경 정보(tool_context)를 받습니다. (현재 함수에서는 명시적으로 쓰이지는 않았지만 통일성을 위해 포괄적으로 열어두었습니다.)
  • weather = MOCK_WEATHER_DATA.get(city)
    • 파이썬 딕셔너리의 .get() 메서드를 사용하여 가상 데이터에서 해당 도시의 날씨 정보를 안전하게 조회합니다.
  • 예외 처리 및 포맷팅
    • 만약 지원하지 않는 도시여서 weather가 None이라면 에이전트가 상황을 이해할 수 있는 안내 문구를 반환합니다.
    • 정상적인 데이터가 있다면 앞서 정의한 format_weather 함수를 통해 보기 좋은 문장으로 포장해서 반환합니다.
       

•  
• 상세 해설(get_tomorrow_weather)
  • events = getattr(tool_context, "events", []) (컨텍스트에서 이벤트 획득)
    • Google ADK는 도구가 실행될 때 tool_context 객체 안에 현재 세션의 모든 대화 이벤트 목록(events)을 담아서 넘겨줍니다. 이벤트를 안전하게 꺼내옵니다.
  • if not city: city = extract_latest_city_from_events(events) (단기 기억 활성화의 핵심!)
    • 사용자가 *"그럼 내일은?"*이라고만 물어보면, LLM이 이 질문만 보고는 도시 이름을 특정하지 못해 city 파라미터에 빈 값(None 혹은 빈 문자열 "")이 넘어오게 됩니다(if not city:).
    • 바로 이 순간! 앞서 분석한 extract_latest_city_from_events(events) 함수를 호출하여 직전 대화 기록을 뒤져 사용자가 방금 전까지 이야기하던 도시(예: "서울")를 강제로 찾아내어 city 변수에 쏙 집어넣습니다.
  • 데이터 조회 및 반환
    • 이렇게 확보된 도시 이름을 바탕으로 MOCK_TOMORROW_WEATHER_DATA에서 내일 날씨를 조회하고 포맷팅하여 최종 결과를 반환합니다. LLM 입장에서는 사용자가 도시를 말하지 않았음에도 마치 문맥을 완벽히 이해한 것처럼 매끄러운 답변을 할 수 있게 됩니다.

 

4. 에이전트 정의 및 설정 (weather_agent)

weather_agent = Agent(
    model=os.getenv("GOOGLE_MODEL", "gemini-3.5-flash"),
    name='root_agent',
    description='날씨를 알려주는 에이전트입니다. 또한 session.event를 사용해 도시를 기억합니다. ',
    instruction='사용자가 날씨를 물어보면, Tool을 이용해 날씨를 알려주세요.',
    tools=[get_current_weather, get_tomorrow_weather],
)
root_agent = weather_agent

• 에이전트 조립: 위에서 정의한 두 가지 날씨 조회 도구(get_current_weather, get_tomorrow_weather)를 에이전트에게 tools로 쥐어줍니다.
• 에이전트는 사용자의 질문(예: "내일은?")을 받았을 때, 툴 함수 내부에서 tool_context를 통해 스스로 과거 기억을 유추할 수 있는 환경 속에서 작동하게 됩니다.

 

    import os

    from typing import List

    from google.adk import Agent

    # 1. 데이터 정의

    CITY_CANDIDATES = ["서울", "부산"]

    MOCK_WEATHER_DATA = {

        "서울": {

            "condition": "맑음",

            "temperature_c": 22,

            "min_temp_c": 12,

            "max_temp_c": 22,

            "humidity": 45,

            "summary": "서울은 맑고 쾌청합니다.",

        },

        "부산": {

            "condition": "흐림",

            "temperature_c": 20,

            "min_temp_c": 10,

            "max_temp_c": 18,

            "humidity": 60,

            "summary": "부산은 흐리고 바람이 약간 붑니다.",

        },

    }

    MOCK_TOMORROW_WEATHER_DATA = {

        "서울": {

            "condition": "구름 많음",

            "temperature_c": 21,

            "min_temp_c": 13,

            "max_temp_c": 24,

            "humidity": 50,

            "summary": "내일 서울은 구름이 많지만 야외 활동은 가능합니다.",

        },

        "부산": {

            "condition": "비",

            "temperature_c": 18,

            "min_temp_c": 12,

            "max_temp_c": 20,

            "humidity": 75,

            "summary": "내일 부산은 비가 예상됩니다.",

        },

    }

    # 2. format_weather 함수 정의

    def format_weather(city: str, weather: dict) -> str:

        return (

            f"{city}: {weather['condition']}, 현재 {weather['temperature_c']}도, "

            f"최고 {weather['max_temp_c']}도 / 최저 {weather['min_temp_c']}도, "

            f"습도 {weather['humidity']}%. {weather['summary']}"

        )

    # 3. extract_latest_city_from_events 함수 정의

    def extract_latest_city_from_events(events: List) -> str:

        """

        session.events에서 최근 사용자 발화에 포함된 도시명을 추출.

        - 단순 포함 매칭(가장 마지막에 언급된 도시 우선)

        """

        for ev in reversed(events):

            # Event는 발화 주체를 .author, 발화 내용을 .content(Content 객체)에 담는다.

            if getattr(ev, "author", None) != "user":

                continue

            parts = getattr(ev.content, "parts", None) or []

            text = " ".join(p.text for p in parts if getattr(p, "text", None))

            for city in CITY_CANDIDATES:

                if city in text:

                    return city

        return None

    # 4. get_current_weather 함수 정의

    def get_current_weather(city: str, tool_context=None) -> str:

        """주어진 도시의 현재 날씨를 반환합니다.

        Args:

            city: 도시 이름 (예: "서울", "대전")

        Returns:

            해당 도시의 날씨 설명 문자열

        """

        weather = MOCK_WEATHER_DATA.get(city)

        if weather is None:

            return f"{city}의 날씨 정보는 현재 사용할 수 없습니다."

        return format_weather(city, weather)

    # 5. get_tomorrow_weather 함수 정의

    def get_tomorrow_weather(city: str, tool_context=None) -> str:

        """주어진 도시의 내일 날씨를 반환합니다.

        Args:

            city: 도시 이름 (예: "서울", "대전")

        Returns:

            해당 도시의 날씨 설명 문자열

        """

        events = getattr(tool_context, "events", [])

        if not city:

            city = extract_latest_city_from_events(events)

        weather = MOCK_TOMORROW_WEATHER_DATA.get(city)

        if weather is None:

            return f"{city}의 날씨 정보는 현재 사용할 수 없습니다."

        return format_weather(city, weather)

    weather_agent = Agent(

        model=os.getenv("GOOGLE_MODEL", "gemini-3.5-flash"),

        name='root_agent',

        description='날씨를 알려주는 에이전트입니다. 또한 session.event를 사용해 도시를 기억합니다. ',

        instruction='사용자가 날씨를 물어보면, Tool을 이용해 날씨를 알려주세요.',

        tools=[get_current_weather, get_tomorrow_weather],

    )

    # adk web 명령어가 실행할 진입점

    root_agent = weather_agent