LangChain v1.0으로 멀티 에이전트 만들기: 미들웨어로 완성하는 리포트 AI

[CLOVA Studio 운영자6]

들어가며

이전에 살펴본 LangGraph로 웹 검색 Agent 만들기 (Web Search Agent Cookbook)에서는 LangGraph를 활용한 단일 에이전트 구축 방법을 다뤘다면, 이번에는 LangChain & LangGraph v1.0을 활용해 멀티 에이전트 시스템을 구축하는 방법을 알아보겠습니다.

기존 LangGraph에서 멀티 에이전트를 구축하려면 모델과 도구를 호출하는 노드를 일일이 정의하고, 워크플로우를 수동으로 연결해야 했습니다. 유연성은 높지만 시스템이 복잡해질수록 그래프 구조를 파악하기 어렵고 유지보수 부담이 커지는 문제가 있었죠.

LangChain과 LangGraph가 v1.0으로 정식 출시되면서 AI 에이전트 개발이 한층 안정적이고 강력해졌습니다. 특히, LangChain v1.0부터 도입된 create_agent는 내부적으로 LangGraph 기반으로 구축되어 있어, LangGraph를 직접 다루지 않아도 Durable Execution, Streaming, Human-In-the-Loop 같은 강력한 기능들을 자동으로 활용할 수 있습니다. 또한, 미들웨어(Middleware) 기능으로 로깅, 프롬프트 수정, 에러 핸들링 같은 공통 로직을 깔끔하게 분리할 수 있게 되었습니다.

이번 쿡북에서는 LangChain v1.0과 HyperCLOVA X 모델을 활용해 Tool Calling 기반의 멀티 에이전트를 구축합니다. 웹 검색(Web Search), 글쓰기(Write), 저장(Save) 에이전트가 협업하여, 최신 정보를 검색하고, 목적에 맞는 글을 작성한 뒤, Notion이나 파일로 자동 저장하는 리포트 AI를 만들어봅니다. 이 가이드를 발판 삼아 여러분만의 창의적인 에이전트 시스템을 자유롭게 구축해 보세요.

 

사전 준비

멀티 에이전트 구축하기 프로젝트 진행을 위해서는 사전 준비 과정이 필요합니다. 각 과정마다 발급되는 키들을 환경 변수로 등록해 둡니다.

API Key 발급 및 연동 설정

CLOVA Studio API

CLOVA Studio 모델을 사용하기 위해 CLOVA Studio에서 API 키를 발급받아야 합니다. 본 예제에서는 HCX-005와 HCX-007 모델을 사용합니다.

• CLOVA Studio 접속 > 로그인 > 좌측 사이드바 'API 키' > 테스트 API 키 발급

발급된 키는 한 번만 표시되므로 반드시 복사하여 안전하게 보관하세요.  API에 대한 자세한 내용은 CLOVA Studio API 가이드를 참고하시기 바랍니다.

네이버 검색 API

네이버 검색 오픈API를 활용하기 위해 네이버 개발자 센터에서 애플리케이션을 등록해야 합니다. 등록을 완료하면 Client ID와 Client Secret 정보를 확인할 수 있습니다. 자세한 내용은 네이버 개발자 센터의 애플리케이션 등록 가이드를 참고해 주세요.

• 네이버 개발자 센터 접속 > 로그인 > Application > 애플리케이션 등록
• 애플리케이션 등록 설정
  1. 애플리케이션 이름을 설정합니다
  2. 사용 API에서 '검색'을 선택합니다
  3. 비로그인 오픈 API 서비스 환경에서 'Web 환경'을 추가합니다.

Tavily API

에이전트가 Tavily 웹 검색 기능을 사용하기 위해, Tavily API 키를 발급받아야 합니다. 무료 플랜의 경우 하루 1,000회까지 검색이 가능합니다.

• Tavily 웹 페이지 접속 > 로그인 > API Keys

Notion API

에이전트가 개인 Notion 데이터베이스에 글을 저장하려면 Notion API Key와 업로드 대상 Data Source의 ID가 필요합니다.

• Notion API Key 발급 : Notion Developers 페이지 접속 > 우측 상단 'View my Integrations' > 로그인 > 새 API 통합
• Data Source ID 확인 : 개인 Notion 접속 > API Key 발급시 설정한 워크스페이스로 이동 > 죄측 상단 '추가 옵션' > '데이터베이스' 선택 > 데이터베이스 설정 > 데이터 소스 관리 > 만들어진 데이터베이스 더보기 > 데이터 소스 ID 복사

 

프로젝트 구성

 프로젝트의 전체 파일 구조는 다음과 같습니다. 프로젝트에 사용한 파이썬 버전은 3.11 입니다.

multi-agent-cookbook/
├── agent.py
├── utils/
│   ├── tool_agents.py
│   ├── prompts.py
│   ├── custom_middleware.py
│   └── tools.py
├── langgraph.json
├── requirements.txt
└── .env

환경 변수 설정

환경 변수를 설정합니다. 루트 디렉터리에 .env 파일을 생성한 뒤, 앞서 발급받은 API Key를 다음과 같이 입력하고 저장합니다. 이때 따옴표 없이 값을 작성해야 하며, VS Code에서 실행할 경우 설정에서 Use Env File 옵션이 활성화되어 있는지 확인하세요.

CLOVA_STUDIO_API_KEY=nv-...
OPENAI_API_KEY=sk-proj-...

LANGSMITH_API_KEY=lsv2_pt_...
LANGSMITH_TRACING=true
LANGSMITH_PROJECT=Multi-agent-cookbook

NAVER_CLIENT_ID=...
NAVER_CLIENT_SECRET=...
TAVILY_API_KEY=tvly-dev-...

