5장 — 도구 사용(함수 호출)

패턴 개요

지금까지의 네 패턴(체이닝, 라우팅, 병렬화, 리플렉션)은 주로 언어 모델 간 상호작용을 관리하고, 에이전트 내부 워크플로에서 정보 흐름을 운영하는 것이었습니다. 하지만 에이전트가 진정으로 유용하려면, 특히 현실 세계나 외부 시스템과 상호작용하려면 도구를 사용할 수 있어야 합니다.

도구 사용 패턴은 흔히 함수 호출(Function Calling) 이라는 메커니즘으로 구현되며, 이를 통해 에이전트는 외부 API, 데이터베이스, 서비스와 상호작용하고 나아가 코드를 실행할 수도 있습니다. LLM은 에이전트의 중추로서, 이 패턴을 이용해 사용자 요청이나 현재 작업 상태를 바탕으로 특정 외부 함수를 언제, 어떻게 사용할지 결정할 수 있습니다.

여섯 단계

단계	하는 일
1. 도구 정의	외부 함수나 기능을 정의하고, LLM이 이해할 수 있도록 설명한다. 설명에는 함수의 용도, 이름, 받는 매개변수, 각 매개변수의 타입과 설명이 포함된다
2. LLM 판단	LLM은 사용자 요청과 사용 가능한 도구 정의를 함께 전달받는다. 요청을 완수하는 데 하나 이상의 도구 호출이 필요한지 판단한다
3. 함수 호출 생성	도구 사용이 필요하다고 판단하면, 호출할 도구의 이름과 요청에서 추출한 인자를 지정하는 구조화된 출력(보통 JSON 객체) 을 생성한다
4. 도구 실행	에이전틱 프레임워크 또는 오케스트레이션 계층이 이 구조화된 출력을 받아 요청된 도구를 식별하고, 전달된 인자로 실제 외부 함수를 실행한다
5. 관찰 / 결과	도구 실행에서 나온 출력 또는 결과가 에이전트에게 반환된다
6. LLM 처리 (선택적이지만 일반적)	LLM은 도구의 출력을 컨텍스트로 받아, 사용자에게 제공할 응답을 작성하거나 워크플로의 다음 단계를 결정한다. 다음 단계는 다른 도구 호출일 수도, 리플렉션 수행이나 최종 답변 도출일 수도 있다

3단계가 이 패턴의 핵심입니다. LLM은 함수를 직접 실행하지 않습니다. “이 함수를 이 인자로 불러 달라”는 구조화된 요청을 생성할 뿐이고, 실제 실행은 오케스트레이션 계층의 몫입니다.

이 패턴이 근본적으로 중요한 이유는 LLM 학습 데이터의 한계를 넘어설 수 있기 때문입니다. 이를 통해 LLM은 최신 정보에 접근하고, 내부적으로 수행할 수 없는 계산을 처리하며, 사용자 고유 데이터와 상호작용하고, 현실 세계에서 동작을 실행할 수 있습니다. 함수 호출은 LLM의 추론 능력과 광범위한 외부 기능 사이의 간극을 메우는 기술적 메커니즘입니다.

함수 호출 vs 도구 호출

“함수 호출”은 사전 정의된 특정 코드 함수를 실행한다는 뜻을 정확히 짚어 주는 표현이지만, 책은 보다 포괄적인 개념인 “도구 호출(tool calling)” 도 함께 생각해 볼 필요가 있다고 짚습니다.

이 포괄적인 용어는 에이전트의 역량이 단순한 함수 실행을 훨씬 넘어설 수 있음을 반영합니다. “도구”는 전통적인 함수일 수도 있지만, 복잡한 API 엔드포인트일 수도 있고, 데이터베이스에 보내는 요청일 수도 있으며, 또 다른 전문 에이전트에게 보내는 지시일 수도 있습니다.

이런 관점을 취하면, 예를 들어 주 에이전트가 복잡한 데이터 분석 작업을 전담 “분석 에이전트”에게 위임하거나 API를 통해 외부 지식 베이스에 질의하는 것처럼 보다 정교한 시스템을 구상할 수 있습니다. “도구 호출”이라는 관점에서 바라보면, 에이전트가 다양한 디지털 리소스와 지능형 개체로 구성된 생태계 전반에서 이들을 연결하고 조율하는 역할까지 수행할 수 있음을 더 잘 설명할 수 있습니다.

