최근에 회사 프로젝트로 Agent를 만들어야 하는 일이 있었다.
내부적으로는 LangFlow, Open Web UI 조합을 사용했고, 두 개의 Agent를 구현했다.
- 세금 관련 DOCX 리포트를 작성하는 AI
- 정보성 데이터를 취합해 요약 브리핑하는 Agent
결과적으로 둘 다 상당히 긴 LangChain으로 이어졌다. 그 과정에서 부딪힌 문제들과 깨달은 것들을 정리해본다.
LangFlow 활용도를 높이려면 결국 커스텀 모듈
LangFlow의 활용도를 끌어올리려면 어느 정도는 직접 커스텀 컴포넌트를 만들 수 있어야 한다. 예를 들어 스트리밍된 LLM 결과를 받아 DOCX 파일로 저장해야 하는 경우가 그렇다.
대략 이런 식으로 구현했다.
import os
import tempfile
import uuid
from pathlib import Path
from docx import Document
from lfx.custom import Component
from lfx.io import MessageInput, StrInput, Output
from lfx.schema import Message
class SaveAsDocx(Component):
display_name = "Save as DOCX"
description = "Streaming passthrough + 이메일 docx 저장 + 다운로드 링크"
icon = "file-text"
_session_counters: dict = {}
inputs = [
MessageInput(name="message", display_name="Message"),
StrInput(name="filename_prefix", display_name="Filename Prefix", value="report_draft"),
]
outputs = [
Output(display_name="Message with Link", name="out", method="build", types=["Message"]),
]
...
def build(self) -> Message:
"""Returns a Message whose text is an async generator (streaming passthrough)."""
from langflow.services.deps import get_storage_service
# 환경변수에서 Langflow public base URL 로드
langflow_base = os.environ.get(
"LANGFLOW_PUBLIC_URL",
"https://your-langflow-host.example.com",
).rstrip("/")
# 업스트림 입력 추출
msg = self.message
upstream_text = msg.text if hasattr(msg, "text") else msg
# flow_id는 stream 시작 전에 미리 확보
flow_id = None
if hasattr(self, "graph") and self.graph is not None:
flow_id = getattr(self.graph, "flow_id", None)
if flow_id is None:
flow_id = getattr(self, "_flow_id", None) or "standalone"
flow_id = str(flow_id)
filename_prefix = self.filename_prefix or "report_draft"
async def _passthrough_stream():
"""Yield chunks downstream while accumulating for docx."""
accumulated = ""
# 1. 업스트림이 string이면 한 번에 yield하고 끝
if isinstance(upstream_text, str):
accumulated = upstream_text
if accumulated:
yield SaveAsDocx._Chunk(accumulated)
# 2. async generator (LLM Agent stream)
elif hasattr(upstream_text, "__aiter__"):
async for chunk in upstream_text:
piece = SaveAsDocx._chunk_to_text(chunk)
if piece:
accumulated += piece
yield SaveAsDocx._Chunk(piece)
# 3. sync generator
elif hasattr(upstream_text, "__iter__") and not isinstance(
upstream_text, (bytes, bytearray)
):
for chunk in upstream_text:
piece = SaveAsDocx._chunk_to_text(chunk)
if piece:
accumulated += piece
yield SaveAsDocx._Chunk(piece)
# 4. 기타 — string으로 캐스팅
else:
accumulated = str(upstream_text)
if accumulated:
yield SaveAsDocx._Chunk(accumulated)
# 5. stream 종료 — 채널 판별
channel = self._detect_channel(accumulated)
# 6. 일부 채널은 docx 생성 안 함
if channel in ("sms", "mms"):
self.status = f"channel={channel}, docx 생성 생략"
return
# 7. docx 생성
tmp_dir = Path(tempfile.gettempdir()) / "langflow_generated"
tmp_dir.mkdir(parents=True, exist_ok=True)
# 세션별 idx 카운터
session_id = (
getattr(self, "session_id", None)
or getattr(getattr(self, "graph", None), "session_id", None)
or "default"
)
session_id = str(session_id)
SaveAsDocx._session_counters[session_id] = (
SaveAsDocx._session_counters.get(session_id, 0) + 1
)
idx = SaveAsDocx._session_counters[session_id]
filename = f"{filename_prefix}_{idx}.docx"
tmp_path = tmp_dir / filename
self._markdown_to_docx(accumulated, tmp_path)
# 8. 스토리지 등록
storage = get_storage_service()
data = tmp_path.read_bytes()
await storage.save_file(flow_id=flow_id, file_name=filename, data=data)
# 9. 다운로드 링크를 마지막 청크로 yield (절대 URL)
download_url = f"{langflow_base}/api/v1/files/download/{flow_id}/{filename}"
link_block = (
f"\n\n---\n"
f"📎 **다운로드**: [{filename}]({download_url})"
)
yield SaveAsDocx._Chunk(link_block)
self.status = f"channel={channel}, stored: {flow_id}/{filename}"
return Message(text=_passthrough_stream())핵심은 passthrough 스트리밍이다. LLM이 스트림으로 뱉어내는 청크를 그대로 다음 노드로 흘려보내면서, 동시에 내부에 누적해두었다가 종료 시점에 한 번에 docx로 저장한다. 사용자는 응답이 실시간으로 나타나는 것처럼 보이고, 마지막 청크에서 다운로드 링크를 받게 된다.
토큰 한계가 만든 징그러운 LangChain
Agent 개발 초기에는 여러 제약 조건이 있었다. 특히 우리는 사내 Agent를 써야 했는데, 초기 max tokens가 1,000개 남짓이었다. 긴 리포트를 한 번에 만들 수 없으니 여러 AI Agent를 병렬 또는 직렬로 엮어서 결과를 조립할 수밖에 없었다.
그렇게 만들어진 것이 아래의 그림 같은 무시무시한 LangChain이다.
프롬프트 작성에 Claude를 진심으로 활용했다
여기서 자연스럽게 짚고 가야 할 부분이 있다. 위처럼 수많은 Agent와 Prompt Template을 엮어 만드는 구조에서는, 각 Agent가 정확히 자기 담당 부분만 작성하도록 프롬프트를 매우 정밀하게 짜야 한다. 예를 들어 한 리포트의 Header / Section 1 / Section 2 / Section 3 / FAQ / 체크리스트를 각각 다른 Agent가 쓰는데, 그 사이의 중복이나 누락을 막으려면:
- 각 Agent에게 "당신이 담당할 섹션은 정확히 여기까지"라고 못박고
- 입력 변수 형식과 출력 형식을 토씨 하나 어긋나지 않게 명세하고
- 분기 규칙(예: 세액공제 납입내역이 0이면 전략 3번을 다른 내용으로)을 빼먹지 않게 적어야 한다.
이걸 사람이 한 번에 깔끔하게 써내려가는 건 거의 불가능에 가깝다. 그래서 프롬프트를 작성할 때마다 Claude를 거의 페어 프로그래머처럼 옆에 두고 활용했다. 초안을 던지고 → 모호한 부분 짚어달라고 하고 → 분기 규칙을 표로 정리하게 하고 → 출력 예시를 만들어 검증하고 → 다시 다듬는 식이다.
특히 Prompt Template 컴포넌트에 들어갈 변수 이름, 시스템 프롬프트의 절대 규칙 섹션, 그리고 LLM이 자유 변형하지 못하게 막는 가드 문구(반드시 그대로 출력, 토씨 하나 바꾸지 말 것 같은 표현)는 거의 다 Claude와의 대화에서 정리된 것들이다. 결과적으로 프롬프트 자체가 제품의 일부가 되었고, 그 품질이 곧 Agent 출력의 품질로 직결됐다.
직접 부딪힌 사내 Agent의 잔고장들
LLM Agent도 사내에서 개발된 Agent를 사용했다. 서버에서 안정화를 위해 자주 업데이트가 있었는데, 매번 새 버전이 올라올 때마다 묘하게 다른 부분에서 깨졌다.
대표적으로 겪은 이슈는 이런 것들이다.
- Prompt Template에 Agent의 출력이 안 들어감 → async 실행 타이밍 이슈로 보였다. 어떤 버전에선 되고, 어떤 버전에선 빈 문자열만 흐르는.
- 병렬 스트리밍이 글자 단위로 깨짐 → 한 리포트의 Header / Section 1 / Section 2 / ... 를 각각 다른 Agent가 동시에 작성했는데, 스트리밍이 동시다발적으로 문자를 뱉으면서 화면에 글자가 섞여 깨져 보이는 현상이 있었다. 한 Agent의 토큰 사이에 다른 Agent의 토큰이 끼어드는 모양새다.
이걸 막아보려고 Chat Output을 중간에 끼워 넣어 Agent를 직렬로 강제해보기도 했다. 그런데 이번엔 다른 문제가 생겼다.
- Agent가 여러 번 실행됨 (idempotent하지 않음)
- 종료가 제대로 안 됨 — 끝났는지 안 끝났는지 모호하게 매달려 있는 상태
플랫폼 자체의 한계도 있고, 우리가 쓰던 Agent 버전의 한계도 있어서, 이 시기에는 "동작하는 조합"을 찾는 게 거의 도박이었다.
입력 분기 — 단순 If-Else로는 부족하다
같은 채팅 입력이라도 의도에 따라 처리 흐름이 완전히 달라야 했다.
예를 들어:
- 슬래시(/)로 구분된 정해진 포맷 — 500만원 / 300만원 / 200만원 / 2023.03.05 / 2025.04.01 — 이 들어오면 → 파서로 분해 → 계산기로 → 리포트 작성 Agent
- 위 포맷이 아니면 → 다음 분류기로
- 질문이면 → 질의응답 Agent
- 수정 요청이면 → 기존 문서를 기반으로 수정하는 Agent
- 정보성 글에 카테고리 입력(정치, 경제, 인공지능 같은 콤마 구분 키워드)이면 → split으로 키워드별 분기 → 각각 검색/요약
특히 마지막 카테고리 분기에는 메타 프롬프팅(Meta-Prompting) 방식을 활용했다.
메타 프롬프팅이란 AI에게 더 나은 결과를 얻기 위해 '프롬프트를 작성하는 방법' 자체를 AI에게 묻고, 그 답변을 다시 프롬프트로 활용하는 고도화된 프롬프트 엔지니어링 기법이다.
다만 실제 운영에서는 실행 속도 문제 때문에 AI에게 매번 프롬프트 작성을 시키지는 않고, 자동화된 템플릿으로 동일한 효과를 내도록 미리 만들어두었다.
이런 분기 로직을 LangFlow의 If-Else만으로 안정적으로 짜는 건 생각보다 까다로웠다. 결국 의도 분류용 보조 컴포넌트를 직접 만들어 쓰는 식으로 풀었다.
Agent가 8K 토큰을 뱉기 시작한 후
이후 사내 Agent가 개선되어 8,000+ 토큰을 한 번에 뱉을 수 있게 되면서 상황이 크게 달라졌다.
기존에 6~8개 Agent를 병렬·직렬로 엮어 만들던 리포트를 하나의 Agent에 통합 프롬프트로 맡길 수 있게 된 것이다.
이건 단순히 비용/속도 문제가 아니라 안정성 문제였다. Agent 수가 줄어드니:
- 병렬 스트리밍이 깨지는 문제 → 사라짐 (단일 스트림이니까)
- async 타이밍으로 변수가 비는 문제 → 사라짐
- "여러 번 실행됨" 같은 라이프사이클 이슈 → 사라짐
- 디버깅 시 따라가야 할 노드 수 → 1/6
물론 단일 Agent에 모든 걸 맡기면 출력 토큰 비용이 늘고, 부분 재생성이 어렵다는 trade-off는 있다. 그래도 종합적으로는 압도적으로 안정적이었다.
이때 통합 프롬프트를 잘 짜는 것이 다시 중요해졌고, 여기서도 Claude의 역할이 컸다. 6개 Agent의 프롬프트 6개를 하나로 합치면서 분기 규칙, 변환 규칙, 출력 형식 가이드, 절대 규칙을 한 곳에 응축시키는 작업은 사람 혼자 하기엔 너무 노이즈가 많다.
아직 풀지 못한 숙제 — Multi-turn
남아 있는 어려운 부분은 Multi-turn이다.
- 특정 리포트를 만든 후, 그 리포트에 대해 계속 질의응답
- "이 섹션 빼줘", "근속연수 25년으로 다시 계산해줘" 같은 수정 요청
- 같은 채팅 세션 내에서는 이전 결과를 기억하되, 새 채팅을 시작하면 이전 세션 파일이 안 보이도록 격리
이건 단순히 Agent에 컨텍스트를 한 줄 더 넣는다고 끝나지 않는다. 결국 Chat Input의 의도를 처음에 정확히 분류해서, LangChain을 적절한 분기점으로 보내야 한다.
지금은 다음과 같은 구조로 풀어가고 있다.
- 매 입력마다 직전 리포트 백업 파일(.md)을 로드 시도 — 있으면 멀티턴, 없으면 신규
- 멀티턴이면 의도 분류 — 질문인지, 수정 요청인지
- 분기에 따라 질의응답 LLM 또는 수정 LLM으로 라우팅
- 수정 결과는 새 docx + 새 .md 백업으로 저장 → 다음 턴이 이걸 자동으로 집어감
- 파일명은 {prefix}_{session_id}_{idx} 형태로 세션 격리
이 구조도 결국 분기 컴포넌트와 프롬프트 설계가 핵심이다. 컴포넌트는 직접 만들 수 있는데, 프롬프트에서 계속 "이건 질문이야 수정이야?"를 LLM에게 다시 묻기 시작하면 비용도 늘고 지연도 늘어나기 때문에, 간단한 키워드 기반 분류기로 1차 라우팅을 하는 식으로 타협했다.
마지막 — 깨달은 점들
정리하면 이렇다.
1. LangFlow는 강력하지만, 커스텀 컴포넌트 없이 한계까지 가긴 어렵다. 스트리밍 passthrough, 파일 저장, 의도 분류 같은 것들은 결국 직접 짜게 된다.
2. Agent의 토큰 한계는 단순한 숫자가 아니라 아키텍처 결정 요인이다. 1K 토큰 시대의 그래프와 8K 토큰 시대의 그래프는 구조 자체가 다르다. 토큰이 늘어나면 더 단순한 구조가 정답이 된다.
3. 프롬프트는 제품의 일부다. Claude를 옆에 두고 같이 다듬어라. 프롬프트의 미세한 표현 하나가 안정성과 출력 품질을 좌우한다. 혼자 쓰지 말고, 상위 LLM과 같이 정리하는 워크플로를 만들어두면 시간을 한 자릿수 배수로 줄일 수 있다.
4. async/streaming/병렬은 화려하지만 깨지기 쉽다. 가능하면 직렬·단일 Agent로 가고, 병렬은 정말 필요할 때만 써라.
5. Multi-turn은 결국 "초기 의도 분류"에서 승부가 난다. 프롬프트로만 풀려고 하지 말고, 분기 컴포넌트와 키워드 기반 라우팅을 적극 활용하는 게 안정성 측면에서 훨씬 낫다.