본 콘텐츠의 이미지 및 내용은 AI로 생성되었습니다.
본 콘텐츠의 이미지 및 내용을 무단으로 복제, 배포, 수정하여 사용할 경우 저작권법에 의해 법적 제재를 받을 수 있습니다.
이미지 로딩 중...
AI Generated
2025. 12. 3. · 17 Views
프롬프트 체이닝 에이전트 구현 완벽 가이드
LLM을 활용한 프롬프트 체이닝 에이전트를 처음부터 끝까지 구현하는 방법을 배웁니다. 에이전트 설계부터 스트림릿 UI까지, 실무에서 바로 활용할 수 있는 체계적인 가이드입니다.
목차
1. 에이전트 구조 설계하기
김개발 씨는 회사에서 고객 문의를 자동으로 분류하고 답변하는 시스템을 만들어 달라는 요청을 받았습니다. 단순히 LLM API를 호출하는 것만으로는 부족했습니다.
여러 단계의 처리가 필요했고, 각 단계가 유기적으로 연결되어야 했습니다.
프롬프트 체이닝 에이전트는 여러 개의 LLM 호출을 순차적으로 연결하여 복잡한 작업을 수행하는 시스템입니다. 마치 공장의 조립 라인처럼, 각 단계에서 특정 작업을 수행하고 그 결과를 다음 단계로 전달합니다.
이를 통해 단일 프롬프트로는 해결하기 어려운 복잡한 문제를 체계적으로 처리할 수 있습니다.
다음 코드를 살펴봅시다.
from dataclasses import dataclass
from typing import List, Callable, Any, Optional
from enum import Enum
class ChainStatus(Enum):
PENDING = "pending"
RUNNING = "running"
COMPLETED = "completed"
FAILED = "failed"
@dataclass
class ChainStep:
"""체인의 각 단계를 정의하는 클래스"""
name: str
prompt_template: str
processor: Optional[Callable] = None
@dataclass
class AgentConfig:
"""에이전트 전체 설정을 담는 클래스"""
steps: List[ChainStep]
model_name: str = "gpt-4"
temperature: float = 0.7
max_retries: int = 3
김개발 씨는 입사 6개월 차 주니어 개발자입니다. 어느 날 팀장님이 다가와 새로운 프로젝트를 맡겼습니다.
"고객 문의가 들어오면 자동으로 분류하고, 적절한 답변을 생성하고, 마지막으로 톤을 조절해서 보내는 시스템을 만들어 주세요." 김개발 씨는 고민에 빠졌습니다. ChatGPT API 호출은 해봤지만, 이렇게 여러 단계를 거쳐야 하는 복잡한 작업은 처음이었습니다.
어떻게 해야 할까요? 선배 개발자 박시니어 씨가 커피를 건네며 말했습니다.
"프롬프트 체이닝이라는 기법을 써보는 게 어때? 복잡한 작업을 여러 단계로 나눠서 처리하는 거야." 프롬프트 체이닝이란 무엇일까요?
쉽게 비유하자면, 마치 릴레이 경주와 같습니다. 첫 번째 주자가 달리고, 바통을 두 번째 주자에게 넘기고, 또 세 번째 주자에게 넘기는 것처럼요.
각 주자는 자신의 구간만 집중해서 달립니다. LLM도 마찬가지입니다.
각 단계에서 하나의 작업에만 집중하면 훨씬 더 정확한 결과를 얻을 수 있습니다. 왜 이런 방식이 필요할까요?
단일 프롬프트로 모든 것을 처리하려고 하면 여러 문제가 생깁니다. 첫째, 프롬프트가 너무 길고 복잡해집니다.
둘째, LLM이 여러 작업을 동시에 처리하다 보니 각각의 품질이 떨어집니다. 셋째, 어느 단계에서 문제가 생겼는지 파악하기 어렵습니다.
바로 이런 문제를 해결하기 위해 에이전트 구조를 체계적으로 설계해야 합니다. 위 코드를 살펴보면, 먼저 ChainStatus 열거형으로 각 단계의 상태를 관리합니다.
대기 중인지, 실행 중인지, 완료되었는지, 실패했는지를 명확하게 구분할 수 있습니다. ChainStep 데이터 클래스는 체인의 각 단계를 정의합니다.
name은 단계의 이름, prompt_template은 해당 단계에서 사용할 프롬프트 템플릿, processor는 LLM 응답을 후처리할 함수입니다. 이렇게 각 단계를 독립적인 객체로 만들면 재사용과 테스트가 훨씬 쉬워집니다.
AgentConfig는 에이전트 전체의 설정을 담습니다. 어떤 단계들로 구성되어 있는지, 어떤 모델을 사용할지, 온도 값은 얼마로 할지, 실패 시 몇 번까지 재시도할지를 한곳에서 관리합니다.
실제 현업에서는 이런 구조가 매우 중요합니다. 예를 들어 고객 문의 처리 시스템이라면, 첫 번째 단계에서 문의 유형을 분류하고, 두 번째 단계에서 관련 정보를 추출하고, 세 번째 단계에서 답변을 생성하고, 네 번째 단계에서 톤을 조절하는 식으로 구성할 수 있습니다.
주의할 점도 있습니다. 너무 많은 단계로 나누면 오히려 복잡해지고 비용도 늘어납니다.
일반적으로 3~5단계 정도가 적당합니다. 또한 각 단계의 책임을 명확하게 정의해야 합니다.
한 단계에서 여러 가지를 처리하려고 하면 체이닝의 장점이 사라집니다. 박시니어 씨의 조언을 들은 김개발 씨는 고개를 끄덕였습니다.
"구조를 먼저 잡으니까 훨씬 명확해지네요. 이제 각 단계를 하나씩 구현해 볼게요!"
실전 팁
💡 - 각 단계는 하나의 책임만 갖도록 설계하세요
- dataclass를 활용하면 보일러플레이트 코드를 줄일 수 있습니다
- 설정은 한곳에서 관리하여 변경이 쉽도록 만드세요
2. 첫 번째 체인 구현
구조 설계를 마친 김개발 씨는 이제 실제로 동작하는 첫 번째 체인을 구현할 차례입니다. OpenAI API를 연동하고, 프롬프트 템플릿을 만들고, 실제로 LLM을 호출하는 코드를 작성해야 합니다.
어디서부터 시작해야 할까요?
첫 번째 체인은 에이전트의 기초가 되는 핵심 구성 요소입니다. OpenAI 클라이언트 초기화, 프롬프트 템플릿 정의, LLM 호출 및 응답 처리라는 세 가지 요소를 구현합니다.
이 기반 위에 나머지 체인들이 쌓아 올려집니다.
다음 코드를 살펴봅시다.
import openai
from string import Template
class ChainExecutor:
"""체인을 실행하는 핵심 클래스"""
def __init__(self, config: AgentConfig):
self.config = config
self.client = openai.OpenAI()
def execute_step(self, step: ChainStep, input_data: str) -> str:
# 프롬프트 템플릿에 입력 데이터 삽입
prompt = Template(step.prompt_template).safe_substitute(
input=input_data
)
# LLM API 호출
response = self.client.chat.completions.create(
model=self.config.model_name,
messages=[{"role": "user", "content": prompt}],
temperature=self.config.temperature
)
result = response.choices[0].message.content
# 후처리 함수가 있으면 실행
if step.processor:
result = step.processor(result)
return result
김개발 씨는 설계한 구조를 바탕으로 실제 코드를 작성하기 시작했습니다. 가장 먼저 해야 할 일은 LLM과 통신하는 핵심 로직을 만드는 것이었습니다.
"일단 하나의 단계를 실행하는 것부터 만들어 볼까?" 김개발 씨는 혼잣말을 하며 키보드에 손을 올렸습니다. ChainExecutor 클래스는 체인의 각 단계를 실행하는 핵심 역할을 담당합니다.
마치 오케스트라의 지휘자처럼, 각 악기(단계)가 언제 어떻게 연주해야 하는지를 조율합니다. 생성자에서 설정을 받아 저장하고, OpenAI 클라이언트를 초기화합니다.
execute_step 메서드가 실제 작업을 수행합니다. 이 메서드는 세 단계로 동작합니다.
첫째, 프롬프트 템플릿에 입력 데이터를 삽입합니다. 둘째, LLM API를 호출합니다.
셋째, 결과를 반환하되, 후처리 함수가 있으면 실행합니다. 프롬프트 템플릿이란 무엇일까요?
편지의 빈칸 채우기와 비슷합니다. "안녕하세요, ${name}님"처럼 템플릿을 만들어 두면, 나중에 name 부분에 실제 이름을 넣을 수 있습니다.
여기서는 Python의 Template 클래스를 사용합니다. safe_substitute 메서드는 변수가 없어도 에러를 발생시키지 않아 안전합니다.
OpenAI API 호출 부분을 자세히 살펴보겠습니다. chat.completions.create 메서드로 대화형 모델을 호출합니다.
model 파라미터에 사용할 모델 이름을, messages에 대화 내용을, temperature에 창의성 수준을 전달합니다. temperature가 0에 가까우면 일관된 답변을, 1에 가까우면 다양한 답변을 생성합니다.
응답에서 실제 텍스트를 추출하는 부분도 중요합니다. response.choices[0].message.content로 첫 번째 선택지의 메시지 내용을 가져옵니다.
OpenAI API는 여러 개의 답변을 생성할 수 있는데, 보통은 첫 번째 것을 사용합니다. 마지막으로 후처리 함수가 있습니다.
LLM의 응답을 그대로 사용하기 어려운 경우가 많습니다. JSON 형식으로 파싱해야 하거나, 특정 형식으로 변환해야 하거나, 불필요한 부분을 제거해야 할 수 있습니다.
processor 함수가 바로 이런 역할을 합니다. 실무에서는 이런 기본 구조가 매우 유용합니다.
예를 들어 감정 분석 체인이라면, 프롬프트 템플릿에 "다음 텍스트의 감정을 분석해주세요: ${input}"처럼 작성하고, processor에서 결과를 파싱하는 함수를 연결하면 됩니다. 김개발 씨는 첫 번째 체인이 동작하는 것을 확인하고 뿌듯해했습니다.
"이제 이걸 연결해서 전체 흐름을 만들어야겠네."
실전 팁
💡 - API 키는 환경 변수로 관리하여 코드에 직접 노출되지 않도록 하세요
- temperature 값은 작업 특성에 따라 조절하세요 (분류는 낮게, 창작은 높게)
- safe_substitute를 사용하면 누락된 변수로 인한 에러를 방지할 수 있습니다
3. 체인 연결 로직 작성
첫 번째 단계가 동작하는 것을 확인한 김개발 씨는 이제 여러 단계를 연결해야 합니다. 첫 번째 단계의 출력이 두 번째 단계의 입력이 되고, 두 번째의 출력이 세 번째의 입력이 되는 식으로요.
마치 도미노처럼 하나가 다음 하나를 이끄는 구조입니다.
체인 연결 로직은 여러 단계를 순차적으로 실행하면서 데이터를 전달하는 역할을 합니다. 각 단계의 실행 상태를 추적하고, 중간 결과를 저장하며, 에러 발생 시 적절히 처리합니다.
이 로직이 있어야 비로소 진정한 체이닝이 완성됩니다.
다음 코드를 살펴봅시다.
from typing import Dict
import logging
logger = logging.getLogger(__name__)
class ChainRunner:
"""여러 체인 단계를 연결하여 실행하는 클래스"""
def __init__(self, executor: ChainExecutor):
self.executor = executor
self.results: Dict[str, str] = {}
self.status: Dict[str, ChainStatus] = {}
def run(self, initial_input: str) -> Dict[str, str]:
current_input = initial_input
for step in self.executor.config.steps:
self.status[step.name] = ChainStatus.RUNNING
logger.info(f"실행 중: {step.name}")
try:
result = self.executor.execute_step(step, current_input)
self.results[step.name] = result
self.status[step.name] = ChainStatus.COMPLETED
current_input = result # 다음 단계의 입력으로 전달
except Exception as e:
self.status[step.name] = ChainStatus.FAILED
logger.error(f"실패: {step.name} - {e}")
raise
return self.results
김개발 씨는 각 단계가 개별적으로는 잘 동작하지만, 이것들을 연결하는 게 생각보다 복잡하다는 것을 깨달았습니다. 그냥 for문으로 돌리면 되는 거 아닌가 싶었는데, 실제로는 고려할 것이 많았습니다.
박시니어 씨가 지나가다 말했습니다. "체인 연결할 때는 세 가지를 꼭 신경 써야 해.
상태 추적, 결과 저장, 에러 처리. 이 세 가지가 없으면 디버깅할 때 정말 고생해." ChainRunner 클래스는 체인의 총괄 관리자입니다.
마치 프로젝트 매니저가 각 팀의 진행 상황을 파악하고 조율하듯이, ChainRunner는 각 단계의 실행을 관리합니다. 생성자에서 두 개의 딕셔너리를 초기화합니다.
results는 각 단계의 결과를 저장하고, status는 각 단계의 현재 상태를 저장합니다. 이렇게 하면 나중에 "3번째 단계에서 무슨 결과가 나왔지?" 또는 "어느 단계까지 성공했지?"라는 질문에 바로 답할 수 있습니다.
run 메서드가 핵심입니다. 이 메서드는 초기 입력을 받아서 모든 단계를 순차적으로 실행합니다.
current_input 변수가 바통 역할을 합니다. 처음에는 사용자가 전달한 initial_input을 담고 있다가, 각 단계가 끝날 때마다 그 결과로 업데이트됩니다.
for 루프 안에서 일어나는 일을 자세히 살펴봅시다. 먼저 해당 단계의 상태를 RUNNING으로 변경합니다.
그리고 로깅을 남깁니다. 로깅은 정말 중요합니다.
나중에 문제가 생겼을 때 어디서 무슨 일이 일어났는지 추적할 수 있게 해줍니다. try-except 블록으로 에러를 처리합니다.
단계 실행이 성공하면 결과를 저장하고 상태를 COMPLETED로 변경합니다. 그리고 가장 중요한 부분, current_input = result로 다음 단계에 결과를 전달합니다.
실패하면 상태를 FAILED로 변경하고 에러를 로깅한 뒤 예외를 다시 발생시킵니다. 왜 예외를 다시 발생시킬까요?
호출한 쪽에서 이 에러를 처리할 수 있도록 하기 위해서입니다. 어떤 경우에는 에러가 발생해도 계속 진행하고 싶을 수 있고, 어떤 경우에는 즉시 중단하고 싶을 수 있습니다.
이 결정은 ChainRunner를 사용하는 쪽에서 내리는 것이 좋습니다. 실무에서 이 패턴은 매우 유용합니다.
예를 들어 고객 문의 처리에서 "분류 -> 정보 추출 -> 답변 생성" 순서로 체인을 구성했다고 합시다. 분류 단계에서 실패하면 전체 처리가 중단되고, 어느 단계에서 실패했는지 정확히 알 수 있습니다.
또한 results 딕셔너리를 통해 "분류 결과가 뭐였지?"라고 나중에 확인할 수도 있습니다. 김개발 씨는 체인 연결 로직을 완성하고 테스트해 보았습니다.
각 단계가 순서대로 실행되고, 결과가 제대로 전달되는 것을 확인할 수 있었습니다.
실전 팁
💡 - 로깅은 반드시 추가하세요. 디버깅 시간을 크게 줄여줍니다
- 중간 결과를 저장해두면 실패 시 처음부터 다시 시작하지 않아도 됩니다
- 상태 추적으로 진행 상황을 실시간으로 모니터링할 수 있습니다
4. 중간 결과 처리하기
체인이 잘 연결되어 동작하는 것을 확인한 김개발 씨에게 새로운 과제가 생겼습니다. LLM의 응답이 항상 예상한 형식으로 오지 않는다는 것이었습니다.
어떤 때는 JSON으로, 어떤 때는 일반 텍스트로, 어떤 때는 불필요한 설명이 붙어서 옵니다. 이것들을 어떻게 처리해야 할까요?
중간 결과 처리는 체인의 안정성을 높이는 핵심 요소입니다. LLM 응답을 파싱하고, 검증하고, 필요한 형식으로 변환합니다.
또한 재시도 로직과 폴백 처리를 통해 예상치 못한 응답에도 대응할 수 있게 합니다.
다음 코드를 살펴봅시다.
import json
import re
from typing import Optional
class ResultProcessor:
"""LLM 응답을 처리하고 검증하는 클래스"""
@staticmethod
def extract_json(text: str) -> Optional[dict]:
# JSON 블록 추출 시도
json_match = re.search(r'\{[\s\S]*\}', text)
if json_match:
try:
return json.loads(json_match.group())
except json.JSONDecodeError:
return None
return None
@staticmethod
def clean_response(text: str) -> str:
# 불필요한 마크다운 제거
text = re.sub(r'```[\w]*\n?', '', text)
return text.strip()
@staticmethod
def validate_response(text: str, required_fields: list) -> bool:
data = ResultProcessor.extract_json(text)
if not data:
return False
return all(field in data for field in required_fields)
김개발 씨는 체인을 테스트하다가 이상한 현상을 발견했습니다. 같은 프롬프트를 보내도 LLM의 응답 형식이 조금씩 달랐습니다.
어떤 때는 순수한 JSON만 오고, 어떤 때는 "다음은 요청하신 결과입니다:"라는 서두가 붙어서 오고, 어떤 때는 마크다운 코드 블록으로 감싸져서 왔습니다. "이거 일일이 처리하려면 정말 귀찮겠는데..." 김개발 씨는 한숨을 쉬었습니다.
박시니어 씨가 웃으며 말했습니다. "LLM은 사람처럼 대답하거든.
매번 똑같이 대답하지 않아. 그래서 중간 결과 처리가 필수야." ResultProcessor 클래스는 이런 불규칙한 응답을 일관된 형식으로 변환하는 역할을 합니다.
마치 통역사가 여러 나라 말을 하나의 언어로 번역해주는 것처럼요. extract_json 메서드는 텍스트에서 JSON을 추출합니다.
정규표현식으로 중괄호로 시작하고 끝나는 부분을 찾습니다. LLM이 "결과는 다음과 같습니다: {"status": "success"}"처럼 응답해도 JSON 부분만 깔끔하게 추출할 수 있습니다.
파싱에 실패하면 None을 반환하여 안전하게 처리합니다. clean_response 메서드는 불필요한 마크다운 문법을 제거합니다.
LLM은 코드를 반환할 때 종종 python이나 json 같은 마크다운 코드 블록으로 감쌉니다. 이런 문법을 제거하고 순수한 내용만 남깁니다.
validate_response 메서드는 응답이 필요한 필드를 모두 포함하는지 검증합니다. 예를 들어 분류 결과에는 반드시 "category"와 "confidence" 필드가 있어야 한다면, 이 메서드로 검증할 수 있습니다.
필드가 하나라도 빠져 있으면 False를 반환합니다. 이런 처리가 왜 중요할까요?
체인에서는 이전 단계의 출력이 다음 단계의 입력이 됩니다. 만약 2번째 단계에서 잘못된 형식의 데이터가 나오면, 3번째 단계가 제대로 동작하지 않습니다.
그리고 에러 메시지는 3번째 단계에서 나오기 때문에, 실제 문제가 어디서 발생했는지 찾기 어렵습니다. 실무에서는 더 정교한 처리가 필요할 수 있습니다.
예를 들어 여러 번 시도해서 올바른 형식을 얻거나, 형식이 잘못되면 LLM에게 다시 요청하거나, 기본값으로 대체하는 등의 전략이 있습니다. 김개발 씨는 ResultProcessor를 ChainStep의 processor로 연결했습니다.
이제 각 단계의 출력이 자동으로 정제되어 다음 단계로 전달됩니다. 불규칙한 LLM 응답도 두렵지 않습니다.
실전 팁
💡 - JSON 파싱 실패에 대비해 항상 예외 처리를 하세요
- 정규표현식은 다양한 형식에 대응할 수 있도록 유연하게 작성하세요
- 검증 실패 시 재시도 로직을 추가하면 안정성이 높아집니다
5. 최종 출력 생성
모든 체인이 성공적으로 실행되었습니다. 이제 마지막 관문이 남았습니다.
여러 단계를 거쳐 나온 결과들을 종합하여 최종 출력을 만들어야 합니다. 사용자에게 보여줄 깔끔한 형태로 정리하는 것이죠.
최종 출력 생성은 체인의 결과를 사용자가 이해하기 쉬운 형태로 변환하는 단계입니다. 여러 단계의 결과를 종합하고, 필요한 형식으로 포맷팅하며, 메타데이터를 추가합니다.
이 단계가 있어야 체인의 결과물이 실제로 활용 가능해집니다.
다음 코드를 살펴봅시다.
from dataclasses import dataclass, field
from datetime import datetime
from typing import List, Optional
@dataclass
class ChainOutput:
"""체인 실행 결과를 담는 클래스"""
final_result: str
intermediate_results: Dict[str, str]
execution_time: float
timestamp: str = field(default_factory=lambda: datetime.now().isoformat())
class OutputFormatter:
"""최종 출력을 포맷팅하는 클래스"""
@staticmethod
def format_result(runner: ChainRunner, start_time: float) -> ChainOutput:
steps = list(runner.results.keys())
final_step = steps[-1] if steps else None
return ChainOutput(
final_result=runner.results.get(final_step, ""),
intermediate_results=runner.results.copy(),
execution_time=time.time() - start_time
)
@staticmethod
def to_json(output: ChainOutput) -> str:
return json.dumps({
"result": output.final_result,
"steps": output.intermediate_results,
"duration_seconds": round(output.execution_time, 2),
"generated_at": output.timestamp
}, ensure_ascii=False, indent=2)
체인의 모든 단계가 성공적으로 완료되었습니다. 김개발 씨는 결과를 확인하려고 runner.results를 출력해 보았습니다.
딕셔너리에 각 단계의 결과가 잘 담겨 있었습니다. 하지만 이대로 사용자에게 보여주기에는 뭔가 부족했습니다.
"결과는 나왔는데, 이걸 어떻게 정리해서 보여주지?" 김개발 씨가 고민하고 있을 때, 박시니어 씨가 다가왔습니다. "최종 출력 형식도 잘 설계해야 해.
사용자가 필요한 정보를 한눈에 볼 수 있어야 하거든." ChainOutput 데이터 클래스는 결과를 구조화합니다. final_result에는 체인의 마지막 단계 결과가, intermediate_results에는 모든 단계의 결과가, execution_time에는 전체 실행 시간이, timestamp에는 실행 시점이 담깁니다.
왜 이렇게 많은 정보를 담을까요? 실무에서는 다양한 요구사항이 있기 때문입니다.
어떤 사용자는 최종 결과만 필요하고, 어떤 사용자는 중간 단계도 확인하고 싶어합니다. 또한 실행 시간과 타임스탬프는 성능 분석과 감사 로그에 필수입니다.
OutputFormatter 클래스는 결과를 원하는 형식으로 변환합니다. format_result 메서드는 ChainRunner의 결과를 ChainOutput 객체로 변환합니다.
마지막 단계의 결과를 최종 결과로 지정하고, 전체 실행 시간을 계산합니다. to_json 메서드는 ChainOutput을 JSON 문자열로 변환합니다.
ensure_ascii=False는 한글이 유니코드로 이스케이프되지 않도록 합니다. indent=2는 읽기 쉽게 들여쓰기를 추가합니다.
실무에서 이 패턴은 매우 유용합니다. API 응답으로 반환할 때는 JSON 형식이 필요하고, 로그로 남길 때는 텍스트 형식이 필요하고, UI에 표시할 때는 또 다른 형식이 필요합니다.
OutputFormatter에 여러 변환 메서드를 추가하면 모든 상황에 대응할 수 있습니다. 특히 실행 시간 기록은 중요합니다.
LLM API 호출은 비용이 들고, 응답 시간도 일정하지 않습니다. 각 요청의 실행 시간을 기록해두면 성능 병목을 찾거나 비용을 최적화하는 데 큰 도움이 됩니다.
김개발 씨는 최종 출력 형식을 완성했습니다. 이제 체인의 결과를 깔끔하게 정리하여 사용자에게 보여줄 수 있습니다.
실전 팁
💡 - 메타데이터(실행시간, 타임스탬프)는 디버깅과 분석에 필수입니다
- ensure_ascii=False로 한글이 깨지지 않게 하세요
- 여러 출력 형식을 지원하면 다양한 상황에 대응할 수 있습니다
6. 스트림릿 UI 구현
백엔드 로직이 완성되었습니다. 이제 사용자가 실제로 사용할 수 있는 인터페이스를 만들 차례입니다.
김개발 씨는 스트림릿을 사용하기로 했습니다. 빠르게 웹 UI를 만들 수 있고, Python만으로 프론트엔드를 구현할 수 있기 때문입니다.
**스트림릿(Streamlit)**은 Python으로 웹 애플리케이션을 빠르게 만들 수 있는 프레임워크입니다. 데이터 입력, 진행 상황 표시, 결과 출력을 몇 줄의 코드로 구현할 수 있습니다.
프롬프트 체이닝 에이전트의 동작을 시각적으로 확인하기에 완벽한 도구입니다.
다음 코드를 살펴봅시다.
import streamlit as st
import time
st.title("프롬프트 체이닝 에이전트")
# 사이드바 설정
with st.sidebar:
model = st.selectbox("모델 선택", ["gpt-4", "gpt-3.5-turbo"])
temperature = st.slider("Temperature", 0.0, 1.0, 0.7)
# 메인 입력
user_input = st.text_area("처리할 내용을 입력하세요", height=150)
if st.button("실행", type="primary"):
if user_input:
# 진행 상황 표시
progress_bar = st.progress(0)
status_text = st.empty()
steps = ["분류", "분석", "생성", "검토"]
for i, step in enumerate(steps):
status_text.text(f"진행 중: {step}")
progress_bar.progress((i + 1) / len(steps))
time.sleep(1) # 실제로는 체인 실행
# 결과 표시
st.success("처리 완료!")
with st.expander("중간 결과 보기"):
st.json({"step1": "결과1", "step2": "결과2"})
김개발 씨는 터미널에서만 동작하는 프로그램을 팀원들에게 보여주려니 불편했습니다. "웹으로 만들면 좋겠는데, React 배울 시간이 없는데..." 박시니어 씨가 해답을 알려주었습니다.
"스트림릿 써봐. Python으로 웹 UI 금방 만들 수 있어." 스트림릿은 데이터 과학자와 ML 엔지니어를 위해 만들어진 프레임워크입니다.
마치 노트북에 글을 쓰듯이, Python 코드를 위에서 아래로 작성하면 그대로 웹 페이지가 됩니다. HTML, CSS, JavaScript를 전혀 몰라도 됩니다.
코드의 구조를 살펴봅시다. st.title은 페이지 제목을 설정합니다.
st.sidebar는 왼쪽 사이드바를 만듭니다. 여기에 모델 선택과 온도 설정을 넣어서 사용자가 쉽게 조절할 수 있게 했습니다.
st.text_area는 여러 줄 입력 필드를 만듭니다. height 파라미터로 높이를 조절할 수 있습니다.
st.button은 버튼을 만들고, 클릭되면 True를 반환합니다. type="primary"로 강조 스타일을 적용했습니다.
진행 상황 표시가 특히 중요합니다. LLM 호출은 시간이 걸리기 때문에, 사용자가 기다리는 동안 무슨 일이 일어나고 있는지 알려줘야 합니다.
st.progress로 진행률 바를, st.empty로 상태 텍스트를 표시합니다. empty()는 나중에 내용을 업데이트할 수 있는 플레이스홀더입니다.
결과 표시 부분에서 st.success는 초록색 성공 메시지를 보여줍니다. st.expander는 접을 수 있는 섹션을 만듭니다.
중간 결과는 기본적으로 숨겨두고, 필요할 때만 펼쳐볼 수 있게 합니다. st.json은 JSON 데이터를 예쁘게 포맷팅해서 보여줍니다.
실무에서 스트림릿의 장점은 빠른 프로토타이핑입니다. 아이디어를 테스트하거나 내부 도구를 만들 때 매우 유용합니다.
물론 대규모 서비스에는 적합하지 않지만, MVP나 PoC에는 완벽합니다. 김개발 씨는 몇 시간 만에 웹 UI를 완성했습니다.
팀원들이 직접 사용해볼 수 있게 되었고, 피드백을 받기도 훨씬 쉬워졌습니다.
실전 팁
💡 - st.cache_data 데코레이터로 결과를 캐싱하면 성능이 향상됩니다
- st.session_state로 상태를 유지할 수 있습니다
- streamlit run app.py로 실행하고, 파일 저장 시 자동 새로고침됩니다
7. 에이전트 테스트 및 개선
UI까지 완성된 김개발 씨의 에이전트. 하지만 아직 끝이 아닙니다.
실제 서비스에 배포하기 전에 충분한 테스트와 개선이 필요합니다. 예상치 못한 입력에도 잘 동작하는지, 성능은 충분한지, 어떻게 개선할 수 있는지 확인해야 합니다.
테스트와 개선은 안정적인 에이전트를 만드는 마지막 단계입니다. 단위 테스트로 각 컴포넌트를 검증하고, 통합 테스트로 전체 흐름을 확인합니다.
또한 성능 측정과 에러 처리 개선으로 프로덕션 수준의 품질을 달성합니다.
다음 코드를 살펴봅시다.
import pytest
from unittest.mock import Mock, patch
class TestChainExecutor:
"""ChainExecutor 테스트 클래스"""
def test_execute_step_success(self):
config = AgentConfig(steps=[], model_name="gpt-4")
executor = ChainExecutor(config)
step = ChainStep(
name="test_step",
prompt_template="분석해주세요: $input"
)
# LLM 호출 모킹
with patch.object(executor.client.chat.completions, 'create') as mock:
mock.return_value.choices = [Mock(message=Mock(content="결과"))]
result = executor.execute_step(step, "테스트 입력")
assert result == "결과"
def test_chain_runner_full_flow(self):
# 전체 체인 흐름 테스트
steps = [
ChainStep(name="step1", prompt_template="1: $input"),
ChainStep(name="step2", prompt_template="2: $input"),
]
# ... 통합 테스트 로직
에이전트가 완성되었다고 생각한 김개발 씨에게 팀장님이 질문했습니다. "테스트 코드는 작성했어요?" 김개발 씨는 멈칫했습니다.
동작은 확인했지만, 체계적인 테스트 코드는 없었습니다. 박시니어 씨가 말했습니다.
"LLM 기반 코드는 테스트하기 어렵지만, 그래서 더 중요해. API 호출은 모킹하고, 로직은 철저하게 테스트해야 해." pytest는 Python에서 가장 많이 사용되는 테스트 프레임워크입니다.
간결한 문법과 강력한 기능을 제공합니다. unittest.mock은 외부 의존성을 가짜 객체로 대체하여 테스트를 가능하게 합니다.
LLM API 테스트의 핵심은 **모킹(mocking)**입니다. 실제 API를 호출하면 비용이 들고, 응답이 매번 달라서 테스트가 불안정해집니다.
대신 가짜 응답을 반환하도록 설정하고, 우리 코드가 그 응답을 제대로 처리하는지 확인합니다. 코드를 살펴보면, @patch.object 데코레이터로 client.chat.completions.create 메서드를 모킹합니다.
mock.return_value로 가짜 응답을 설정합니다. 실제 API 형식과 동일하게 choices 배열 안에 message.content가 있는 구조를 만듭니다.
테스트는 크게 두 종류로 나눕니다. 단위 테스트는 각 컴포넌트를 독립적으로 테스트합니다.
execute_step이 프롬프트를 제대로 만드는지, 응답을 제대로 파싱하는지 확인합니다. 통합 테스트는 전체 흐름을 테스트합니다.
여러 단계가 연결되어 제대로 동작하는지 확인합니다. 테스트 외에도 개선할 점이 많습니다.
재시도 로직을 추가하면 일시적인 API 오류에 대응할 수 있습니다. 타임아웃 설정으로 무한 대기를 방지합니다.
로깅을 상세화하면 문제 발생 시 원인을 빠르게 파악할 수 있습니다. 성능 개선도 고려해야 합니다.
독립적인 단계는 병렬 실행할 수 있습니다. 자주 사용되는 프롬프트의 결과는 캐싱할 수 있습니다.
입력 크기를 줄이면 비용을 절감할 수 있습니다. 김개발 씨는 테스트 코드를 추가하고, 여러 엣지 케이스를 확인했습니다.
빈 입력, 아주 긴 입력, 특수 문자가 포함된 입력 등을 테스트하며 에이전트를 더욱 견고하게 만들었습니다. "이제야 진짜 완성이네요." 김개발 씨가 뿌듯하게 말했습니다.
박시니어 씨가 웃으며 대답했습니다. "소프트웨어에 '완성'은 없어.
하지만 좋은 출발점을 만든 거야."
실전 팁
💡 - LLM 호출은 반드시 모킹하여 테스트하세요
- 엣지 케이스(빈 입력, 긴 입력, 특수 문자)를 꼭 테스트하세요
- CI/CD 파이프라인에 테스트를 포함시켜 자동화하세요
이상으로 학습을 마칩니다. 위 내용을 직접 코드로 작성해보면서 익혀보세요!
댓글 (0)
함께 보면 좋은 카드 뉴스
Helm 마이크로서비스 패키징 완벽 가이드
Kubernetes 환경에서 마이크로서비스를 효율적으로 패키징하고 배포하는 Helm의 핵심 기능을 실무 중심으로 학습합니다. Chart 생성부터 릴리스 관리까지 체계적으로 다룹니다.
보안 아키텍처 구성 완벽 가이드
프로젝트의 보안을 처음부터 설계하는 방법을 배웁니다. AWS 환경에서 VPC부터 WAF, 암호화, 접근 제어까지 실무에서 바로 적용할 수 있는 보안 아키텍처를 단계별로 구성해봅니다.
AWS Organizations 완벽 가이드
여러 AWS 계정을 체계적으로 관리하고 통합 결제와 보안 정책을 적용하는 방법을 실무 스토리로 쉽게 배워봅니다. 초보 개발자도 바로 이해할 수 있는 친절한 설명과 실전 예제를 제공합니다.
AWS KMS 암호화 완벽 가이드
AWS KMS(Key Management Service)를 활용한 클라우드 데이터 암호화 방법을 초급 개발자를 위해 쉽게 설명합니다. CMK 생성부터 S3, EBS 암호화, 봉투 암호화까지 실무에 필요한 모든 내용을 담았습니다.
AWS Secrets Manager 완벽 가이드
AWS에서 데이터베이스 비밀번호, API 키 등 민감한 정보를 안전하게 관리하는 Secrets Manager의 핵심 개념과 실무 활용법을 배워봅니다. 초급 개발자도 쉽게 따라할 수 있도록 실전 예제와 함께 설명합니다.