LangGraph

AgentOps로 LangGraph 워크플로우를 추적하고 분석하는 방법을 소개해요. 핵심 개념부터 설치, 기본·고급 예제, 주요 습관까지 알아볼게요.

출처: 문서

본문

LangGraph는 LLM을 그래프로 사용해 상태 기반(stateful) 다단계 애플리케이션을 구축하는 프레임워크예요. AgentOps는 LangGraph를 자동으로 계측해 그래프 기반 에이전트 워크플로우에 대한 포괄적인 관측성을 제공합니다.

핵심 개념 (Core Concepts)

LangGraph를 사용하면 복잡한 에이전트 워크플로우를 그래프로 구축할 수 있으며, 다음과 같은 요소로 구성됩니다.

Nodes: 워크플로우의 개별 단계 (에이전트, 도구, 함수)
Edges: 흐름을 정의하는 노드 사이의 연결
State: 그래프를 통해 흐르는 공유 데이터
Conditional Edges: 상태나 출력에 기반한 동적 라우팅
Cycles: 반복 워크플로우와 피드백 루프 지원

설치 (Installation)

AgentOps와 LangGraph를 LangChain 의존성과 함께 설치하세요.

```bash pip theme={null} pip install agentops langgraph langchain-openai python-dotenv ```

poetry add agentops langgraph langchain-openai python-dotenv

uv pip install agentops langgraph langchain-openai python-dotenv

API 키 설정 (Setting Up API Keys)

AgentOps와 LLM 프로바이더에 대한 API 키가 필요해요.

OPENAI_API_KEY: OpenAI Platform에서
AGENTOPS_API_KEY: AgentOps Dashboard에서

이것들을 환경 변수나 .env 파일로 설정하세요.

```bash Export to CLI theme={null} export OPENAI_API_KEY="your_openai_api_key_here" export AGENTOPS_API_KEY="your_agentops_api_key_here" ```

OPENAI_API_KEY="your_openai_api_key_here"
AGENTOPS_API_KEY="your_agentops_api_key_here"

그리고 Python 코드에서 로드합니다.

from dotenv import load_dotenv
import os

load_dotenv()

AGENTOPS_API_KEY = os.getenv("AGENTOPS_API_KEY")
OPENAI_API_KEY = os.getenv("OPENAI_API_KEY")

사용법 (Usage)

애플리케이션 시작 시 AgentOps를 초기화하면 모든 LangGraph 작업이 자동으로 추적됩니다.

import agentops
from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI

# Initialize AgentOps
agentops.init()

# Define your graph state
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# Create your LLM
model = ChatOpenAI(temperature=0)

# Define nodes
def agent_node(state: AgentState):
    messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", agent_node)
workflow.set_entry_point("agent")
workflow.add_edge("agent", END)

# Compile and run
app = workflow.compile()
result = app.invoke({"messages": [{"role": "user", "content": "Hello!"}]})

무엇이 추적되나요? (What Gets Tracked)

AgentOps는 자동으로 다음을 캡처합니다.

Graph Structure: 컴파일 중의 노드, 엣지, 엔트리 포인트
Execution Flow: 그래프를 통해 지나간 경로
Node Executions: 입력과 출력을 포함한 각 노드 실행
LLM Calls: 노드 안에서의 모든 언어 모델 상호작용
Tool Usage: 그래프 안에서 호출된 모든 도구
State Changes: 워크플로우를 통과하며 상태가 어떻게 진화하는지
Timing Information: 각 노드의 기간과 전체 실행 시간

고급 예제 (Advanced Example)

조건부 라우팅과 도구를 사용한 더 복잡한 예제는 다음과 같습니다.

import agentops
from typing import Annotated, Literal, TypedDict
from langgraph.graph import StateGraph, END
from langgraph.graph.message import add_messages
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool

# Initialize AgentOps
agentops.init()

# Define tools
@tool
def search(query: str) -> str:
    """Search for information."""
    return f"Search results for: {query}"

@tool
def calculate(expression: str) -> str:
    """Evaluate a mathematical expression."""
    try:
        return str(eval(expression))
    except:
        return "Error in calculation"

# Configure model with tools
tools = [search, calculate]
model = ChatOpenAI(temperature=0).bind_tools(tools)

# Define state
class AgentState(TypedDict):
    messages: Annotated[list, add_messages]

# Define conditional logic
def should_continue(state: AgentState) -> Literal["tools", "end"]:
    messages = state["messages"]
    last_message = messages[-1]
    
    if hasattr(last_message, "tool_calls") and last_message.tool_calls:
        return "tools"
    return "end"

# Define nodes
def call_model(state: AgentState):
    messages = state["messages"]
    response = model.invoke(messages)
    return {"messages": [response]}

def call_tools(state: AgentState):
    messages = state["messages"]
    last_message = messages[-1]
    
    tool_responses = []
    for tool_call in last_message.tool_calls:
        # Execute the appropriate tool
        if tool_call["name"] == "search":
            result = search.invoke(tool_call["args"])
        elif tool_call["name"] == "calculate":
            result = calculate.invoke(tool_call["args"])
        
        tool_responses.append({
            "role": "tool",
            "content": result,
            "tool_call_id": tool_call["id"]
        })
    
    return {"messages": tool_responses}

# Build the graph
workflow = StateGraph(AgentState)
workflow.add_node("agent", call_model)
workflow.add_node("tools", call_tools)
workflow.set_entry_point("agent")
workflow.add_conditional_edges(
    "agent",
    should_continue,
    {
        "tools": "tools",
        "end": END
    }
)
workflow.add_edge("tools", "agent")

# Compile and run
app = workflow.compile()
result = app.invoke({
    "messages": [{"role": "user", "content": "Search for AI news and calculate 25*4"}]
})

대시보드 인사이트 (Dashboard Insights)

AgentOps 대시보드에서 다음을 볼 수 있습니다.

Graph Visualization: 컴파일된 그래프의 시각적 표현
Execution Trace: 노드를 통한 단계별 흐름
Node Metrics: 각 노드의 성능 데이터
LLM Analytics: 모든 모델 호출의 토큰 사용량과 비용
Tool Usage: 호출된 도구와 그 결과
Error Tracking: 노드 실행에서의 모든 실패

예제 (Examples)

도구를 사용한 에이전트 워크플로우를 보여주는 완전한 예제

모범 사례 (Best Practices)

일찍 초기화하세요: 그래프를 만들기 전에 agentops.init()을 호출하세요
설명적인 이름을 사용하세요: 더 나은 트레이스를 위해 노드를 명확하게 명명하세요
오류를 처리하세요: 노드에서 오류 처리를 구현하세요
상태 크기를 모니터링하세요: 큰 상태는 성능에 영향을 줄 수 있어요
조건부 엣지를 활용하세요: 동적 워크플로우에 사용하세요

더 알아보기 (Learn more)

AgentOps 공식 문서