실용적 활용 분야와 사용 사례

도구 사용 패턴은 에이전트가 텍스트 생성을 넘어 실제 동작을 수행하거나 특정한 동적 정보를 가져와야 하는 사실상 모든 시나리오에 적용할 수 있습니다.

분야	사용 사례	도구	동작 흐름
외부 소스에서 정보 검색	날씨 에이전트	위치를 입력받아 현재 기상 정보를 반환하는 날씨 API	“런던 날씨가 어때?” → LLM이 "London" 을 인자로 도구 호출 → 반환 데이터를 이해하기 쉬운 응답으로 구성
데이터베이스 및 API와 상호작용	전자상거래 에이전트	제품 재고 확인, 주문 상태 조회, 결제 처리 API	“제품 X 재고 있어?” → 재고 API 호출 → 재고 수량을 사용자에게 안내
계산 및 데이터 분석 수행	금융 에이전트	계산기 함수, 주식 시장 데이터 API, 스프레드시트 도구	“AAPL 현재 가격이 얼마야? 150달러에 100주 매수했다면 잠재 수익은?” → 주식 API 호출로 현재가를 받고, 이어서 계산기 도구를 호출해 결과를 얻은 뒤 응답 구성
커뮤니케이션 전송	개인 비서 에이전트	이메일 전송 API	“내일 회의 건으로 존에게 이메일 보내 줘” → 요청에서 수신자·제목·본문을 추출하여 이메일 도구 호출
코드 실행	코딩 도우미 에이전트	코드 인터프리터	파이썬 코드를 제시하며 “이 코드가 뭘 하는 거야?” → 인터프리터 도구로 코드를 실행하고 출력을 분석
다른 시스템 또는 기기 제어	스마트홈 에이전트	스마트 조명 제어 API	“거실 불 꺼 줘” → 명령과 대상 기기 정보를 담아 스마트홈 도구 호출

금융 에이전트 사례는 도구를 연달아 두 번 호출하는 예입니다. 한 도구의 결과가 다음 도구의 입력이 되는 구조로, 6단계의 “다음 단계는 다른 도구 호출일 수도 있다”에 해당합니다.

‘도구 사용’은 언어 모델을 단순한 텍스트 생성기에서 디지털 세계와 물리 세계를 감지하고 추론하며 행동하는 에이전트로 변모시키는 요소입니다.

                                      ┌──────── 프록시 도구 ────────┐
                                      │  메모리    데이터베이스  스토리지 │
  사용자 ──► 프롬프트 ──► 에이전트 ⇄ 다단계  웹브라우저  웹검색   도구로   │
                              ▲       도구 호출                  사용되는 │
                              │              채팅      지도    에이전트   │
                              └──────────────  더 많은 도구들 ───────────┘

_{그림 5.1 — 도구를 사용하는 에이전트의 몇 가지 예}

실습 코드 예제 (LangChain)

LangChain에서 도구 사용을 구현하는 과정은 두 단계입니다. 먼저 하나 이상의 도구를 정의하는데, 보통 기존 파이썬 함수나 다른 실행 가능한 컴포넌트를 래핑하는 방식을 씁니다. 그 다음 이 도구를 언어 모델에 제공(바인딩) 하여, 사용자 질의를 이행하려면 외부 함수 호출이 필요하다고 판단할 때 모델이 구조화된 도구 사용 요청을 생성할 수 있게 합니다.

import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
from langchain_google_genai import ChatGoogleGenerativeAI
import logging
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool as langchain_tool
from langchain.agents import create_tool_calling_agent, AgentExecutor

# 사용자로부터 API 키를 안전하게 입력받아 환경 변수로 설정
os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter your Google API key: ")
os.environ["OPENAI_API_KEY"] = getpass.getpass("Enter your OpenAI API key: ")

try:
    # 함수/도구 호출 기능을 지원하는 모델이 필요하다.
    llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash", temperature=0)
    print(f"Language model initialized: {llm.model}")
except Exception as e:
    print(f"Error initializing language model: {e}")
    llm = None