NOTION_API_KEY=ntn_...
NOTION_DATA_SOURCE_ID=...

라이브러리 설치

프로젝트에 필요한 패키지 목록은 아래 다운로드 링크에서 확인할 수 있습니다. 해당 내용을 복사해 루트 디렉터리에 requirements.txt 파일로 저장하세요.

requirements.txt 다운로드

루트 디렉터리에서 터미널을 실행하여 다음과 같이 예제 실행에 필요한 패키지를 설치합니다. 가상환경 설치를 권장합니다.

# 1. 파이썬 가상환경 생성
python -m venv .venv

# 2. 가상환경 활성화 (macOS/Linux)
source .venv/bin/activate
#    (Windows)
#    .venv/Scripts/activate.ps1

# 3. 패키지 설치
pip install -r requirements.txt

 

멀티 에이전트 구축하기

멀티 에이전트 패턴

LangChain v1.0은 두 가지 멀티 에이전트 패턴을 제공합니다. 

1. Tool Calling 패턴

중앙 Controller Agent가 모든 작업을 조율하고, 하위 에이전트들을 도구로 호출하는 구조입니다. 사용자와의 소통은 Controller Agent만 담당하며, Tool Agent는 Controller Agent의 도구로서 특정 작업만 수행합니다. Tool Calling 패턴으로 멀티 에이전트를 구현하면, 명확한 계층 구조와 모듈화로 관리가 쉽고 디버깅이 용이하지만, 모든 결정이 Controller를 거쳐야 하므로 토큰 소비량이 크고, 응답 시간이 다소 길어질 수 있습니다.

2. Handoffs 패턴

현재 에이전트가 다른 에이전트에게 제어권을 넘기는 방식으로, 각 에이전트가 순차적으로 사용자와 소통합니다. 이는 쿡북 작성일 기준으로 LangChain에서 아직 제공하지 않는 기능입니다.

이번 쿡북에서는 Tool Calling 패턴을 사용하며, 다음은 이를 기반으로 한 멀티 에이전트 아키텍처입니다.

• 관리자 에이전트(Controller Agent)
  • 작업에 적합한 전문가 에이전트를 판단하여 호출하고, 그 결과를 받아 전체 흐름을 조율하는 오케스트레이터(Orchestrator) 역할을 수행합니다.
    • LoggingMiddleware: 에이전트의 모든 실행 단계를 로깅하여 디버깅을 돕습니다.
    • DynamicModelMiddleware: 대화 길이가 5개 보다 많아지면, 자동으로 더 강력한 모델로 전환합니다.
    • HumanInTheLoopMiddleware: 민감한 도구 호출 전, 정지상태가 되는 interrupt를 발생시킵니다.
• 전문가 에이전트(Tool Agents)
  • 웹 검색 에이전트(Web Search Agent): 네이버/Tavily API를 활용한 웹 검색을 수행합니다. 네이버 웹 검색 실패 시, 자동으로 Tavily 웹 검색으로 전환하여 웹 검색 안정성을 높입니다.
  • 글쓰기 에이전트(Write Agent): 검색 결과를 바탕으로 리포트 또는 블로그 형식의 글을 작성합니다.
  • 저장 에이전트(Save Agent): 작성된 콘텐츠를 Notion 또는 파일 시스템에 저장합니다.

 

미들웨어 구현하기

미들웨어는 에이전트의 실행 과정에 개입하여 기능을 확장하는 컴포넌트입니다. 미들웨어는 LangChain v1.0의 핵심 기능으로, 에이전트 로직을 수정하지 않고도 로깅, 프롬프트 변경, 에러 처리 등을 추가할 수 있게 합니다.

미들웨어가 개입할 수 있는 시점

미들웨어는 에이전트 실행의 다양한 시점에서 작동할 수 있습니다.

Node-style hooks

특정 실행 지점에서 순차적으로 실행됩니다. 로깅, 유효성 검사, 상태 업데이트에 사용합니다.

• before_agent / after_agent: 에이전트 시작 전후
• before_model / after_model: 모델 호출 전후

Wrap-style hooks

핸들러 호출을 가로채고 실행을 제어합니다. 재시도, 캐싱, 변환에 사용합니다. 핸들러를 0번(조기 종료), 1번(정상 흐름), 또는 여러 번(재시도 로직) 호출할지 결정할 수 있습니다.

• wrap_tool_call: 도구 실행 시
• wrap_model_call: 모델 실행 시

Convenience

• dynamic_prompts: 동적 시스템 프롬프트 생성

미들웨어 구축 방식

LangChain에서는 목적과 복잡도에 따라 여러 방식으로 미들웨어를 구성할 수 있습니다. 

• Decorator-based middleware: 단일 훅(Hook)만 필요한 간단한 로직을 적용하거나, 별도의 설정 없이 빠르게 프로토타이핑을 진행할 때 사용하면 좋습니다.
• Class-based middleware: 동기/비동기 처리를 모두 지원해야 하거나, 여러 개의 훅과 복잡한 설정을 하나의 모듈로 묶어 체계적으로 관리해야 할 때 적합합니다.
• Built-In Middleware: LangChain에서 제공하는 기본 미들웨어를 바로 사용할 수 있습니다.

이 프로젝트에서 사용할 미들웨어

이 프로젝트에서는 에이전트 실행 흐름을 제어하고 안정성을 높이기 위해 다음과 같은 미들웨어를 사용합니다. 커스텀 미들웨어를 구성할 때는 각 미들웨어가 실행 흐름에 개입하는 시점을 고려해야 하며, 이때 LoggingMiddleware는 미들웨어의 개입 지점을 이해하는 데 도움이 될 수 있습니다.

