LangFlow 완전 가이드 4편 — 심화 튜토리얼 2 : 멀티 에이전트, MCP, 프로덕션 배포

3편에서 에이전트, 메모리, 커스텀 컴포넌트를 다뤘어요.

이번 편은 시리즈 마지막이에요.

이번 편에서 다루는 것:
1. 멀티 에이전트 파이프라인
2. MCP 서버 연결 (클라이언트)
3. LangFlow를 MCP 서버로 노출 (Claude Desktop 연결)
4. 프로덕션 배포 (Docker + PostgreSQL)
5. 모니터링과 트레이싱

 

---

실전 1 — 멀티 에이전트 파이프라인

에이전트가 다른 에이전트를 툴로 써요. 복잡한 작업을 역할별로 분담해요.

구조 설계

[사용자 질문]
      ↓
[총괄 에이전트] ── 판단
      ├── 리서치 필요? ──→ [리서치 에이전트] (웹 검색 전문)
      ├── 코드 필요?   ──→ [코딩 에이전트]   (코딩 전문)
      └── 직접 답변 가능 ──→ 바로 응답

컴포넌트 배치

1. Chat Input
2. Agent (총괄)          ← 오케스트레이터
3. Agent (리서치 전문)   ← Tool Mode 활성화
4. Agent (코딩 전문)     ← Tool Mode 활성화
5. Chat Output

리서치 에이전트 설정

Model: claude-sonnet-4-6
Instructions:
"당신은 웹 리서치 전문가입니다.
주어진 주제를 웹에서 검색해서 정확하고 최신 정보를 제공합니다.
반드시 검색 툴을 사용하고, 출처를 명시하세요."

Tools: DuckDuckGo Search, URL Tool

Tool Mode: ✅ 활성화  ← 이걸 켜야 다른 에이전트가 툴로 쓸 수 있음

Tool Mode 활성화:

리서치 에이전트 컴포넌트 클릭
→ 상단 헤더 메뉴 → "Tool Mode" 클릭
→ 컴포넌트에 도구 아이콘 생김

코딩 에이전트 설정

Model: claude-opus-4-7  ← 코딩은 제일 좋은 모델로
Instructions:
"당신은 시니어 소프트웨어 엔지니어입니다.
코드 작성, 디버깅, 코드 리뷰를 전문으로 합니다.
항상 실행 가능한 코드를 작성하고 타입 힌트를 포함하세요."

Tool Mode: ✅ 활성화

총괄 에이전트 설정

Model: claude-sonnet-4-6
Instructions:
"당신은 팀 리더 AI입니다.
사용자 요청을 분석하고 적절한 전문가 에이전트에게 위임하세요.

사용 가능한 에이전트:
- 리서치 에이전트: 최신 정보 검색, 뉴스, 데이터 조사
- 코딩 에이전트: 코드 작성, 디버깅, 기술 질문

작업을 적절히 분배하고 결과를 종합해서 답변하세요."

Tools:
→ 리서치 에이전트 (Tool Mode 활성화된 것)
→ 코딩 에이전트 (Tool Mode 활성화된 것)

연결

[Chat Input] ──→ [총괄 Agent] ──→ [Chat Output]
                      ↑
          [리서치 Agent (Tool Mode)]
          [코딩 Agent (Tool Mode)]

테스트

"2026년 최신 Python 비동기 프레임워크 트렌드 조사하고
FastAPI vs Litestar 코드 비교 예시 만들어줘"

→ 총괄 에이전트 판단:
  1. 리서치 에이전트에게 트렌드 조사 위임
  2. 코딩 에이전트에게 코드 비교 위임
  3. 두 결과 합쳐서 최종 답변

---

실전 2 — MCP 서버 연결 (클라이언트)

LangFlow 에이전트가 외부 MCP 서버의 툴을 쓰는 방법이에요.

GitHub MCP 서버 연결

GitHub 이슈, PR, 레포를 에이전트가 직접 조회해요.

Node.js 설치 먼저:

# Node.js 없으면 설치 필요 (MCP 서버 실행에 필요)
node --version  # 확인

컴포넌트 배치:

1. Chat Input
2. MCP Tools    ← MCP 서버 연결 담당
3. Agent
4. Chat Output

MCP Tools 설정:

컴포넌트 클릭 → "Add MCP Server"
→ STDIO 모드 선택

Command: npx
Arguments:
  - -y
  - @modelcontextprotocol/server-github

Environment Variables:
  GITHUB_PERSONAL_ACCESS_TOKEN: ghp_xxxxx

Tool Mode 활성화 후 Agent에 연결:

MCP Tools → Tool Mode 활성화
→ Agent의 Tools 포트에 연결

Agent Instructions:

당신은 GitHub 레포지토리 관리 어시스턴트입니다.
GitHub MCP 툴을 사용해서 이슈, PR, 코드를 조회하고 관리합니다.