# --- 도구 정의 ---
@langchain_tool
def search_information(query: str) -> str:
    """
    주어진 주제에 대한 사실 정보를 제공한다. 'capital of France'나
    'weather in London?'과 같은 질문에 대한 답변을 찾을 때
    이 도구를 사용한다.
    """
    print(f"\n--- Tool Called: search_information with query: '{query}' ---")
    # 미리 정의된 결과 딕셔너리를 사용해 검색 도구를 시뮬레이션한다.
    simulated_results = {
        "weather in london": "The weather in London is currently cloudy with a temperature of 15°C.",
        "capital of france": "The capital of France is Paris.",
        "population of earth": "The estimated population of Earth is around 8 billion people.",
        "tallest mountain": "Mount Everest is the tallest mountain above sea level.",
        "default": f"Simulated search result for '{query}': No specific information found, but the topic seems interesting."
    }
    result = simulated_results.get(query.lower(), simulated_results["default"])
    print(f"--- TOOL RESULT: {result} ---")
    return result

tools = [search_information]

# --- 도구 호출 에이전트 생성 ---
if llm:
    # 이 프롬프트 템플릿에는 에이전트의 내부 처리 과정을 위한 `agent_scratchpad` 플레이스홀더가 필요하다.
    agent_prompt = ChatPromptTemplate.from_messages([
        ("system", "You are a helpful assistant."),
        ("human", "{input}"),
        ("placeholder", "{agent_scratchpad}"),
    ])

    # LLM, 도구, 프롬프트를 결합하여 에이전트를 생성한다.
    agent = create_tool_calling_agent(llm, tools, agent_prompt)

    # AgentExecutor는 에이전트를 호출하고 선택된 도구를 실행하는 런타임이다.
    # 도구는 이미 에이전트에 바인딩되어 있으므로, 여기서 'tools' 인자는 필수가 아니다.
    agent_executor = AgentExecutor(agent=agent, verbose=True, tools=tools)

    async def run_agent_with_tool(query: str):
        """질의를 사용해 에이전트 실행기를 호출하고 최종 응답을 출력한다."""
        print(f"\n--- Running Agent with Query: '{query}' ---")
        try:
            response = await agent_executor.ainvoke({"input": query})
            print("\n--- Final Agent Response ---")
            print(response["output"])
        except Exception as e:
            print(f"\nAn error occurred during agent execution: {e}")

    async def main():
        """모든 에이전트 질의를 동시에 실행한다."""
        tasks = [
            run_agent_with_tool("What is the capital of France?"),
            run_agent_with_tool("What's the weather like in London?"),
            run_agent_with_tool("Tell me something about dogs.")  # 기본 도구 응답이 실행되어야 한다
        ]
        await asyncio.gather(*tasks)

    nest_asyncio.apply()
    asyncio.run(main())

읽는 포인트

• @tool 데코레이터가 도구 정의를 간소화합니다. 함수의 독스트링이 곧 LLM에게 주는 도구 설명이 되고, 타입 힌트(query: str)가 매개변수 명세가 됩니다. 1단계 “도구 정의”가 파이썬 함수 한 개로 끝납니다.
• agent_scratchpad 플레이스홀더가 반드시 필요합니다. 에이전트가 “도구를 호출했고 이런 결과를 받았다”는 중간 처리 과정을 여기에 쌓아 가며 다음 판단을 내립니다.
• AgentExecutor 가 4단계(도구 실행)를 담당하는 런타임입니다. LLM이 만든 구조화된 요청을 받아 실제 파이썬 함수를 부르고 결과를 되돌려주는 루프를 돌립니다.
• 세 질의를 asyncio.gather 로 동시에 실행하는데, 두 개는 사전 정의 응답을, 하나(“dogs”)는 기본 응답을 타도록 설계되어 있습니다.

실습 코드 예제 (CrewAI)

CrewAI는 에이전트(Agent)·태스크(Task)·크루(Crew) 라는 세 개념으로 협업 워크플로를 구성합니다. 시뮬레이션된 주가를 조회하는 금융 분석 시나리오입니다.

pip install crewai langchain-openai

import os
from crewai import Agent, Task, Crew
from crewai.tools import tool
import logging