• HumanInTheLoopMiddleware : 민감한 도구 호출 전, 정지상태가 되는 interrupt를 발생시킵니다.
• LoggingMiddleware : 에이전트의 모든 실행 단계를 로깅하여 디버깅을 돕습니다.
• DynamicModelMiddleware : 대화 길이가 5개 보다 많아지면, 자동으로 더 강력한 모델로 전환합니다.
• NaverToTavilyFallbackMiddleware : 네이버 웹 검색 실패 시, 자동으로 Tavily 웹 검색으로 전환하여 웹 검색 안정성을 높입니다.
• writing_farmat : 사용자가 요청한 글 타입(report/blog)에 따라 동적으로 시스템 프롬프트를 변경합니다.

# custom_middleware.py

import os
from typing import Any, Callable
from langchain_naver import ChatClovaX
from langchain.agents.middleware import AgentMiddleware, AgentState, ModelRequest, dynamic_prompt
from langchain.agents.middleware.types import ModelResponse, ToolCallRequest
from langchain.messages import ToolMessage
from langgraph.types import Command
from langgraph.runtime import Runtime

from .tools import tavily_web_search
from .prompts import WRITING_PROMPTS

from dotenv import load_dotenv

load_dotenv()

class ReportAgentState(AgentState):
    content_type : str
    destination : str

class LoggingMiddleware(AgentMiddleware):
    """에이전트 실행 과정 로깅"""
    async def abefore_agent(self, state: ReportAgentState, runtime: Runtime) -> dict[str, Any] | None:
        print("\n" + "="*60)
        print(f"🔄 LoggingMiddleware.abefore_agent")
        print(f"🚀 에이전트 시작")
        print(f"   메시지 수: {len(state['messages'])}개")
        print("="*60)
        return None

    async def abefore_model(self, state: ReportAgentState, runtime: Runtime) -> dict[str, Any] | None:
        print("\n" + "-"*60)
        print(f"🔄 LoggingMiddleware.abefore_model")
        if state['messages']:
            last = state['messages'][-1]
            print(f"🤖 모델 호출")
            print(f"   입력: {type(last).__name__}")
            if hasattr(last, 'content') and last.content:
                preview = last.content[:60] + "..." if len(last.content) > 60 else last.content
                print(f"   내용: {preview}")
        print("-"*60)
        print(f"\nController 응답 중..")
        return None

    async def aafter_model(self, state: ReportAgentState, runtime: Runtime) -> dict[str, Any] | None:
        print("\n" + "-"*60)
        print(f"🔄 LoggingMiddleware.aafter_model")
        if state['messages']:
            last = state['messages'][-1]
            if hasattr(last, 'tool_calls') and last.tool_calls:
                tools = [tc['name'] for tc in last.tool_calls]
                print(f"🔧 도구 호출 예정")
                for tool in tools:
                    print(f"   → {tool}")
            elif hasattr(last, 'content') and last.content:
                print(f"✅ 모델 응답 완료")
        print("-"*60)
        return None

    async def aafter_agent(self, state: ReportAgentState, runtime: Runtime) -> dict[str, Any] | None:
        print("\n" + "="*60)
        print(f"🔄 LoggingMiddleware.aafter_agent")
        print(f"🏁 에이전트 완료")
        print(f"   총 메시지: {len(state['messages'])}개")
        print("="*60 + "\n")
        return None
    
    async def awrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command],
    ) -> ToolMessage | Command:
        tool_name = request.tool_call.get('name', 'unknown')
        
        print("\n" + "-"*60)
        print(f"🔄 LoggingMiddleware.awrap_tool_call")
        print(f"⚙️  도구 실행: {tool_name}")
        print("-"*60)
        print("\nTool Calling\n", end="", flush=True)
        
        result = await handler(request)
        
        print("\n" + "-"*60)
        print(f"🔄 LoggingMiddleware.awrap_tool_call (완료)")
        print(f"✅ 도구 완료: {tool_name}")
        if isinstance(result, ToolMessage) and result.content:
            preview = result.content[:100] + "..." if len(result.content) > 100 else result.content
            print(f"   결과: {preview}")

        print("-"*60)
        
        return result

class DynamicModelMiddleware(AgentMiddleware):
    """대화 길이에 따라 모델 변경"""
    def awrap_model_call(
        self,
        request: ModelRequest,
        handler: Callable[[ModelRequest], ModelResponse],
    ) -> ModelResponse:
        CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY")
        # 대화 길이에 따라 다른 모델 사용
        model_name = "HCX-007"
        new_model = ChatClovaX(model=model_name, reasoning_effort="none", api_key=CLOVA_STUDIO_API_KEY)
        if len(request.messages) > 5:
            request.model = new_model
            print("\n" + "-"*60)
            print(f"🔄 DynamicModelMiddleware:  Controller Using {model_name} for long conversation")
            print("-"*60)

        return handler(request)


class NaverToTavilyFallbackMiddleware(AgentMiddleware):
    """네이버 검색 실패 시 Tavily로 폴백"""
    async def awrap_tool_call(
        self,
        request: ToolCallRequest,
        handler: Callable[[ToolCallRequest], ToolMessage | Command],
    ) -> ToolMessage | Command:
        tool_name = request.tool_call.get('name', 'unknown')
        
        if tool_name != 'naver_web_search':
            return await handler(request)
        
        try:
            result = await handler(request) # 네이버 검색 실행
            print("\n" + "-"*60)
            print("🔄 NaverToTavilyFallbackMiddleware: ✅ 네이버 검색 성공")
            print("-"*60)
            return result
            
        except Exception as e:
            print("\n" + "-"*60)
            print(f"🔄 NaverToTavilyFallbackMiddleware: ⚠️  네이버 검색 실패 → Tavily 검색 전환: {str(e)[:50]}...")
            print("-"*60)

            args = request.tool_call.get('args', {})
            query = args.get('query', '')
            display = args.get('display', 5)
            
            # ainvoke 사용 (딕셔너리로 전달)
            tavily_result = await tavily_web_search.ainvoke({
                "query": query, 
                "display": display
            })
            
            return ToolMessage(
                content=f"[Tavily 검색 결과]\n{tavily_result}",
                tool_call_id=request.tool_call.get('id', '')
            )
        