테스트:

"langflow-ai/langflow 레포에서 최근 열린 이슈 5개 보여줘"
"내 레포에서 머지 안 된 PR 목록 알려줘"

다른 MCP 서버 추가

# Slack
npx -y @modelcontextprotocol/server-slack

# 파일시스템
npx -y @modelcontextprotocol/server-filesystem /path/to/dir

# PostgreSQL
npx -y @modelcontextprotocol/server-postgres postgresql://user:pass@host/db

각각 MCP Tools 컴포넌트로 추가하고 Agent에 연결하면 돼요.

---

실전 3 — LangFlow를 MCP 서버로 노출

내가 만든 LangFlow 플로우를 Claude Desktop이나 Cursor에서 직접 쓸 수 있어요.

플로우를 MCP 툴로 설정

플로우 우상단 "Share" 버튼
→ "MCP Server" 탭 클릭
→ "Edit Tools" 클릭

각 플로우에서:
Tool Name:        rag_document_search
Tool Description: 업로드된 회사 문서에서 정보를 검색합니다.
                  제품 정책, 계약 조건, 내부 가이드라인 등을 조회하세요.

툴 이름과 설명이 명확할수록 Claude가 언제 이 툴을 써야 할지 정확히 판단해요.

Claude Desktop에 연결

Share → MCP Server → "Claude" 탭
→ "Auto Install" 클릭
→ Claude Desktop 자동 설정

또는 수동으로:

Claude Desktop 설정 파일 열기:

# Mac
open ~/Library/Application\ Support/Claude/claude_desktop_config.json

# Windows
notepad %APPDATA%\Claude\claude_desktop_config.json

설정 추가:

{
  "mcpServers": {
    "langflow": {
      "command": "npx",
      "args": [
        "mcp-remote",
        "http://localhost:7860/api/v1/mcp/streamable"
      ],
      "env": {
        "API_KEY": "your-langflow-api-key"
      }
    }
  }
}

Claude Desktop 재시작 후:

Claude Desktop에서:
"회사 환불 정책이 어떻게 돼?"
→ LangFlow RAG 플로우 자동 실행
→ 결과 반환

Cursor에 연결

Cursor → Settings → MCP
→ + Add Server
→ URL: http://localhost:7860/api/v1/mcp/streamable
→ API Key: your-langflow-api-key

이제 Cursor에서 코딩하다가:

"우리 API 문서에서 인증 엔드포인트 스펙 찾아줘"
→ LangFlow RAG 플로우 실행
→ 내부 문서에서 검색 후 결과 반환

---

실전 4 — 프로덕션 배포

Docker + PostgreSQL 배포

로컬 SQLite 대신 PostgreSQL을 쓰면 다중 사용자, 고가용성 환경에서도 안정적으로 동작해요.

docker-compose.yml:

version: "3.8"

services:
  langflow:
    image: langflowai/langflow:latest
    ports:
      - "7860:7860"
    environment:
      # 데이터베이스
      - LANGFLOW_DATABASE_URL=postgresql://langflow:password@postgres:5432/langflow

      # 인증
      - LANGFLOW_SECRET_KEY=your-secret-key-here-change-this
      - LANGFLOW_AUTO_LOGIN=false

      # 성능
      - LANGFLOW_WORKERS=2
      - LANGFLOW_MAX_FILE_SIZE_UPLOAD=1024

      # API 키 (옵션: 환경변수로 관리)
      - ANTHROPIC_API_KEY=${ANTHROPIC_API_KEY}
      - OPENAI_API_KEY=${OPENAI_API_KEY}
    depends_on:
      postgres:
        condition: service_healthy
    volumes:
      - langflow_data:/app/langflow
    restart: unless-stopped

  postgres:
    image: postgres:16-alpine
    environment:
      - POSTGRES_USER=langflow
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=langflow
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U langflow"]
      interval: 10s
      timeout: 5s
      retries: 5
    restart: unless-stopped

volumes:
  langflow_data:
  postgres_data:

.env 파일:

ANTHROPIC_API_KEY=sk-ant-xxxxx
OPENAI_API_KEY=sk-xxxxx

실행:

docker-compose up -d

# 로그 확인
docker-compose logs -f langflow

# 상태 확인
docker-compose ps

Nginx 리버스 프록시 (HTTPS 적용)

# /etc/nginx/sites-available/langflow
server {
    listen 80;
    server_name langflow.mycompany.com;
    return 301 https://$host$request_uri;
}

server {
    listen 443 ssl;
    server_name langflow.mycompany.com;

    ssl_certificate /etc/letsencrypt/live/langflow.mycompany.com/fullchain.pem;
    ssl_certificate_key /etc/letsencrypt/live/langflow.mycompany.com/privkey.pem;

    location / {
        proxy_pass http://localhost:7860;
        proxy_set_header Host $host;
        proxy_set_header X-Real-IP $remote_addr;

        # 스트리밍 응답을 위한 설정
        proxy_buffering off;
        proxy_read_timeout 300s;
        proxy_send_timeout 300s;
    }
}