# --- 모범 사례: 로깅 설정 ---
# 기본적인 로깅을 설정해 두면 크루의 실행을 디버깅하고 추적하는 데 유용하다.
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# --- API 키 설정 ---
# 프로덕션 환경에서는 런타임에 로드하는 환경 변수나 시크릿 매니저 등
# 보다 안전한 키 관리 방법을 사용하는 것이 권장된다.
#
# 선택한 LLM 프로바이더의 환경 변수를 설정한다(예: OPENAI_API_KEY)
# os.environ["OPENAI_API_KEY"] = "YOUR_API_KEY"
# os.environ["OPENAI_MODEL_NAME"] = "gpt-4o"

# --- 1. 리팩터링된 도구: 깔끔한 데이터 반환 ---
# 이 도구는 이제 원시 데이터(float)를 반환하거나 표준 파이썬 오류를 발생시킨다.
# 이렇게 하면 재사용성이 높아지고, 에이전트가 결과를 적절히 처리하도록 강제할 수 있다.
@tool("Stock Price Lookup Tool")
def get_stock_price(ticker: str) -> float:
    """
    주어진 종목 심벌에 대한 최신 시뮬레이션 주가를 조회한다.
    가격을 float로 반환한다. 종목을 찾을 수 없으면 ValueError를 발생시킨다.
    """
    logging.info(f"Tool Call: get_stock_price for ticker '{ticker}'")
    simulated_prices = {
        "AAPL": 178.15,
        "GOOGL": 1750.30,
        "MSFT": 425.50,
    }
    price = simulated_prices.get(ticker.upper())
    if price is not None:
        return price
    else:
        # 문자열을 반환하는 것보다 구체적인 오류를 발생시키는 편이 낫다.
        # 에이전트는 예외를 처리할 수 있으며, 다음 행동을 스스로 결정할 수 있다.
        raise ValueError(f"Simulated price for ticker '{ticker.upper()}' not found.")

# --- 2. 에이전트 정의 ---
# 에이전트 정의 자체는 동일하지만, 이제 개선된 도구를 활용하게 된다.
financial_analyst_agent = Agent(
    role='Senior Financial Analyst',
    goal='Analyze stock data using provided tools and report key prices.',
    backstory="You are an experienced financial analyst adept at using data sources to find stock information. You provide clear, direct answers.",
    verbose=True,
    tools=[get_stock_price],
    # 위임 기능은 유용할 수 있으나, 이 간단한 작업에서는 필요하지 않다.
    allow_delegation=False,
)

# --- 3. 개선된 태스크: 더 명확한 지시와 오류 처리 ---
# 태스크 설명을 더 구체적으로 작성하여, 데이터 조회 성공과 잠재적 오류
# 양쪽 경우에 에이전트가 어떻게 대응해야 하는지 안내한다.
analyze_aapl_task = Task(
    description=(
        "What is the current simulated stock price for Apple (ticker: AAPL)? "
        "Use the 'Stock Price Lookup Tool' to find it. "
        "If the ticker is not found, you must report that you were unable to retrieve the price."
    ),
    expected_output=(
        "A single, clear sentence stating the simulated stock price for AAPL. "
        "For example: 'The simulated stock price for AAPL is $178.15.' "
        "If the price cannot be found, state that clearly."
    ),
    agent=financial_analyst_agent,
)

# --- 4. 크루 구성 ---
# 크루는 에이전트와 태스크가 함께 작동하는 방식을 오케스트레이션한다.
financial_crew = Crew(
    agents=[financial_analyst_agent],
    tasks=[analyze_aapl_task],
    verbose=True  # 프로덕션에서는 False로 설정하면 로그 출력이 줄어든다
)

# --- 5. 메인 실행 블록에서 크루 실행 ---
# __name__ == "__main__": 블록을 사용하는 것은 표준적인 파이썬 모범 사례이다.
def main():
    """크루를 실행하는 메인 함수."""
    # 런타임 오류를 방지하기 위해 시작 전에 API 키가 설정되어 있는지 확인한다.
    if not os.environ.get("OPENAI_API_KEY"):
        print("ERROR: The OPENAI_API_KEY environment variable is not set.")
        print("Please set it before running the script.")
        return

    print("\n## Starting the Financial Crew...")
    print("---------------------------------")

    # kickoff 메서드가 실행을 시작한다.
    result = financial_crew.kickoff()

    print("\n---------------------------------")
    print("## Crew execution finished.")
    print("\nFinal Result:\n", result)