@dynamic_prompt
def writing_format(request: ModelRequest) -> str:
    """사용자 요청에 알맞는 시스템 프롬프트 생성"""
    content_type = request.runtime.context.get("content_type", "report")
    base_prompt = "당신은 글쓰기 어시스턴트 입니다. 다음 형식으로 글을 작성하세요."
    
    print("\n" + "-"*60)
    print(f"🔄 writing_format: {content_type} 형식으로 작성")
    print("-"*60)

    if content_type == "blog":
        return f"{base_prompt}\n\n{WRITING_PROMPTS['blog']}"
    elif content_type == "report":
        return f"{base_prompt}\n\n{WRITING_PROMPTS['report']}"

    return base_prompt

 

관리자 에이전트(Controller Agent) 구성하기

관리자 에이전트는 사용자의 요청을 받아 적절한 전문가 에이전트를 호출하여 작업을 조율하는 중앙 관제 역할을 수행합니다. 이를 위해 HCX-007 모델을 사용하며, 구체적인 역할은 다음과 같습니다.

• 사용자 요청을 분석하고 필요한 전문가 에이전트를 선택합니다.
• 여러 전문가 에이전트를 순차적으로 호출하여 복합 작업을 수행합니다.
• Human-In-the-Loop을 통해 중요한 작업 전 사용자 승인을 받습니다.
• 최종 결과를 사용자에게 전달합니다.

# agent.py

import os
import asyncio
from langchain.agents import create_agent
from langchain.messages import SystemMessage, HumanMessage
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain_naver import ChatClovaX
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command

from utils.tool_agents import call_write_agent, call_web_search_agent, call_save_agent
from utils.custom_middleware import ReportAgentState, DynamicModelMiddleware, LoggingMiddleware
from utils.prompts import CONTROLLER_PROMPT

from dotenv import load_dotenv

load_dotenv()

# LangGraph Studio용 graph 생성 함수
def build_graph():
    """LangGraph 서버에서 호출하는 graph 빌더"""

    CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY")

    model = ChatClovaX(model="HCX-005", api_key=CLOVA_STUDIO_API_KEY)

    agent = create_agent(
        model=model,
        tools=[call_write_agent, call_web_search_agent, call_save_agent],
        checkpointer=InMemorySaver(),
        middleware=[
            # LoggingMiddleware(),
            DynamicModelMiddleware(),
            HumanInTheLoopMiddleware(
                interrupt_on={
                    "call_web_search_agent": {"allowed_decisions": ["approve", "reject"]},
                    "call_write_agent": False,
                    "call_save_agent": False
                }
            )
        ],
        state_schema=ReportAgentState,
        system_prompt=CONTROLLER_PROMPT
    )
    return agent


async def main():
    agent = build_graph()
    config = {"configurable": {"thread_id": "asd123"}} 

    print("Multi Agent System Created!\n")

    while True:
        user_input = input("\nUser: ")
        if user_input.lower() in ["종료", "exit"]:
            print("AI: 대화를 종료합니다. 이용해주셔서 감사합니다.")
            break

        try:
            # 첫 실행
            result = await agent.ainvoke(
                {"messages": [HumanMessage(user_input)]},
                config=config
            )
            
            # interrupt 확인 및 처리
            while "__interrupt__" in result:
                print("\n" + "="*60)
                print("⏸️  승인이 필요한 작업이 있습니다")
                
                print("="*60)
                
                interrupt_data = result["__interrupt__"][0].value
                action_requests = interrupt_data.get("action_requests", [])
                
                print(f"\n📋 총 {len(action_requests)}개의 작업 대기 중\n")
                
                decisions = []
                
                for i, action in enumerate(action_requests, 1):
                    tool_name = action.get("name", "unknown")
                    tool_args = action.get("args", {})
                    
                    print(f"작업 {i}:")
                    print(f"  🔧 도구: {tool_name}")
                    print(f"  📝 인자: {tool_args}")
                    
                    decision = input(f"\n\n승인하시겠습니까? (approve/reject): ").strip().lower()
                    
                    if decision == "approve":
                        decisions.append({"type": "approve"})
                        print("✅ 승인됨\n")
                    else:
                        # approve가 아니면 모두 reject
                        decisions.append({"type": "reject"})
                        print("❌ 거부됨\n")
                
                # 결정 전달 및 재실행
                print("="*60)
                print("🔄 작업 재개 중...")
                print("="*60 + "\n")
                
                result = await agent.ainvoke(
                    Command(resume={"decisions": decisions}),
                    config
                )
            
            # 최종 결과 출력
            print("\nAI: ", end="", flush=True)
            final_message = result["messages"][-1]
            if hasattr(final_message, 'content'):
                print(final_message.content)
            else:
                print(final_message)
            print()

        except Exception as e:
            print(f"\n❌ 오류 발생: {e}")
            import traceback
            traceback.print_exc()


if __name__ == "__main__":
    asyncio.run(main())

 

Quote

주의 사항