---

실전 5 — 모니터링과 트레이싱

LangSmith 연결

# .env에 추가
LANGCHAIN_TRACING_V2=true
LANGCHAIN_API_KEY=ls__xxxxx
LANGCHAIN_PROJECT=my-langflow-project

docker-compose.yml에도 추가:

environment:
  - LANGCHAIN_TRACING_V2=true
  - LANGCHAIN_API_KEY=${LANGCHAIN_API_KEY}
  - LANGCHAIN_PROJECT=production

이후 smith.langchain.com에서:

모든 플로우 실행 추적
각 컴포넌트 소요 시간
LLM 호출 횟수, 토큰 수
에러 발생 위치

LangFuse 연결 (오픈소스 대안)

# .env에 추가
LANGFUSE_SECRET_KEY=sk-lf-xxxxx
LANGFUSE_PUBLIC_KEY=pk-lf-xxxxx
LANGFUSE_HOST=https://cloud.langfuse.com

LangFlow 내장 트레이싱

LangFlow 1.8부터 내장 트레이싱을 지원해요.

플로우 실행 후:
상단 메뉴 → "Logs" 버튼
→ 각 컴포넌트 실행 시간, 입출력 확인

컴포넌트 클릭 → "Inspect"
→ 실행 중 중간 결과 확인

---

LangFlow Python SDK로 완전 통합

외부 앱에서 LangFlow를 완전히 제어해요.

from langflow.load import upload_file
import requests
import asyncio

LANGFLOW_URL = "http://localhost:7860"
API_KEY = "your-langflow-api-key"

class LangFlowClient:
    def __init__(self, url: str, api_key: str):
        self.url = url
        self.headers = {
            "x-api-key": api_key,
            "Content-Type": "application/json"
        }

    def run_flow(
        self,
        flow_id: str,
        message: str,
        session_id: str = None
    ) -> str:
        """플로우 실행"""
        payload = {
            "input_value": message,
            "output_type": "chat",
            "input_type": "chat",
        }
        if session_id:
            payload["session_id"] = session_id

        response = requests.post(
            f"{self.url}/api/v1/run/{flow_id}",
            headers=self.headers,
            json=payload,
            timeout=60
        )
        response.raise_for_status()
        data = response.json()
        return data["outputs"][0]["outputs"][0]["results"]["message"]["text"]

    def upload_file(self, flow_id: str, file_path: str) -> str:
        """파일 업로드"""
        with open(file_path, "rb") as f:
            response = requests.post(
                f"{self.url}/api/v2/files/",
                headers={"x-api-key": self.headers["x-api-key"]},
                files={"file": f},
                params={"flow_id": flow_id}
            )
        return response.json()["path"]

    def stream_flow(self, flow_id: str, message: str):
        """스트리밍 응답"""
        response = requests.post(
            f"{self.url}/api/v1/run/{flow_id}?stream=true",
            headers=self.headers,
            json={
                "input_value": message,
                "output_type": "chat",
                "input_type": "chat",
            },
            stream=True
        )

        for chunk in response.iter_content(chunk_size=None):
            if chunk:
                yield chunk.decode()

# 사용
client = LangFlowClient(LANGFLOW_URL, API_KEY)

# 일반 실행
answer = client.run_flow(
    flow_id="your-rag-flow-id",
    message="계약서 조항 3번이 뭔가요?",
    session_id="user-123"  # 사용자별 세션 분리
)
print(answer)

# 파일 업로드 후 실행
file_path = client.upload_file("your-rag-flow-id", "contract.pdf")
answer = client.run_flow("your-rag-flow-id", "이 계약서 핵심 조항 요약해줘")

# 스트리밍
for chunk in client.stream_flow("your-flow-id", "질문"):
    print(chunk, end="", flush=True)

---

시리즈 마무리

4편에 걸쳐 LangFlow를 완전히 다뤘어요.

1편: 개요, 개념, 환경 세팅
2편: 기본 챗봇, PDF RAG, API 배포
3편: 에이전트, 메모리, 커스텀 컴포넌트, 조건 분기
4편: 멀티 에이전트, MCP, 프로덕션 배포, 모니터링

LangFlow가 특히 빛나는 상황:

✅ RAG 앱 빠르게 프로토타입
✅ 비개발자와 협업
✅ 여러 LLM 실험
✅ 내부 지식베이스 챗봇
✅ MCP 서버로 Claude/Cursor 확장

 

 

📌 이전 글: LangFlow 3편 — 에이전트, 메모리, 커스텀 컴포넌트

LangFlow 완전 가이드 3편 — 심화 튜토리얼 : 에이전트, 메모리, 커스텀 컴포넌트