if __name__ == "__main__":
    main()

도구 설계에서 눈여겨볼 점

책이 이 예제에 “리팩터링된 도구”라는 주석을 붙인 이유가 있습니다. 도구는 문자열 대신 원시 데이터(float)를 반환하고, 실패하면 문자열이 아니라 ValueError 를 발생시킵니다.

문자열을 반환하는 것보다 구체적인 오류를 발생시키는 편이 낫다. 에이전트는 예외를 처리할 수 있으며, 다음 행동을 스스로 결정할 수 있다.

"종목을 찾을 수 없습니다" 같은 문자열을 반환하면 에이전트 입장에서는 그것이 정상적인 데이터인지 실패인지 구분할 수 없습니다. 예외를 던지면 실패가 명확해지고, 재사용성도 높아집니다. 그리고 태스크 설명에는 성공과 실패 양쪽 경우에 어떻게 대응해야 하는지를 명시했습니다.

실습 코드 (구글 ADK)

구글 ADK에는 에이전트가 수행할 수 있는 작업에 직접 통합할 수 있는 네이티브 도구 라이브러리가 포함되어 있습니다.

구글 서치 도구

구글 검색 엔진에 직접 연결되는 인터페이스 역할을 하며, 에이전트가 웹 검색을 수행하고 외부 정보를 가져올 수 있게 합니다.

from google.adk.agents import Agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.genai import types
import nest_asyncio
import asyncio

# 세션 설정 및 에이전트 실행에 필요한 변수 정의
APP_NAME = "Google Search_agent"
USER_ID = "user1234"
SESSION_ID = "1234"

# 검색 도구를 사용할 수 있는 에이전트 정의
root_agent = Agent(
    name="basic_search_agent",
    model="gemini-2.0-flash-exp",
    description="Agent to answer questions using Google Search.",
    instruction="I can answer your questions by searching the internet. Just ask me anything!",
    tools=[google_search]   # Google Search는 Google 검색을 수행하는 사전 구축 도구이다.
)

# 에이전트 상호작용
async def call_agent(query):
    """
    질의를 사용해 에이전트를 호출하는 헬퍼 함수.
    """
    # 세션 및 러너
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    runner = Runner(agent=root_agent, app_name=APP_NAME, session_service=session_service)
    content = types.Content(role='user', parts=[types.Part(text=query)])
    events = runner.run(user_id=USER_ID, session_id=SESSION_ID, new_message=content)
    for event in events:
        if event.is_final_response():
            final_response = event.content.parts[0].text
            print("Agent Response: ", final_response)

nest_asyncio.apply()
asyncio.run(call_agent("what's the latest ai news?"))

코드 실행 도구

built_in_code_execution 도구는 에이전트에게 샌드박스 처리된 파이썬 인터프리터를 제공합니다. 이를 통해 모델이 코드를 작성·실행하여 연산 작업을 수행하고, 데이터 구조를 조작하며, 절차적 스크립트를 실행할 수 있습니다.

이러한 기능은 결정적 논리와 정밀한 계산이 필요한 문제를 다루는 데 매우 중요합니다. 이러한 문제들은 확률적 언어 생성으로 해결하는 영역을 벗어나기 때문입니다.

import os, getpass
import asyncio
import nest_asyncio
from typing import List
from dotenv import load_dotenv
import logging
from google.adk.agents import Agent as ADKAgent, LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools import google_search
from google.adk.code_executors import BuiltInCodeExecutor
from google.genai import types

# 세션 설정 및 에이전트 실행에 필요한 변수 정의
APP_NAME = "calculator"
USER_ID = "user1234"
SESSION_ID = "session_code_exec_async"

# 에이전트 정의
code_agent = LlmAgent(
    name="calculator_agent",
    model="gemini-2.0-flash",
    code_executor=BuiltInCodeExecutor(),
    instruction="""You are a calculator agent.
When given a mathematical expression, write and execute Python code to calculate the result.
Return only the final numerical result as plain text, without markdown or code blocks.
""",
    description="Executes Python code to perform calculations.",
)