• thread_id는 대화 세션을 구분하는 고유 ID입니다. 실제 환경에서는 사용자별로 다른 ID를 사용하세요
• interrupt 처리 중 사용자가 'reject'하면 해당 도구는 실행되지 않고 다음 단계로 넘어갑니다
• interrupt 처리 방식 중 'edit'도 있지만, 작성하는 시점에 interrupt가 두 번 발생하는 버그가 있어 사용을 지양합니다(참고).
• HCX-007 모델의 경우 tool calling과 추론을 병행 할 수 없습니다. 따라서, tool calling이 필요한 경우 추론 파라미터 resoning_effort='none'을 설정합니다(기본값 'low').

 

전문가 에이전트(Tool Agents)구성하기

각 전문가 에이전트는 @tool 데코레이터로 래핑되어 관리자 에이전트가 호출할 수 있는 함수 형태가 됩니다. 작업 완료 후 Command 객체 반환을 통해 State를 업데이트하여 결과를 관리자 에이전트에게 전달합니다. 이번 프로젝트에서는 HCX-005 모델로 전문가 에이전트를 구성했습니다.

전문가 에이전트의 역할은 다음과 같습니다.

• @tool 데코레이터로 일반 함수를 LangChain 도구로 변환합니다.
• 각 Agent는 독립적인 create_agent로 생성되어 고유한 미들웨어와 도구를 가집니다.
• Command 객체를 통해 메시지와 상태를 업데이트하여 관리자 에이전트에 전달합니다.

# tool_agents.py

import os
from typing import Annotated, Literal
from langgraph.types import Command
from langchain.tools import tool
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langchain.messages import HumanMessage, ToolMessage
from langchain.tools import InjectedToolCallId
from langchain_naver import ChatClovaX

from .tools import naver_web_search, save_to_file, save_to_notion
from .custom_middleware import writing_format, NaverToTavilyFallbackMiddleware
from .prompts import WEB_SEARCH_PROMPT, SAVE_PROMPT

from dotenv import load_dotenv

load_dotenv()

CLOVA_STUDIO_API_KEY = os.getenv("CLOVA_STUDIO_API_KEY")

write_model = ChatClovaX(model="HCX-005", api_key=CLOVA_STUDIO_API_KEY)
web_search_model = ChatClovaX(model="HCX-005", api_key=CLOVA_STUDIO_API_KEY)
save_model = ChatClovaX(model="HCX-DASH-002", api_key=CLOVA_STUDIO_API_KEY)

write_agent = create_agent(
    model=write_model,
    middleware=[writing_format]
)

web_search_agent = create_agent(
    model=web_search_model,
    tools=[naver_web_search],
    middleware=[NaverToTavilyFallbackMiddleware()],
    system_prompt=WEB_SEARCH_PROMPT
)

save_agent = create_agent(
    model=save_model,
    tools=[save_to_notion, save_to_file],
    middleware=[HumanInTheLoopMiddleware(
        interrupt_on={
            "save_to_notion":{"allowed_decisions": ["approve", "reject"]},
            "save_to_file":{"allowed_decisions": ["approve", "reject"]}
        }
    )],
    system_prompt=SAVE_PROMPT
)


