Vertex AI Agent Engine을 사용한 나만의 LLM Agent 배포

김민형·2025년 4월 14일

AI/ML Agent Engine GCP Gemini Generative AI LLM Vertex AI agent langChain langgraph

GCP - AI,ML

목록 보기

22/23

Vertex AI Agent Engine이란?

LLM 기반 애플리케이션을 GCP에서 배포, 관리, 확장할 수 있는 완전관리형 서비스다.

LangChain, LangGraph 프레임워크 기반 LLM 애플리케이션을 로컬에서 개발한 후 해당 코드를 GCP상으로 호스팅하여 배포 및 관리할 수 있다.

애플리케이션을 로컬에 배포하는 것과의 차이점은?
GCP의 Observability를 확보할 수 있다는 것이고 이를 통해 관리가 된다는 점이다.

아래에선 LangGraph 기반 LLM Agent를 로컬에서 개발한 후 테스트 하고 Agent Engine에 배포해볼 것이다.

Agent 개발

다음은 날짜 혹은 인구 정보를 얻을 수 있는 tool을 정의해놓고 호출하는 간단한 Agent이다.

여기선 Gemini를 사용했지만 OPENAI_API_KEY 환경변수를 선언해주고 아래 라이브러리를 import 하고 ChatVertexAI 부분만 ChatOpenAI로 변경하면 GPT 기반 Agent가 동작하게 된다.
from langchain_openai import ChatOpenAI

import os
os.environ["GOOGLE_API_KEY"] = "<API KEY>"

from typing import Annotated, Literal
from typing_extensions import TypedDict
from langgraph.graph import START, END
from langgraph.graph import StateGraph
from langchain_core.tools import tool
from langgraph.graph.message import add_messages
from langgraph.prebuilt import ToolNode
from langchain_google_vertexai import ChatVertexAI

# State 선언
class State(TypedDict):
    messages: Annotated[list, add_messages]

@tool
def get_weather(location: str):
    """A tool that returns weather information for a location."""
    return f"{location}의 날씨는 맑습니다."

@tool
def get_population(location: str):
    """A tool that returns population information for a location."""
    return f"{location}의 인구는 100명 입니다."

def should_continue(state: State) -> Literal["tools", "end"]:
    messages = state['messages']
    last_message = messages[-1]

    if last_message.tool_calls:
        return "continue"
    else:
        return "end"

class LGApp:
    def __init__(self, project: str, location: str) -> None:
        self.project_id = project
        self.location = location

    def set_up(self) -> None:
        tools = [get_weather, get_population]
        model = ChatVertexAI(model="gemini-1.5-pro-002", temperature=0)
        model_with_tools = model.bind_tools(tools)

        def agents(state: State):
            response = model_with_tools.invoke(state["messages"])
            state["messages"].append(response)
            return state

        # Graph 선언
        workflow = StateGraph(State)

        # Node 추가
        workflow.add_node("agent", agents)
        tool_node = ToolNode(tools)
        workflow.add_node("tools", tool_node)

        # 노드 연결
        workflow.add_edge(START, "agent")
        workflow.add_conditional_edges(
            "agent",
            should_continue,
            {
                "continue": "tools",
                # Otherwise we finish.
                "end": END,
            },
        )

        self.runnable = workflow.compile()
    
    def query(self, message: str):
        result = self.runnable.invoke({"messages": message})
        return result

개발시 기존 LangChain, LangGraph 프레임워크에 맞게 사용해서 개발해도 되지만 Agent Engine에서 지원하는 LangChain, LangGraph 에이전트 개발 양식을 따라도 된다.

LangChain
Vertex AI Agent Engine - LangChain Agent 개발

from vertexai import agent_engines

agent = agent_engines.LangchainAgent(
    model=model,                # Required.
    model_kwargs=model_kwargs,  # Optional.
)

LangGraph
Vertex AI Agent Engine - LangGraph Agent 개발

from vertexai import agent_engines

agent = agent_engines.LanggraphAgent(
    model=model,                # Required.
    model_kwargs=model_kwargs,  # Optional.
)

Agent 로컬 테스트

로컬에서 테스트

message = "서울의 날씨는 어때?"

vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=f"gs://{BUCKET_NAME}",
)

agent_app = LGApp(project=PROJECT_ID, location=LOCATION)
agent_app.set_up()
response = agent_app.query(message)
print(response)

결과

적절한 tool을 호출하여 '서울의 날씨는 맑다'는 답변을 제대로 한 것을 솩인할 수 있다.

Agent 배포

Agent Engine을 사용하여 Agent 인스턴스 생성.

GPT를 사용할 경우 requirements에 langchain-openai를 추가해주면 된다.

vertexai.init(
    project=PROJECT_ID,
    location=LOCATION,
    staging_bucket=f"gs://{BUCKET_NAME}",
)

mhk_agent = agent_engines.create(
    LGApp(project=PROJECT_ID, location=LOCATION),
    requirements=[
        "google-cloud-aiplatform[agent_engines,langchain]",
        "langchain-google-vertexai",
        "langgraph==0.2.76",
        "cloudpickle>=3.0.0",
        "pydantic>=2.10",
    ],
    gcs_dir_name='Agent_Engine/dev',
    display_name="Agent_Engine_Test",
    description="Agent Engine based on LangGraph",
    extra_packages=["agent/langgraph_template.py"],
)

Tips

해당 기능의 경우 google-cloud-aiplatform==1.81.0 이상의 최신 버전에서 동작하고 LangChain과 LangGraph 등 기타 패키지들의 종속성이 호환되는 버전이어야 하므로 로컬 테스트를 진행해보고 정상 동작하면 로컬에 설치된 패키지 버전을 그대로 지정해주는 것을 권장한다.
requirements에 대한 내용은 requirements.txt파일을 생성한 후 파일 경로를 참조해도 된다.
인스턴스에 대한 관련 파일들은 지정된 버킷 경로에 저장되고 관리된다.
다른 파일의 모듈을 import해서 실행되는 코드라면 extra_packages에서 해당 코드 경로도 지정해줘야 한다.

결과

인스턴스 생성 성공!!
생성 완료 메시지에서 Resource name은 뒤에서 사용되니 복사해두자.

검색

RESOURCE_ID="<위에서 복사해놓은 Resource name>"

mhk_agent = agent_engines.get(RESOURCE_ID)
response = mhk_agent.query(message={"role": "user", "content": "<질문>"})

print(response)