# 에이전트 상호작용(비동기)
async def call_agent_async(query):
    # 세션 및 러너
    session_service = InMemorySessionService()
    session = await session_service.create_session(app_name=APP_NAME, user_id=USER_ID, session_id=SESSION_ID)
    runner = Runner(agent=code_agent, app_name=APP_NAME, session_service=session_service)
    content = types.Content(role='user', parts=[types.Part(text=query)])
    print(f"\n--- Running Query: {query} ---")
    final_response_text = "No final text response captured."
    try:
        # run_async 사용
        async for event in runner.run_async(user_id=USER_ID, session_id=SESSION_ID, new_message=content):
            print(f"Event ID: {event.id}, Author: {event.author}")
            if event.content and event.content.parts and event.is_final_response():
                for part in event.content.parts:   # 모든 파트를 순회
                    if part.executable_code:
                        # .code를 통해 실제 코드 문자열에 접근
                        print(f"  Debug: Agent generated code:\n```python\n{part.executable_code.code}\n```")
                    elif part.code_execution_result:
                        # outcome과 output에 올바르게 접근
                        print(f"  Debug: Code Execution Result: {part.code_execution_result.outcome} - Output:\n{part.code_execution_result.output}")
                    elif part.text and not part.text.isspace():
                        print(f"  Text: '{part.text.strip()}'")
                text_parts = [part.text for part in event.content.parts if part.text]
                final_result = "".join(text_parts)
                print(f"==> Final Agent Response: {final_result}")
    except Exception as e:
        print(f"ERROR during agent run: {e}")
    print("-" * 30)

# 예제를 실행하는 메인 비동기 함수
async def main():
    await call_agent_async("Calculate the value of (5 + 7) * 3")
    await call_agent_async("What is 10 factorial?")

# 메인 비동기 함수 실행
try:
    nest_asyncio.apply()
    asyncio.run(main())
except RuntimeError as e:
    # 이미 실행 중인 이벤트 루프(Jupyter/Colab 등)에서 asyncio.run을 호출할 때 발생하는 오류 처리
    if "cannot be called from a running event loop" in str(e):
        print("\nRunning in an existing event loop (like Colab/Jupyter).")
        print("Please run `await main()` in a notebook cell instead.")
        # 노트북 같은 인터랙티브 환경에서는 아래와 같이 실행해야 할 수 있다:
        # await main()
    else:
        raise e  # 그 외 런타임 오류는 다시 발생시킨다

이벤트를 순회하면서 생성된 파이썬 코드(part.executable_code)와 실행 결과(part.code_execution_result)를 구분해 출력하는 부분이 이 예제의 핵심입니다. 중간 단계와 최종 수치 답변이 담긴 이벤트를 구분할 수 있습니다.

엔터프라이즈 검색 (Vertex AI Search)

지정된 Vertex AI Search 데이터 저장소를 검색하여 질문에 답변하도록 설계된 VSearchAgent 입니다.

import asyncio
from google.genai import types
from google.adk import agents
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
import os

# --- 설정 ---
# GOOGLE_API_KEY와 DATASTORE_ID 환경 변수가 설정되어 있는지 확인한다
# 예시:
# os.environ["GOOGLE_API_KEY"] = "YOUR_API_KEY"
# os.environ["DATASTORE_ID"] = "YOUR_DATASTORE_ID"

DATASTORE_ID = os.environ.get("DATASTORE_ID")

# --- 애플리케이션 상수 ---
APP_NAME = "vsearch_app"
USER_ID = "user_123"      # 사용자 ID 예시
SESSION_ID = "session_456"  # 세션 ID 예시

# --- 에이전트 정의(가이드에 나온 최신 모델로 업데이트) ---
vsearch_agent = agents.VSearchAgent(
    name="q2_strategy_vsearch_agent",
    description="Answers questions about Q2 strategy documents using Vertex AI Search.",
    model="gemini-2.0-flash-exp",   # 가이드 예제 기준으로 업데이트된 모델
    datastore_id=DATASTORE_ID,
    model_parameters={"temperature": 0.0}
)

# --- 러너 및 세션 초기화 ---
runner = Runner(
    agent=vsearch_agent,
    app_name=APP_NAME,
    session_service=InMemorySessionService(),
)