@tool
async def call_write_agent(
    article: str, 
    content_type: Literal["report", "blog"],
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
    """
    글쓰기 서브에이전트를 호출하여 지정된 형식의 글을 작성합니다.
    
    Args:
        article: 작성할 주제나 원본 내용
        content_type: 글 형식 타입 ('report'|'blog')
        tool_call_id: LLM 도구 호출 ID (자동 주입)
    
    Returns:
        Command: 작성된 글과 업데이트된 상태를 포함하는 Command 객체
            - messages: 작성된 글이 담긴 ToolMessage
    """
    result = await write_agent.ainvoke(
        {"messages": [HumanMessage(article)]},
        # context에 넣어서 middleware에 전달
        context={
            "content_type": content_type
        },
    )

    return Command(update={
        "messages": [
            ToolMessage(
                content=result["messages"][-1].content,
                tool_call_id=tool_call_id
            )
        ],
        "content_type": content_type
    })

@tool
async def call_web_search_agent(
    query: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
    """
    사용자가 검색을 요청하면 웹 검색 서브에이전트를 호출하여 웹 검색을 수행합니다
    
    Args:
        query: 검색에 사용할 쿼리
        runtime: 메인 에이전트의 상태 접근을 위한 런타임 객체
        tool_call_id: LLM 도구 호출 ID (자동 주입)
    
    Returns:
        Command: 작성된 글과 업데이트된 상태를 포함하는 Command 객체
            - messages: 작성된 글이 담긴 ToolMessage
    """
    result = await web_search_agent.ainvoke(
        {"messages": [HumanMessage(query)]},
        context={"current_agent": "web_search_agent"}
    )

    return Command(update={
        "messages": [
            ToolMessage(
                content=result["messages"][-1].content,
                tool_call_id=tool_call_id
            )
        ]
    })


@tool
async def call_save_agent(
    content: str,
    destination: Literal["file", "notion"],
    filename_or_title: str,
    tool_call_id: Annotated[str, InjectedToolCallId],
) -> Command:
    """
    콘텐츠를 파일 또는 Notion에 저장합니다.
    
    Args:
        content: 저장할 콘텐츠
        destination: 저장 위치 ('file' 또는 'notion')
        filename_or_title: 파일명(file) 또는 노션 페이지 제목(notion)
        tool_call_id: LLM 도구 호출 ID (자동 주입)
    
    Returns:
        Command: 저장 결과를 포함하는 Command 객체
    """
    # destination에 따라 다른 메시지 전달
    if destination == "notion":
        context_msg = f"다음 내용을 Notion에 '{filename_or_title}' 제목으로 저장해주세요:\n\n{content}"
    else:  # file
        context_msg = f"다음 내용을 '{filename_or_title}' 파일로 저장해주세요:\n\n{content}"

    result = await save_agent.ainvoke(
        {"messages": [HumanMessage(context_msg)]})
    
    return Command(update={
        "messages": [
            ToolMessage(
                content=result["messages"][-1].content,
                tool_call_id=tool_call_id
            )
        ]
    })

 

Quote

주의 사항

• @tool 데코레이터는 함수의 docstring을 자동으로 도구 설명으로 사용합니다. 도구 호출을 위해서 명확하게 작성하세요.
• Literal 타입을 사용하여 입력값을 제한하면 모델이 잘못된 값을 전달하는 것을 방지할 수 있습니다.
• Command 객체에에 전달해야 커스텀 State의 업데이트가 가능합니다.
• tool_call_id를 포함하여 도구 호출을 추적할 수 있습니다 .

 

도구 구현하기

전문가 에이전트인 웹 검색 에이전트와 저장 에이전트가 사용하는 도구를 구현합니다. 이 도구들은 멀티 에이전트와 미들웨어의 활용 방법을 보여주기 위한 예시로 구성되었습니다.

웹 검색 에이전트의 경우 naver_web_search 도구를 가지고 있고, NaverToTavilyFallbackMiddleware에 의해 tavily_web_search 도구 또한 활용할 수 있습니다. 저장 에이전트는 개인 노션에 업로드 할 수 있는 save_to_notion, 시스템에 저장할 수 있는 save_to_file 두 가지 도구를 가지고 있습니다.

# tools.py

import os
import re
import httpx
from pathlib import Path
from langchain.tools import tool
from tavily import TavilyClient
from notion_client import Client

from dotenv import load_dotenv

# 환경 변수 로드
load_dotenv()

NAVER_CLIENT_ID = os.getenv("NAVER_CLIENT_ID")
NAVER_CLIENT_SECRET = os.getenv("NAVER_CLIENT_SECRET")
TAVILY_API_KEY = os.getenv("TAVILY_API_KEY")


@tool
async def naver_web_search(query: str, display: int = 10) -> dict:
    """
    네이버 검색 API를 호출해 결과를 구조화하여 반환합니다.

    Args:
        query: 검색어
        display: 검색 결과 수(1~100)

    Returns:
        {
          "query": str,
          "total": int,
          "items": [{"title": str, "link": str, "description": str}]}
        }
    """
    url = "https://openapi.naver.com/v1/search/webkr.json"
    params = {"query": query, "display": display}
    headers = {
        "X-Naver-Client-Id": NAVER_CLIENT_ID,
        "X-Naver-Client-Secret": NAVER_CLIENT_SECRET,
    }

    async with httpx.AsyncClient() as client:
        r = await client.get(url, headers=headers, params=params)
        r.raise_for_status()
        data = r.json()

    results = []
    for item in data.get("items", []):
        results.append({
            "title": re.sub(r"<.*?>", "", item.get("title") or "").strip(),
            "link": item.get("link"),
            "description": re.sub(r"<.*?>", "", item.get("description") or "").strip(),
        })

    return {
        "query": query,
        "total": data.get("total", 0),
        "items": results,
    }


# 본 쿡북에서는 해당 도구를 NaverToTavilyFallbackMiddleware에서 풀백 도구로 활용하였습니다
@tool
async def tavily_web_search(query: str, display: int = 5) -> str:
    """
    Tavily API를 사용하여 웹 검색을 수행합니다.
    
    Args:
        query: 검색어
        display: 검색 결과 수(1~100)
    """
    try:
        tavily_client = TavilyClient(api_key=TAVILY_API_KEY)

        # Tavily는 동기 라이브러리이므로, 비동기 처리를 위해 asyncio.to_thread 사용
        import asyncio
        response = await asyncio.to_thread(
            tavily_client.search, 
            query=query, 
            max_results=display
        )

        # 결과 포맷팅
        results = []
        for idx, result in enumerate(response.get('results', []), 1):
            result_text = f"\n{idx}. {result.get('title', 'No title')}\n"
            result_text += f"   URL: {result.get('url', 'No URL')}\n"
            result_text += f"   내용: {result.get('content', 'No content')}\n"
            results.append(result_text)
        
        if not results:
            return "검색 결과가 없습니다."
        
        return "".join(results)
        
    except Exception as e:
        return f"검색 중 오류 발생: {str(e)}"
    

@tool
def save_to_file(filename: str, content: str) -> str:
    """
    지정된 파일 이름으로 리포트 내용을 시스템에 마크다운 파일로 저장합니다.
    
    Args:
        filename: 저장할 파일 이름 (확장자 없으면 자동으로 .md 추가)
        content: 파일에 저장할 텍스트 내용
    
    Returns:
        저장 성공/실패 메시지와 파일 경로
    """
    try:
        # 확장자 없으면 자동으로 .md 추가
        if not filename.endswith(('.md', '.markdown', '.txt')):
            filename = f"{filename}.md"
        
        # 저장 디렉토리 생성 (없으면)
        save_dir = Path("reports")  # 또는 원하는 디렉토리
        save_dir.mkdir(exist_ok=True)
        
        # 전체 경로
        file_path = save_dir / filename
        
        # 마크다운 형식으로 저장
        with open(file_path, "w", encoding="utf-8") as f:
            f.write(content)
        
        # 절대 경로 반환
        abs_path = file_path.resolve()
        return f"✅ 마크다운 파일로 저장 완료!\n📁 경로: {abs_path}"
        
    except Exception as e:
        return f"❌ 파일 저장 중 오류 발생: {e}"
    

@tool
def save_to_notion(page_title: str, content: str) -> str:
    """
    Notion 데이터베이스에 새 페이지를 생성하고 콘텐츠를 저장합니다.
    
    사용자가 작성한 글, 보고서, 메모 등을 Notion에 저장할 때 사용합니다.
    페이지 제목과 본문 내용을 받아서 지정된 Notion 데이터베이스에 자동으로 추가합니다.
    
    Args:
        page_title: Notion 페이지의 제목 (예: "주간 보고서", "회의 내용")
        content: 페이지 본문에 저장할 텍스트 내용 (마크다운 형식 지원)
    
    Returns:
        성공 시: 생성된 페이지 제목과 URL
        실패 시: 오류 메시지
    
    Examples:
        - "이 리포트를 Notion에 '월간 분석'이라는 제목으로 저장해줘"
        - "방금 작성한 글을 Notion에 저장"
    """
    try:
        notion = Client(
            auth=os.getenv("NOTION_API_KEY"),
            notion_version="2025-09-03"  # 최신 버전
        )
        
        data_source_id = os.getenv("NOTION_DATA_SOURCE_ID")
        
        # data_source_id 사용
        new_page = notion.pages.create(
            parent={
                "type": "data_source_id",
                "data_source_id": data_source_id
            },
            properties={
                "title": {  # 기본 title
                    "title": [
                        {"text": {"content": page_title}}
                    ]
                }
            },
            children=[
                {
                    "object": "block",
                    "type": "paragraph",
                    "paragraph": {
                        "rich_text": [
                            {"text": {"content": content}}
                        ]
                    }
                }
            ]
        )
        
        return f"✅ '{page_title}' 생성 완료: {new_page['url']}"
        
    except Exception as e:
        return f"❌ 오류: {str(e)}"

 

Quote

주의 사항

• 이 도구들은 테스트 목적으로 제작되었습니다.
• 실제 NaverToTavilyFallbackMiddleware에 의한 풀백 메커니즘을 테스트 해보시려면 naver_web_search 도구 내부 로직을 실패하도록 수정해야 합니다.

 

프롬프트

프로젝트에서 사용하는 모든 프롬프트를 정의합니다. 각 에이전트의 역할과 동작 방식을 명확히 지정하여 일관된 응답을 보장합니다. 사용된 프롬프트는 아래 다운로드 링크에서 확인할 수 있습니다. 해당 내용을 복사해 prompt.py 파일로 저장하세요.

prompts.txt 다운로드 

 

실행 및 동작 예시

다음은 HumanInTheLoopMiddleware가 적용된 멀티 에이전트 시스템의 동작 흐름입니다. 각 에이전트의 도구 호출 과정에서 사용자 승인 단계가 개입되며, 전체 응답 흐름은 다음과 같습니다.

에이전트 실행 결과는 다음과 같습니다.

User: 이번 네이버 DAN25에서 발표된 네이버의 AI 전략에 대해 찾아보고 리포트로 작성해서 노션에 업로드 해줘

============================================================
⏸️  승인이 필요한 작업이 있습니다
============================================================

📋 총 1개의 작업 대기 중

작업 1:
  🔧 도구: call_web_search_agent
  📝 인자: {'query': '네이버 DAN25에서 발표된 네이버의 AI 전략'}


승인하시겠습니까? (approve/reject): approve
✅ 승인됨

============================================================
🔄 작업 재개 중...
============================================================


------------------------------------------------------------
🔄 NaverToTavilyFallbackMiddleware: ✅ 네이버 검색 성공
------------------------------------------------------------

------------------------------------------------------------
🔄 writing_format: report 형식으로 작성
------------------------------------------------------------

============================================================
⏸️  승인이 필요한 작업이 있습니다
============================================================

📋 총 1개의 작업 대기 중

작업 1:
  🔧 도구: save_to_notion
  📝 인자: {'page_title': '네이버 AI 전략 DAN25 발표 분석', 'content': '# 네이버의 AI 전략: DAN25 발표 내용 분석\n\n## 1. 개요\n본 리포트는 네이버의 DAN25 발표 내용을 바탕으로 한 AI 전략에 대해 분석합니다. 네이버는 AI 에이전트 도입 확대, 핵심 제조 산업 경쟁력 강화, 새로운 AI 도구 및 플랫폼 전략 공개, AI 산업 거품론 대응, 그리고 미래 비전과 글로벌 확장 계획을 통해 AI 기술의 발전과 실질적 가치 창출을 목표로 하고 있습니다.\n\n## 2. 핵심 발견사항\n- **AI 에이전트 도입 확대**: 네이버는 주요 서비스에 AI 에이전트를 도입해 개인화된 사용자 경험 제공.\n- **핵심 제조 산업 경쟁력 강화**: 반도체, 자동차, 조선 등 제조 산업에서의 AI 활용 방안 모색.\n- **신규 AI 도구와 플랫폼 전략**: 산업별 버티컬 AI와 경량화 모델을 통한 실질적인 가치 창출 목표.\n- **산업 거품론 대응**: 경량화 모델과 산업 특화 AI를 통해 실질적인 가치 창출 중요성 강조.\n- **글로벌 확장 계획**: 차세대 AI 전략 발표 및 글로벌 시장 진출 도모.\n\n## 3. 분석 및 인사이트\n네이버는 AI 기술의 전방위적 도입을 통해 사용자 맞춤형 서비스 제공을 강화하고 있으며, 제조업 분야에서도 AI 트랜스포메이션을 추진 중입니다. 또한, DAN25에서는 산업별 맞춤형 AI 솔루션과 경량화된 모델을 통해 실질적인 성과를 도출하고자 했습니다. AI 산업 내 거품론을 경계하며 실체 있는 기술 개발에 주력하고 있고, 글로벌 확장을 위한 미래 비전을 제시하고 있습니다.\n\n## 4. 결론 및 제언\n네이버의 AI 전략은 다각도로 전개되고 있으며, 이는 궁극적으로 사용자 경험 개선과 산업 전반의 혁신을 촉진할 것입니다. 향후 네이버는 AI 기술의 고도화와 더불어 글로벌 시장에서의 입지 강화를 위해 지속적인 투자와 연구 개발이 필요합니다.\n\n## 5. 참고 자료\n[AI 에이전트 도입 확대](https://www.etnews.com/20251023000297), [핵심 제조 산업 경쟁력 강화](http://www.efnews.co.kr/news/articleView.html?idxno=124617), [신규 AI 도구와 플랫폼 전략](https://www.asiatoday.co.kr/kn/view.php?key=20250930010016279), [AI 산업 거품론 대응](https://www.econovill.com/news/articleView.html?idxno=717550), [미래 비전과 글로벌 확장 계획](https://www.kmjournal.net/news/articleView.html?idxno=4023)'}


승인하시겠습니까? (approve/reject): approve
✅ 승인됨

============================================================
🔄 작업 재개 중...
============================================================


------------------------------------------------------------
🔄 DynamicModelMiddleware:  Controller Using HCX-007 for long conversation
------------------------------------------------------------

AI: 리포트가 성공적으로 노션에 저장되었습니다!

### 세부 사항
- **제목:** 네이버 AI 전략 DAN25 발표 분석
- **노션 링크:** [네이버 AI 전략 DAN25 발표 분석](https://www.notion.so/AI-DAN25-2bf87d6d35378107a508c5b0bc8f478a)

추가로 필요하신 것이 있으면 말씀해 주세요!

 

LangSmith Studio

LangSmith Studio는 AI 에이전트 개발을 위한 전용 IDE(통합 개발 환경)입니다. 그래프 기반 시각화 인터페이스로 에이전트가 실행되는 동안 각 노드의 전환과 상태 변화를 실시간으로 추적할 수 있어, 복잡한 로직의 흐름을 한눈에 파악할 수 있습니다. 또한 프롬프트를 수정하면 즉시 반영되는 hot-reloading 기능으로 빠른 반복 개발이 가능하고, 멀티턴 대화를 테스트할 수 있는 Chat UI가 내장되어 있습니다.

LangSmith API

에이전트의 동작 과정을 모니터링하고, LangSmith Studio를 사용하기 위해 LangSmith의 API 키를 발급받아야 합니다.

• LangSmith 접속 > 로그인 > 좌측 사이드바 'Settings' > API Keys 탭 > + API key

 

langgraph.json

LangGraph 애플리케이션의 구성 정보를 담고 있는 설정 파일입니다. 그래프의 위치, 의존성, 환경 변수 등 에이전트의 구조를 정의합니다. 자세한 설정은 공식 문서를 참고하세요.

{
  "dependencies": ["."],
  "graphs": {
    "controller": "./agent.py:build_graph"
  },
  "env": ".env"
}

 

LangSmith Studio 활성화

다음 명령어를 통해 LangSmith Studio를 활성화합니다. 다음 명령어를 실행하면 다음 과정이 순차적으로 수행됩니다.

langgraph dev

• langgraph-cli가 langgraph.json 파일을 읽습니다.
• 로컬 API 서버가 시작됩니다.
• LangSmith Studio의 웹 UI가 이 서버에 자동으로 연결됩니다. 다음 URL을 통해 LangSmith Studio에 접근할 수 있습니다.
  • https://smith.langchain.com/studio/?baseUrl=http://127.0.0.1:2024

LangSmith Studio에서는 내장된 Chat UI로 직접 구축한 에이전트와 대화하며 실시간으로 테스트할 수 있고, 토큰 소모량, 실행 시간, 각 컴포넌트의 입출력 등을 모니터링하여 디버깅과 최적화를 진행할 수 있습니다. 또한, 프롬프트를 수정하면 즉시 반영되어 다양한 프롬프트 변형을 빠르게 실험할 수 있으며, 그래프 시각화를 통해 에이전트의 실행 흐름을 직관적으로 파악할 수 있습니다.

Quote

LangSmith Studio에서의 Human-In-the-Loop

LangSmith Studio에서도 HumanInTheLoopMiddleware에 의한 interrupt가 발생합니다. agent.py에서는 내부 로직에서 직접 interrupt에 대한 처리를 진행했지만, LangSmith Studio에서는 interrupt가 발생하면 Command 객체의 resume 파라미터에 들어갈 적절한 형태의 값을 전달해야 합니다.

승인 시 입력값

{
  "decisions": [
    {
      "type": "approve"
    }
  ]
}

거부 시 입력값

{
  "decisions": [
    {
      "type": "reject"
    }
  ]
}

 

마무리

이번 쿡북을 통해 LangChain v1.0이 제시하는 새로운 멀티 에이전트 구축 방법을 익혔습니다.

특히, create_agent와 미들웨어를 적극 활용함으로써 로깅, 에러 처리, Human-In-the-Loop 같은 핵심 기능을 모듈화할 수 있었고, 복잡한 그래프 구축 과정 없이도 유지보수가 용이한 에이전트 설계가 가능해졌습니다. 또한, HyperCLOVA X를 기반으로 관리자 에이전트가 하위 전문가 에이전트들을 정교하게 조율하는 협업 구조를 구현할 수 있었습니다.

이제 이 가이드를 발판 삼아, LangChain v1.0의 유연한 구조 위에 CLOVA Studio의 강력한 모델들을 더해 여러분만의 창의적인 에이전트 서비스를 완성해보시기 바랍니다.