# --- 에이전트 호출 로직 ---
async def call_vsearch_agent_async(query: str):
    """세션을 초기화하고 에이전트의 응답을 스트리밍한다."""
    print(f"User: {query}")
    print("Agent: ", end="", flush=True)
    try:
        # 메시지 콘텐츠를 올바르게 구성
        content = types.Content(role='user', parts=[types.Part(text=query)])
        # 비동기 러너에서 도착하는 이벤트를 순차 처리
        async for event in runner.run_async(
            user_id=USER_ID,
            session_id=SESSION_ID,
            new_message=content
        ):
            # 응답 텍스트를 토큰 단위로 스트리밍
            if hasattr(event, 'content_part_delta') and event.content_part_delta:
                print(event.content_part_delta.text, end="", flush=True)
            # 최종 응답과 관련 메타데이터 처리
            if event.is_final_response():
                print()   # 스트리밍 응답 뒤에 줄바꿈
                if event.grounding_metadata:
                    print(f"  (Source Attributions: {len(event.grounding_metadata.grounding_attributions)} sources found)")
                else:
                    print("  (No grounding metadata found)")
                print("-" * 30)
    except Exception as e:
        print(f"\nAn error occurred: {e}")
        print("Please ensure your datastore ID is correct and that the service account has the necessary permissions.")
        print("-" * 30)

# --- 실행 예제 ---
async def run_vsearch_example():
    # 자신의 데이터 저장소 내용에 맞는 질문으로 교체한다
    await call_vsearch_agent_async("Summarize the main points about the Q2 strategy document.")
    await call_vsearch_agent_async("What safety procedures are mentioned for lab X?")

# --- 실행 ---
if __name__ == "__main__":
    if not DATASTORE_ID:
        print("Error: DATASTORE_ID environment variable is not set.")
    else:
        try:
            asyncio.run(run_vsearch_example())
        except RuntimeError as e:
            # 이미 실행 중인 이벤트 루프가 있는 환경(주피터(Jupyter) 노트북 등)에서
            # asyncio.run을 호출할 때 발생할 수 있는 문제를 처리한다.
            if "cannot be called from a running event loop" in str(e):
                print("Skipping execution in a running event loop. Please run this script directly.")
            else:
                raise e

응답을 토큰 단위로 스트리밍하고, 데이터 저장소에서 가져온 출처 정보(source attribution) 를 grounding_metadata 로 함께 출력하는 점이 특징입니다.

Vertex 확장 프로그램 — 함수 호출과 무엇이 다른가

Vertex AI 확장 프로그램은 모델이 외부 API에 연결하여 실시간 데이터 처리와 동작 실행을 수행할 수 있게 하는 구조화된 API 래퍼로, 엔터프라이즈급 보안·데이터 프라이버시를 지원하며 성능 수준도 보장합니다. 구글은 Code Interpreter와 Vertex AI Search 같은 일반적인 사용 사례를 위해 사전 구축 확장 프로그램을 제공하며, 필요하면 맞춤형 확장 프로그램을 만들 수도 있습니다.

확장 프로그램과 함수 호출의 핵심 차이는 실행 방식에 있습니다. Vertex AI가 확장 프로그램을 자동으로 실행하는 반면, 함수 호출은 사용자나 클라이언트가 수동으로 실행해야 합니다.

이것이 앞서 “3단계에서 LLM은 요청만 생성하고 실행은 오케스트레이션 계층의 몫”이라고 한 부분과 직결됩니다. 확장 프로그램은 그 오케스트레이션까지 플랫폼이 가져갑니다.

정리

도구 사용이란 무엇인가?

LLM은 강력한 텍스트 생성기이지만, 본질적으로 외부 세계와 단절되어 있습니다. LLM의 지식은 정적이고 학습 데이터에 한정되며, 동작을 수행하거나 실시간 정보를 가져오는 능력이 없습니다. 이러한 고유한 한계 때문에 LLM은 외부 API, 데이터베이스, 서비스와 상호작용해야 하는 작업을 완수할 수 없습니다. 외부 시스템과 연결해 주는 다리가 없다면, 현실 세계의 문제 해결에서 LLM의 활용성은 크게 제한됩니다.

왜 사용하는가?

도구 사용 패턴은 흔히 함수 호출로 구현되며, 텍스트 생성만으로는 해결하지 못하는 문제에 대한 표준화된 해법을 제공합니다. 사용 가능한 외부 함수, 즉 “도구”를 LLM이 이해할 수 있는 방식으로 기술하는 것이 핵심입니다. 사용자 요청을 받으면 에이전틱 LLM은 도구가 필요한지 판단하고, 어떤 함수를 어떤 인자로 호출할지 지정하는 구조화된 데이터 객체(JSON 등)를 생성합니다. 오케스트레이션 계층이 이 함수 호출을 실행하고, 결과를 가져와 LLM에게 다시 전달합니다. 이를 통해 LLM은 최신 외부 정보나 동작 결과를 최종 응답에 반영할 수 있으며, 사실상 행동 능력을 갖추게 됩니다.

언제 사용해야 하는가?

에이전트가 LLM의 내부 지식을 벗어나 외부 세계와 상호작용해야 할 때마다 도구 사용 패턴을 적용합니다.

• 실시간 데이터가 필요한 작업 (날씨, 주가 조회)
• 비공개 또는 독점 정보에 접근해야 하는 작업 (기업 데이터베이스 질의)
• 정밀한 계산, 코드 실행
• 다른 시스템에서 동작 실행 (이메일 전송, 스마트 기기 제어)

도구 사용 패턴 개요도

                          ┌────► 에이전트 ────┐
  사용자 ──► 프롬프트 ────┘        ▲   │       │
     ▲                             │   ▼       ▼
     │                            도구 (Tools)  출력
     └───────────────────────────────────────────┘

_{그림 5.2 — 도구 사용 패턴}

핵심 정리

• 도구 사용(함수 호출)은 에이전트가 외부 시스템과 상호작용하고 동적 정보에 접근할 수 있게 합니다.
• 이를 위해 LLM이 이해할 수 있도록 명확한 설명과 매개변수를 갖춘 도구를 정의해야 합니다.
• LLM은 언제 도구를 사용할지 판단하고, 구조화된 함수 호출을 생성합니다.
• 에이전틱 프레임워크는 실제 도구 호출을 실행하고 그 결과를 LLM에게 반환합니다.
• ‘도구 사용’은 현실 세계에서 동작을 수행하고 최신 정보를 제공할 수 있는 에이전트를 구축할 때 핵심적인 역할을 수행합니다.
• LangChain은 @tool 데코레이터로 도구 정의를 간소화하고, create_tool_calling_agent 와 AgentExecutor 를 제공하여 도구를 활용하는 에이전트를 구축할 수 있게 합니다.
• 구글 ADK에는 구글 검색, 코드 실행, Vertex AI Search Tool 등 매우 유용한 사전 구축 도구가 다수 포함되어 있습니다.

결론

도구 사용 패턴은 대규모 언어 모델의 기능 범위를 고유한 텍스트 생성 능력 너머로 확장하기 위한 핵심 설계 원칙입니다. 모델이 외부 소프트웨어와 데이터 소스에 연결될 수 있게 함으로써, 이 패러다임은 에이전트가 동작을 수행하고, 연산을 실행하며, 다른 시스템에서 정보를 가져올 수 있게 합니다.

이 과정에서 모델은 사용자 질의를 이행하는 데 외부 도구 호출이 필요하다고 판단하면 구조화된 요청을 생성합니다. LangChain, Google ADK, CrewAI 같은 프레임워크는 이러한 외부 도구 통합을 쉽게 할 수 있도록 구조화된 추상화와 구성 요소를 제공합니다. 프레임워크들은 도구 명세를 모델에 노출하고, 모델이 생성하는 도구 사용 요청을 파싱하는 과정을 관리합니다. 그 결과 외부 디지털 환경과 상호작용하고 그 안에서 행동을 수행하는 정교한 에이전틱 시스템을 더 쉽게 개발할 수 있습니다.

참고 문헌

• CrewAI 공식 문서 (도구): https://docs.crewai.com/concepts/tools
• 구글 ADK 공식 문서 (도구): https://google.github.io/adk-docs/tools/
• LangChain 공식 문서 (도구): https://python.langchain.com/docs/integrations/tools/
• 오픈AI 함수 호출 공식 문서: https://platform.openai.com/docs/guides/function-calling