Agent와 Workflow
탐색적 작업에는 강하지만 반드시 분류 후 배정 같은 정해진 업무에는 고정 워크플로우가 맞다. createagent는 LLM이 자유롭게 도구를 골라 일했다. 이에 따라오는 단점이 있다면 예측 불가, 검증어려움, 비율 변동이 있다.
Workflow는 노드단위로 나뉘어 디버깅이 쉽고 어디서 잘못됬는지를 찾을 수 있다.
분류, 우선순위, 배정 각 단계가 분리되어 분류가 맞는데 우선순위가 틀렸다면 해당 영역만 고치면 된다 .
또 '분류'는 자연어 이해가 필요하니 LLM을 쓰지만 우선순위나 배정은 간단한 규칙이면 됨으로 LLM없이 코드로 해 빠르고, 공짜에 정확하다. https://standout.tistory.com/1918
Agent와 Workflow
Agent와 WorkflowLLM은 무엇을 언제할지 스스로 결정하는 자유로운 에이전트로 만족할까? 탐색적 작업에는 강하지만 반드시 분류 후 배정 같은 정해진 업무에는 고정 워크플로우가 맞다. createagent는
standout.tistory.com
| 구분 | Workflow | Agent |
| 실행 방식 | 미리 정해진 순서대로 실행 | LLM이 상황에 따라 다음 행동을 결정 |
| 흐름 | 고정(Deterministic) | 동적(Adaptive) |
| 도구 사용 | 개발자가 지정 | LLM이 필요할 때 선택 |
| 제어 | 코드가 제어 | LLM이 제어 |
| 예측 가능성 | 높음 | 상대적으로 낮음 |
| 대표 사례 | ETL, RAG 파이프라인 | AI 비서, 코딩 에이전트, 연구 에이전트 |
LangGraph
함수들을 그래프처럼 연결해서 실행순서를 관리해주는 프레임워크
LangGraph가 없다면 RAG를 직접만들때 점점 복잡해질경우 각 함수들을 연결하고, Tool이 추가되면 if문, while문, for문이 범벅이 될것이다 .코드를 if문으로 연결하지말고, 함수들을 연결하자 라는 아이디어.
StateGraph
LangGraph에서 여러 노드(Node)가 하나의 상태(State)를 공유하며 작업을 수행하도록 만드는 워크플로우 그래프
State
일반 함수는 result로 결과를 반환하지만 LLM 애플리케이션은 사용자질문, 이전대화, 검색결과, tool실행결과, 최종답변 등 여러정보를 계속 이어받아야함으로 하나의 공유 상태 State를 사용한다 .
state = {
"question": "...",
"documents": [...],
"answer": "...",
"messages": [...]
}https://standout.tistory.com/1919
LangGraph와 StateGraph(State, Node, Edge)
LangGraph함수들을 그래프처럼 연결해서 실행순서를 관리해주는 프레임워크함수들을 화살표로 연결한 것과 같다. LangGraph가 없다면 RAG를 직접만들때 점점 복잡해질경우 각 함수들을 연결하고, Tool
standout.tistory.com
프로세스를 이해해보자. 고객문의를 자동으로 분류해 우선순위를 결정하고 담당팀을 배정하는 workflow를 구현한다고 가정하자.
state는 워크플로우를 지나가면서 계속 채워지는 공용 데이터이다 . 초기에는 content밖에 없지만 nodes를 거치며 채워진다.
from typing import TypedDict
class TicketState(TypedDict):
content: str
category: str
priority: str
team: str
result: str{
"content": "결제가 두 번 됐어요."
}
분류 노드를 이해해보자.
def classify(state):
text = state["content"]
if "결제" in text:
category = "payment"
elif "로그인" in text:
category = "account"
else:
category = "etc"
return {
"category": category
}{
"content":"결제가 두 번 됐어요.",
"category":"payment"
}
def priority(state):
if state["category"] == "payment":
return {"priority":"high"}
return {"priority":"normal"}{
"content":"결제가 두 번 됐어요.",
"category":"payment",
"priority":"high"
}
def assign_team(state):
mapping = {
"payment":"결제팀",
"account":"계정팀",
"etc":"고객센터"
}
return {
"team": mapping[state["category"]]
}{
"content":"결제가 두 번 됐어요.",
"category":"payment",
"priority":"high",
"team":"결제팀"
}
샘플 코드로 이해해보자.
temperature = 0 모델실행시에 정확도를 위한 설정
try except 에러처리
if not 입력검증용도 raise 경고
if ~ elif ~ 를 활용해 분류
add_node() 노드 만들기
add_conditional_edges() 조건분기 priority에 따라 다음 노드가 결정되도록 한다 . if else를 직접쓰는대신에 사용
graph.compile() 지금까지 만든 그래프를 실행가능한 프로그램으로 만든다 .
invoke() 이 state를 가지고 Workflow를 처음부터 끝까지 실행하라
from typing import TypedDict
class State(TypedDict):
content: str
category: str
priority: str
team: str
def classify(state):
# LLM 호출 (temperature=0)
if "결제" in state["content"]:
return {"category": "payment"}
return {"category": "etc"}
def priority(state):
if state["category"] == "payment":
return {"priority": "high"}
return {"priority": "normal"}
def assign_team(state):
if state["category"] == "payment":
return {"team": "결제팀"}
return {"team": "고객센터"}
graph = StateGraph(State)
graph.add_node("classify", classify)
graph.add_node("priority", priority)
graph.add_node("assign", assign_team)
graph.add_edge(START, "classify")
graph.add_edge("classify", "priority")
graph.add_edge("priority", "assign")
graph.add_edge("assign", END)
app = graph.compile()
result = app.invoke({
"content": "결제가 두 번 됐어요."
})
print(result)
프로젝트 agent_workflow_console_project를 분석해보자.
common.py
import os, pathlib, dotenv
각 환결변수 설정
require_key 키 확인해 get_genai_client클라이언트 생성해 return.
get_chat() 사용자의 질문을 정규화해proviider에게 chat 모델 클래스 생성해 return
get_embeddings() 마찬가지로 임베딩 객체 생성해 return 임베딩은 텍스트를 숫자 벡터로 변환해주는 객체.
print_environment_status() 루트결고나 모델명 등의 정보를 출력
# -*- coding: utf-8 -*-
"""
common.py — 모든 실습에서 공통으로 사용하는 설정 모듈입니다.
주요 역할
1. 프로젝트 루트의 .env 파일을 읽습니다.
2. Gemini 또는 OpenAI용 LangChain 채팅 모델을 생성합니다.
3. 실습 데이터 폴더 경로를 한 곳에서 관리합니다.
4. API 키가 비어 있을 때 이해하기 쉬운 오류 메시지를 출력합니다.
"""
# 운영체제 환경변수를 읽기 위해 os 모듈을 가져옵니다.
import os
# 운영체제와 관계없이 파일 경로를 안전하게 처리하기 위해 pathlib을 가져옵니다.
import pathlib
# .env 파일의 환경변수를 현재 파이썬 프로세스에 로드하기 위해 load_dotenv를 가져옵니다.
from dotenv import load_dotenv
# 현재 common.py 파일은 프로젝트루트/code/common.py 위치에 있으므로 parent.parent가 프로젝트 루트입니다.
ROOT = pathlib.Path(__file__).resolve().parent.parent
# CSV 같은 실습 데이터를 저장하는 data 폴더 경로를 정의합니다.
DATA = ROOT / "data"
# 문서 기반 실습에서 사용할 수 있도록 data/docs 폴더 경로도 함께 정의합니다.
DOCS = DATA / "docs"
# 프로젝트 루트의 .env 파일을 읽어 API 키와 모델 설정을 환경변수에 등록합니다.
load_dotenv(ROOT / ".env")
# .env에 GEMINI_MODEL이 없으면 비교적 가벼운 기본 모델명을 사용합니다.
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
# .env에 OPENAI_MODEL이 없으면 교육용으로 비용이 비교적 낮은 기본 모델명을 사용합니다.
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
# Gemini 임베딩 실습에서 사용할 모델명을 환경변수 또는 기본값으로 읽습니다.
GEMINI_EMBED_MODEL = os.getenv("GEMINI_EMBED_MODEL", "models/gemini-embedding-001")
# OpenAI 임베딩 실습에서 사용할 모델명을 환경변수 또는 기본값으로 읽습니다.
OPENAI_EMBED_MODEL = os.getenv("OPENAI_EMBED_MODEL", "text-embedding-3-small")
def require_key(name: str) -> str:
"""지정한 API 키가 없거나 예시 문자열이면 프로그램을 안전하게 종료합니다."""
# 환경변수 이름으로 실제 저장된 값을 읽습니다.
value = os.getenv(name)
# 값이 없거나 예시 문구로 시작하면 아직 실제 키가 입력되지 않은 상태입니다.
if not value or value.startswith("여기에"):
# 사용자가 바로 조치할 수 있도록 설정 순서를 포함한 종료 메시지를 발생시킵니다.
raise SystemExit(
f"[설정 필요] {name} 값이 .env 파일에 없습니다.\n"
f"1) 프로젝트 루트의 .env.example 파일을 .env로 복사합니다.\n"
f"2) .env 파일을 열어 {name}에 실제 API 키를 입력합니다."
)
# 검증을 통과한 실제 API 키 문자열을 호출한 함수에 반환합니다.
return value
def get_genai_client():
"""Google의 공식 google-genai SDK 클라이언트를 생성하여 반환합니다."""
# google-genai 패키지에서 genai 모듈을 함수 실행 시점에 가져옵니다.
from google import genai
# 검증된 Google API 키를 사용해 Gemini API 클라이언트를 생성합니다.
return genai.Client(api_key=require_key("GOOGLE_API_KEY"))
def get_chat(provider: str = "gemini", temperature: float = 0.0):
"""provider 값에 따라 Gemini 또는 OpenAI LangChain ChatModel을 반환합니다."""
# 사용자가 대소문자를 섞어 입력해도 처리하도록 공급자명을 소문자로 정규화합니다.
normalized_provider = provider.strip().lower()
# Gemini가 선택된 경우 Google용 LangChain 통합 모델을 생성합니다.
if normalized_provider == "gemini":
# 모델을 만들기 전에 Google API 키가 정상인지 검증합니다.
require_key("GOOGLE_API_KEY")
# Gemini용 LangChain 채팅 모델 클래스를 가져옵니다.
from langchain_google_genai import ChatGoogleGenerativeAI
# .env에서 읽은 모델명과 호출자가 지정한 temperature로 모델 객체를 반환합니다.
return ChatGoogleGenerativeAI(model=GEMINI_MODEL, temperature=temperature)
# OpenAI가 선택된 경우 OpenAI용 LangChain 통합 모델을 생성합니다.
if normalized_provider == "openai":
# 모델을 만들기 전에 OpenAI API 키가 정상인지 검증합니다.
require_key("OPENAI_API_KEY")
# OpenAI용 LangChain 채팅 모델 클래스를 가져옵니다.
from langchain_openai import ChatOpenAI
# .env에서 읽은 모델명과 호출자가 지정한 temperature로 모델 객체를 반환합니다.
return ChatOpenAI(model=OPENAI_MODEL, temperature=temperature)
# 지원하지 않는 공급자 이름은 조용히 무시하지 않고 명확한 예외로 알립니다.
raise ValueError(f"알 수 없는 provider입니다: {provider} (허용값: gemini, openai)")
def get_embeddings(provider: str = "gemini"):
"""provider 값에 따라 Gemini 또는 OpenAI 임베딩 모델을 반환합니다."""
# 입력된 공급자 이름을 비교하기 쉬운 소문자로 변환합니다.
normalized_provider = provider.strip().lower()
# Gemini 임베딩 모델이 선택된 경우의 처리입니다.
if normalized_provider == "gemini":
# Google API 키가 설정되어 있는지 먼저 검사합니다.
require_key("GOOGLE_API_KEY")
# Gemini 임베딩용 LangChain 클래스를 가져옵니다.
from langchain_google_genai import GoogleGenerativeAIEmbeddings
# 저장 공간과 실습 속도를 고려하여 출력 차원을 768로 고정한 모델을 반환합니다.
return GoogleGenerativeAIEmbeddings(
model=GEMINI_EMBED_MODEL,
output_dimensionality=768,
)
# OpenAI 임베딩 모델이 선택된 경우의 처리입니다.
if normalized_provider == "openai":
# OpenAI API 키가 설정되어 있는지 먼저 검사합니다.
require_key("OPENAI_API_KEY")
# OpenAI 임베딩용 LangChain 클래스를 가져옵니다.
from langchain_openai import OpenAIEmbeddings
# .env에서 지정한 OpenAI 임베딩 모델 객체를 생성하여 반환합니다.
return OpenAIEmbeddings(model=OPENAI_EMBED_MODEL)
# 허용되지 않은 공급자 이름이면 즉시 오류를 발생시킵니다.
raise ValueError(f"알 수 없는 provider입니다: {provider} (허용값: gemini, openai)")
def print_environment_status() -> None:
"""프로젝트 경로, 모델명, API 키 로드 여부를 키 원문 노출 없이 출력합니다."""
# 프로젝트 루트의 절대 경로를 출력합니다.
print("ROOT :", ROOT)
# 데이터 폴더 경로와 실제 존재 여부를 함께 출력합니다.
print("DATA :", DATA, "(존재:", DATA.exists(), ")")
# 현재 선택될 Gemini 모델명을 출력합니다.
print("GEMINI_MODEL :", GEMINI_MODEL)
# 현재 선택될 OpenAI 모델명을 출력합니다.
print("OPENAI_MODEL :", OPENAI_MODEL)
# 실제 키 문자열은 노출하지 않고 설정 여부만 True 또는 False로 출력합니다.
print(
"API 키 로드 상태 :",
"GOOGLE_API_KEY=",
bool(os.getenv("GOOGLE_API_KEY")),
"/ OPENAI_API_KEY=",
bool(os.getenv("OPENAI_API_KEY")),
)
# common.py 파일을 직접 실행했을 때만 환경 점검 정보를 출력합니다.
if __name__ == "__main__":
# 다른 모듈에서 import할 때는 실행되지 않고 직접 실행할 때만 호출됩니다.
print_environment_status()
torch_demo.py
import torch
가중치를 float형으로 torch.tensor 설정하기
text_to_feature_tensor() feature 텐서로 묶어 반환하기 결제 단어가 포함될경우 0.0 으로 반환, 오류등의 실패표현이 포함되면 1, 환불 교환 표현이 있다면 1.0해 torch.tensor 형으로 묶어 반환한다.
calculate_priority_score() 중요도 확인하기 문장을 text_to_feature_tensor() 로 세개의 이진 특징값을 가진 텐서로 벼니환해 torch.dot() 로 가중치 텐서 계산을해 하나의 점수를 만들고, float형으로 변환한다. 점수가 3점이상이면 긴급으로 설정한다.
run_torch_demo() 사용자의 입력을 받아 사용하되 없으면 기본 예제를 사용한다. calculate_priority_score 각 텐서특징, 점수, 우선순위를 계산해 pritn
# -*- coding: utf-8 -*-
"""PyTorch 텐서를 사용하여 규칙 기반 우선순위 점수 계산 원리를 확인합니다."""
# 텐서 생성과 행렬 연산을 위해 PyTorch를 가져옵니다.
import torch
# 세 개의 특징을 긴급도 점수로 합산할 때 사용할 가중치를 정의합니다.
FEATURE_WEIGHTS = torch.tensor([3.0, 2.0, 1.0], dtype=torch.float32)
def text_to_feature_tensor(text: str) -> torch.Tensor:
"""티켓 문장을 결제·오류·환불 특징을 나타내는 0/1 텐서로 변환합니다."""
# 결제 관련 단어가 포함되면 1.0, 없으면 0.0으로 변환합니다.
payment_feature = 1.0 if "결제" in text else 0.0
# 오류 또는 작동 실패 표현이 포함되면 1.0, 없으면 0.0으로 변환합니다.
error_feature = 1.0 if any(word in text for word in ["오류", "안 돼", "안돼", "먹통"]) else 0.0
# 환불 또는 교환 표현이 포함되면 1.0, 없으면 0.0으로 변환합니다.
refund_feature = 1.0 if any(word in text for word in ["환불", "교환"]) else 0.0
# 세 특징을 float32 자료형의 1차원 텐서로 묶어 반환합니다.
return torch.tensor(
[payment_feature, error_feature, refund_feature],
dtype=torch.float32,
)
def calculate_priority_score(text: str) -> tuple[torch.Tensor, float, str]:
"""특징 텐서와 가중치의 내적으로 점수를 계산하고 우선순위 문자열을 반환합니다."""
# 입력 문장을 세 개의 이진 특징값을 가진 텐서로 변환합니다.
features = text_to_feature_tensor(text)
# 특징 텐서와 가중치 텐서의 내적을 계산하여 하나의 점수를 만듭니다.
score_tensor = torch.dot(features, FEATURE_WEIGHTS)
# 출력과 조건 비교에 사용하기 쉽도록 0차원 텐서를 파이썬 float로 변환합니다.
score = float(score_tensor.item())
# 점수가 3 이상이면 긴급으로 판정합니다.
if score >= 3.0:
priority = "긴급"
# 점수가 1 이상 3 미만이면 높음으로 판정합니다.
elif score >= 1.0:
priority = "높음"
# 특징이 하나도 없으면 보통으로 판정합니다.
else:
priority = "보통"
# 특징 텐서, 숫자 점수, 최종 우선순위를 함께 반환합니다.
return features, score, priority
def run_torch_demo() -> None:
"""콘솔에서 티켓 문장을 입력받아 PyTorch 기반 점수 계산 과정을 출력합니다."""
# 현재 설치된 PyTorch 버전을 출력합니다.
print(f"PyTorch 버전: {torch.__version__}")
# CUDA 사용 가능 여부를 확인하여 현재 실행 장치를 안내합니다.
print(f"CUDA 사용 가능: {torch.cuda.is_available()}")
# 사용자가 분석할 티켓 문장을 입력하도록 요청합니다.
text = input("티켓 내용을 입력하세요: ").strip()
# 빈 입력이면 이해하기 쉬운 기본 예제를 사용합니다.
if not text:
text = "결제가 안 돼요. 오류가 계속 발생합니다."
# 텐서 특징, 점수, 우선순위를 계산합니다.
features, score, priority = calculate_priority_score(text)
# 각 특징의 순서를 사용자가 이해할 수 있도록 출력합니다.
print("특징 순서 : [결제, 오류/작동실패, 환불/교환]")
# 실제 생성된 특징 텐서를 출력합니다.
print("특징 텐서 :", features)
# 가중치 텐서를 출력합니다.
print("가중치 텐서 :", FEATURE_WEIGHTS)
# 내적으로 얻은 숫자 점수를 출력합니다.
print("긴급도 점수 :", score)
# 점수 구간에 따라 결정된 최종 우선순위를 출력합니다.
print("판정 우선순위 :", priority)
rules.py
카테고리, word 긴급우선운위 단어목록, high_words 설정, team_map dict형으로 설정
normalize_category() 모델 output에서 공백과 줄바꿈을 제거해 카테고리와 일치하면 선택, 설명문장속에 카테고리가 섞여있으면 단어를 추출한다. 카테고리만 한단어로 답하라고 해도 LLM은 말을 잘 듣지않음으로공백제거, 카테고리랑 일치하는지 확인, 문장형으로 반환되었다면 category만 반환.
calculate_priority() 결제카테고리, 긴급이면 긴급판정, 환불교환은 높음, 외는 보통으로 판정한다.
calculate_team 카테고리가 있으면 팀을, 없으면 일반상담팀을 반환
calculate_route() 긴급한경우 escalate, 아닐경우에는 일반 assign
# -*- coding: utf-8 -*-
"""외부 API나 LangGraph 설치 없이도 테스트할 수 있는 순수 파이썬 규칙 모듈입니다."""
# LLM이 반환할 수 있는 허용 카테고리를 한 곳에 정의합니다.
CATEGORIES = ["결제", "배송", "환불", "교환", "회원", "기술지원", "기타"]
# 긴급 우선순위 판단에 사용할 핵심 단어 목록입니다.
URGENT_WORDS = ["결제", "오류", "안 돼", "안돼", "파손", "안와", "안 와", "먹통"]
# 높은 우선순위 판단에 사용할 핵심 단어 목록입니다.
HIGH_WORDS = ["환불", "교환", "튕", "취소"]
# 카테고리를 실제 담당팀으로 변환하기 위한 매핑표입니다.
TEAM_MAP = {
"결제": "결제지원팀",
"환불": "정산팀",
"교환": "물류팀",
"배송": "물류팀",
"회원": "회원관리팀",
"기술지원": "기술지원팀",
}
def normalize_category(raw_output: str) -> tuple[str, str]:
"""LLM 출력에서 허용 카테고리를 검증하고 보정 결과와 오류를 반환합니다."""
# 모델 출력의 앞뒤 공백과 줄바꿈을 제거합니다.
cleaned_output = raw_output.strip()
# 출력이 허용 목록의 단어와 정확히 일치하면 정상 결과로 채택합니다.
if cleaned_output in CATEGORIES:
return cleaned_output, ""
# 설명 문장 속에 허용 카테고리가 섞여 있으면 해당 단어를 추출합니다.
for category in CATEGORIES:
# 현재 카테고리 단어가 모델 출력에 포함되어 있는지 검사합니다.
if category in cleaned_output:
return category, ""
# 빈 문자열이거나 허용 범위 밖이면 오류 메시지에 사용할 미리보기를 만듭니다.
preview = cleaned_output[:30] if cleaned_output else "빈 응답"
# 안전한 기본값인 기타와 추적 가능한 검증 오류 메시지를 반환합니다.
return "기타", f"분류값 미허용('{preview}') → 기타 보정"
def calculate_priority(content: str, category: str) -> str:
"""티켓 내용과 카테고리를 이용하여 규칙 기반 우선순위를 계산합니다."""
# 결제 카테고리 또는 긴급 단어가 있으면 긴급으로 판정합니다.
if category == "결제" or any(word in content for word in URGENT_WORDS):
return "긴급"
# 환불·교환 카테고리 또는 높은 우선순위 단어가 있으면 높음으로 판정합니다.
if category in ("환불", "교환") or any(word in content for word in HIGH_WORDS):
return "높음"
# 그 외 일반 문의는 보통으로 판정합니다.
return "보통"
def calculate_team(category: str) -> str:
"""카테고리에 맞는 담당팀을 반환하고 미등록 카테고리는 일반상담팀으로 보냅니다."""
# 딕셔너리에 카테고리가 있으면 해당 팀을, 없으면 일반상담팀을 반환합니다.
return TEAM_MAP.get(category, "일반상담팀")
def calculate_route(priority: str) -> str:
"""우선순위에 따라 조건부 그래프가 이동할 경로 이름을 반환합니다."""
# 긴급은 에스컬레이션으로, 나머지는 일반 배정으로 라우팅합니다.
return "escalate" if priority == "긴급" else "assign"
workflow.py
import typing notrequired, typedict 정해진 딕셔너리 상태를 만들기 위한.
langchain_core.message, humanmessage, systemmessage langchain모델에 전달할 시스템메시지와 사용자 메세지.
langgraph.graph end, start, stategrapth 상태기반 그래프와 노드가져오기
common, rules 불러오기
TicketState() 사용자 입력원문, 카테고리, 우선순위 ,팀, 조건부 route, action, error 초기화
message_to_text() content에서 공백을 제거해 isinstance() 일부모델은 여러 콘텐츠 블록을 리스트로 반환, 텍스트만추출, 문자열로 변환해 return.
make_classify_node() temperature는 0으로 설정해 classify_node() state.get() 스테이트 값을 읽어 content가 없으면 기타로 분류, SystemMessage에 categroies를 할당에 메세지 작성, humanmessage 작성. 해 try llm.invoke() 응답을 정의한 message_to_text(), normalize_category()를 통해 카테고리목록과 error보정을 수행해 return.
priority_node() 내용르읽어 categrory로 읽되 없으면 기타로 처리, calculate_priority() 우선순위 계산해 return.
assign_node() 기타설정, team 계산해 return.
route_by_priority() 우선순위를 읽어 긴급경로, 일반 배정경로 return calculate_route()
escalate_node() return 팀, 경로명, 후속조치 문자열 반환
assign_route_node() 기존 assign_node 재사용하되 일반백정 경로명 추가, 후속조치 문장을 생성해 return. 팀, 경로, 조치가 담긴 상태가 된다 .
build_linear_workflow() StateGraph 생성, add_node, add_edge 후 compile()
build_conditional_workflow() 마찬가지로 수행하되add_conditional_edges() 수행해 우선순위 반환값에 따라 동적으로 선택하도록 한다.
# -*- coding: utf-8 -*-
"""LangGraph 기반 CS 티켓 처리 워크플로우의 핵심 기능을 정의합니다."""
# 타입이 정해진 딕셔너리 상태를 만들기 위해 TypedDict와 NotRequired를 가져옵니다.
from typing import NotRequired, TypedDict
# LangChain 채팅 모델에 전달할 시스템 메시지와 사용자 메시지를 가져옵니다.
from langchain_core.messages import HumanMessage, SystemMessage
# 상태 기반 그래프와 시작/종료 가상 노드를 가져옵니다.
from langgraph.graph import END, START, StateGraph
# 공통 모듈의 채팅 모델 생성 함수를 가져옵니다.
from common import get_chat
# API와 그래프에 의존하지 않는 순수 파이썬 규칙 함수를 가져옵니다.
from rules import (
CATEGORIES,
calculate_priority,
calculate_route,
calculate_team,
normalize_category,
)
class TicketState(TypedDict):
"""모든 노드가 공유하며 단계별 결과를 누적하는 상태 스키마입니다."""
# 사용자가 입력한 CS 티켓 원문입니다.
content: str
# 분류 노드가 채우는 카테고리 값입니다.
category: NotRequired[str]
# 우선순위 노드가 채우는 긴급도 값입니다.
priority: NotRequired[str]
# 배정 또는 에스컬레이션 노드가 채우는 담당팀입니다.
team: NotRequired[str]
# 조건부 분기 실습에서 어떤 경로를 선택했는지 기록합니다.
route: NotRequired[str]
# 조건부 분기 실습에서 후속 조치 내용을 기록합니다.
action: NotRequired[str]
# 검증 실패 또는 API 예외 내용을 기록합니다.
error: NotRequired[str]
def message_to_text(response: object) -> str:
"""LangChain 공급자별 응답 형식을 안전하게 일반 문자열로 변환합니다."""
# 대부분의 LangChain AIMessage는 content 속성에 최종 응답을 저장합니다.
content = getattr(response, "content", response)
# 일반 문자열이면 앞뒤 공백을 제거해 바로 반환합니다.
if isinstance(content, str):
return content.strip()
# 일부 모델은 여러 콘텐츠 블록을 리스트로 반환할 수 있습니다.
if isinstance(content, list):
# 최종 문자열 조각을 순서대로 모을 임시 리스트입니다.
text_parts: list[str] = []
# 리스트에 포함된 각 콘텐츠 블록을 차례로 확인합니다.
for item in content:
# 블록이 딕셔너리이고 text 키가 있으면 실제 텍스트만 추가합니다.
if isinstance(item, dict) and "text" in item:
text_parts.append(str(item["text"]))
# 다른 형식은 정보 유실을 막기 위해 문자열로 변환하여 추가합니다.
else:
text_parts.append(str(item))
# 여러 조각을 하나의 문장으로 합치고 앞뒤 공백을 제거합니다.
return " ".join(text_parts).strip()
# 예상하지 못한 형식도 프로그램이 중단되지 않도록 문자열로 변환합니다.
return str(content).strip()
def make_classify_node(provider: str):
"""선택한 LLM 공급자를 사용하는 분류 노드 함수를 생성합니다."""
# 같은 입력의 결과 변동을 줄이기 위해 temperature를 0으로 설정합니다.
llm = get_chat(provider=provider, temperature=0.0)
def classify_node(state: TicketState) -> dict:
"""티켓 내용을 LLM으로 분류하고 출력값을 허용 목록으로 검증합니다."""
# 티켓 내용이 없을 수도 있으므로 get과 기본값으로 안전하게 읽습니다.
content = state.get("content", "").strip()
# 빈 입력은 불필요한 API 호출 없이 기타로 보정하고 오류를 기록합니다.
if not content:
return {"category": "기타", "error": "빈 티켓 내용 → 기타 보정"}
# 모델의 역할과 출력 형식을 강제하는 시스템 메시지를 작성합니다.
system_message = SystemMessage(
content=(
f"다음 CS 티켓을 {CATEGORIES} 중 정확히 하나로 분류하세요. "
"설명 없이 카테고리 단어만 출력하세요."
)
)
# 실제 사용자가 입력한 티켓 내용을 HumanMessage로 작성합니다.
human_message = HumanMessage(content=content)
try:
# 시스템 메시지와 사용자 메시지를 모델에 전달하여 분류 결과를 받습니다.
response = llm.invoke([system_message, human_message])
# 공급자별 응답 구조 차이를 흡수하여 일반 문자열로 변환합니다.
raw_output = message_to_text(response)
# 허용 목록 검증과 부분 매칭 보정을 수행합니다.
category, error = normalize_category(raw_output)
# 이번 노드가 채운 필드만 반환하면 LangGraph가 기존 상태에 병합합니다.
return {"category": category, "error": error}
except Exception as exc:
# API 호출 실패가 전체 배치를 중단시키지 않도록 기타로 폴백합니다.
return {"category": "기타", "error": f"분류 API 예외: {type(exc).__name__}: {exc}"}
# 공급자가 연결된 실제 분류 노드 함수를 호출자에게 반환합니다.
return classify_node
def priority_node(state: TicketState) -> dict:
"""카테고리와 키워드를 이용해 LLM 없이 우선순위를 계산합니다."""
# 입력 티켓 내용을 안전하게 문자열로 읽습니다.
text = state.get("content", "")
# 이전 분류 노드가 채운 카테고리를 읽되 없으면 기타로 처리합니다.
category = state.get("category", "기타")
# 외부 의존성이 없는 공통 규칙 함수로 우선순위를 계산합니다.
priority = calculate_priority(text, category)
# 새로 계산한 priority 필드만 상태 업데이트 값으로 반환합니다.
return {"priority": priority}
def assign_node(state: TicketState) -> dict:
"""카테고리별 담당팀 매핑 규칙으로 팀을 배정합니다."""
# 분류 결과가 없거나 매핑에 없을 때를 대비해 기본값 기타를 사용합니다.
category = state.get("category", "기타")
# 외부 의존성이 없는 공통 규칙 함수로 담당팀을 계산합니다.
team = calculate_team(category)
# 배정 결과인 team 필드만 반환합니다.
return {"team": team}
def route_by_priority(state: TicketState) -> str:
"""우선순위를 읽어 긴급 경로 또는 일반 배정 경로 이름을 반환합니다."""
# 외부 의존성이 없는 공통 라우팅 함수로 다음 노드 이름을 결정합니다.
return calculate_route(state.get("priority", "보통"))
def escalate_node(state: TicketState) -> dict:
"""긴급 티켓을 긴급대응팀으로 즉시 에스컬레이션합니다."""
# 긴급 경로에서 필요한 담당팀, 경로명, 후속 조치를 한 번에 반환합니다.
return {
"team": "긴급대응팀",
"route": "긴급에스컬",
"action": "SLA 30분 내 즉시 에스컬레이션(담당자 호출)",
}
def assign_route_node(state: TicketState) -> dict:
"""일반 티켓을 카테고리 규칙에 따라 배정하고 SLA 정보를 추가합니다."""
# 중복 구현을 피하기 위해 기존 assign_node 함수를 재사용합니다.
update = assign_node(state)
# 조건부 분기 결과를 확인할 수 있도록 일반배정 경로명을 추가합니다.
update["route"] = "일반배정"
# 이미 계산된 팀 이름을 이용하여 후속 조치 문장을 생성합니다.
update["action"] = f"{update['team']} 일반 배정(SLA 24시간)"
# 팀, 경로, 조치가 담긴 부분 상태를 반환합니다.
return update
def build_linear_workflow(provider: str):
"""분류 → 우선순위 → 배정 순서가 고정된 선형 워크플로우를 만듭니다."""
# TicketState 스키마를 공유 상태로 사용하는 그래프 빌더를 생성합니다.
graph_builder = StateGraph(TicketState)
# 선택한 공급자를 사용하는 분류 노드를 classify라는 이름으로 등록합니다.
graph_builder.add_node("classify", make_classify_node(provider))
# 규칙 기반 우선순위 노드를 priority라는 이름으로 등록합니다.
graph_builder.add_node("priority", priority_node)
# 규칙 기반 팀 배정 노드를 assign이라는 이름으로 등록합니다.
graph_builder.add_node("assign", assign_node)
# 시작 지점 다음에 반드시 분류 노드가 실행되도록 연결합니다.
graph_builder.add_edge(START, "classify")
# 분류 다음에 반드시 우선순위 노드가 실행되도록 연결합니다.
graph_builder.add_edge("classify", "priority")
# 우선순위 다음에 반드시 배정 노드가 실행되도록 연결합니다.
graph_builder.add_edge("priority", "assign")
# 배정이 끝나면 워크플로우를 종료하도록 연결합니다.
graph_builder.add_edge("assign", END)
# 선언한 노드와 엣지를 실제 실행 가능한 그래프로 컴파일하여 반환합니다.
return graph_builder.compile()
def build_conditional_workflow(provider: str):
"""긴급 티켓과 일반 티켓이 서로 다른 경로를 타는 조건부 워크플로우를 만듭니다."""
# TicketState를 공유 상태로 사용하는 새 그래프 빌더를 생성합니다.
graph_builder = StateGraph(TicketState)
# LLM 기반 분류 노드를 등록합니다.
graph_builder.add_node("classify", make_classify_node(provider))
# 규칙 기반 우선순위 노드를 등록합니다.
graph_builder.add_node("priority", priority_node)
# 긴급 티켓을 처리할 에스컬레이션 노드를 등록합니다.
graph_builder.add_node("escalate", escalate_node)
# 일반 티켓을 처리할 일반 배정 노드를 등록합니다.
graph_builder.add_node("assign", assign_route_node)
# 시작에서 분류까지의 고정 엣지를 연결합니다.
graph_builder.add_edge(START, "classify")
# 분류에서 우선순위까지의 고정 엣지를 연결합니다.
graph_builder.add_edge("classify", "priority")
# priority 다음 노드는 route_by_priority의 반환값에 따라 동적으로 선택합니다.
graph_builder.add_conditional_edges(
"priority",
route_by_priority,
{"escalate": "escalate", "assign": "assign"},
)
# 긴급 경로 처리가 끝나면 그래프를 종료합니다.
graph_builder.add_edge("escalate", END)
# 일반 경로 처리가 끝나면 그래프를 종료합니다.
graph_builder.add_edge("assign", END)
# 조건부 분기가 포함된 실행 가능한 그래프를 반환합니다.
return graph_builder.compile()
main.py
import csv,
collections.abc callable 타입힌트에서 callable 사용하기 이 매개변수에는 함수 혹은 호출가능한 객체가 들어와야한다는 것을 나타내는 타입힌트. 여기에는 함수를 넘겨달라고 알려준다.
from collections.abc import Callable
def calculate(
a: int,
b: int,
func: Callable[[int, int], int]
):
return func(a, b)common, torch_demo, workflow 불러오기
LLM 공급자 저장
print_title() 타이틀 출력, choose_provider() 사용자 메뉴입력을 읽고1번시 gemini, 2시 openai로 설정
run_rule_nodes_demo() 우선순위 노드 호출, state.update, 담당팀 배정도 마찬가지. 실행결과
run_single_linear() 워크플로우 제목 출력, 사용자가 입력한 타켓 내용을 읽어 build_linear_workflow(), invoke()
# -*- coding: utf-8 -*-
"""OpenAI/Gemini와 LangGraph, PyTorch를 확인하는 콘솔 메뉴 프로그램입니다."""
# CSV 파일을 딕셔너리 형태로 읽기 위해 csv 모듈을 가져옵니다.
import csv
# 타입 힌트에서 Callable을 사용하기 위해 가져옵니다.
from collections.abc import Callable
# 공통 환경 점검 함수와 데이터 폴더 경로를 가져옵니다.
from common import DATA, print_environment_status
# PyTorch 실습 실행 함수를 가져옵니다.
from torch_demo import run_torch_demo
# 워크플로우 생성 함수와 개별 규칙 노드를 가져옵니다.
from workflow import (
assign_node,
build_conditional_workflow,
build_linear_workflow,
priority_node,
)
# 프로그램 실행 중 현재 선택된 LLM 공급자를 저장합니다.
CURRENT_PROVIDER = "gemini"
def print_title(title: str) -> None:
"""각 실습 화면의 제목을 일정한 형식으로 출력합니다."""
# 제목 위쪽 구분선을 출력합니다.
print("\n" + "=" * 78)
# 전달받은 제목을 출력합니다.
print(title)
# 제목 아래쪽 구분선을 출력합니다.
print("=" * 78)
def choose_provider() -> None:
"""Gemini 또는 OpenAI 중 이후 실습에서 사용할 공급자를 선택합니다."""
# 모듈 수준 공급자 변수를 변경할 것임을 파이썬에 알립니다.
global CURRENT_PROVIDER
# 공급자 선택 화면 제목을 출력합니다.
print_title("LLM 공급자 선택")
# 사용 가능한 첫 번째 공급자를 표시합니다.
print("1. Gemini API")
# 사용 가능한 두 번째 공급자를 표시합니다.
print("2. OpenAI API")
# 사용자의 메뉴 입력을 읽고 앞뒤 공백을 제거합니다.
selected = input("선택 번호 [현재: " + CURRENT_PROVIDER + "]: ").strip()
# 1번을 입력하면 Gemini를 현재 공급자로 지정합니다.
if selected == "1":
CURRENT_PROVIDER = "gemini"
# 2번을 입력하면 OpenAI를 현재 공급자로 지정합니다.
elif selected == "2":
CURRENT_PROVIDER = "openai"
# 빈 입력은 현재 값을 유지하고 다른 값은 안내 메시지만 출력합니다.
elif selected:
print("지원하지 않는 번호입니다. 기존 공급자를 유지합니다.")
# 최종 선택된 공급자를 출력합니다.
print("현재 공급자:", CURRENT_PROVIDER)
def run_rule_nodes_demo() -> None:
"""LLM 호출 없이 우선순위와 담당팀 규칙 노드를 각각 실행합니다."""
# 개별 노드 실행 화면 제목을 출력합니다.
print_title("규칙 기반 노드 개별 실행")
# 사용자가 티켓 내용을 입력하도록 요청합니다.
content = input("티켓 내용: ").strip() or "환불하고 싶습니다."
# 분류 결과는 이번 실습에서 사용자가 직접 선택하도록 안내합니다.
category = input("분류 카테고리 [기본: 환불]: ").strip() or "환불"
# 두 규칙 노드가 읽을 초기 상태 딕셔너리를 만듭니다.
state = {"content": content, "category": category}
# 우선순위 노드를 독립적으로 호출합니다.
priority_update = priority_node(state)
# 우선순위 결과를 기존 상태에 합칩니다.
state.update(priority_update)
# 담당팀 배정 노드를 독립적으로 호출합니다.
team_update = assign_node(state)
# 팀 배정 결과를 기존 상태에 합칩니다.
state.update(team_update)
# 각 노드가 채운 최종 상태를 출력합니다.
print("실행 결과:", state)
def run_single_linear() -> None:
"""선택한 LLM으로 한 건의 티켓을 선형 StateGraph에서 처리합니다."""
# 선형 워크플로우 화면 제목을 출력합니다.
print_title(f"선형 StateGraph 단일 처리 - {CURRENT_PROVIDER}")
# 사용자가 처리할 티켓 내용을 입력합니다.
content = input("티켓 내용: ").strip() or "결제가 안 돼요. 계속 오류가 납니다."
# 현재 공급자에 맞는 실행 가능한 선형 그래프를 생성합니다.
workflow = build_linear_workflow(CURRENT_PROVIDER)
# content만 가진 초기 상태를 워크플로우에 전달합니다.
result = workflow.invoke({"content": content})
# 분류, 우선순위, 담당팀이 누적된 최종 상태를 출력합니다.
print("최종 상태:", result)
def load_tickets() -> list[dict[str, str]]:
"""data/support_tickets.csv 파일을 읽어 티켓 딕셔너리 목록으로 반환합니다."""
# 공통 DATA 경로 아래의 CSV 파일 경로를 만듭니다.
csv_path = DATA / "support_tickets.csv"
# 엑셀에서 저장된 CSV의 BOM까지 안전하게 처리하도록 utf-8-sig로 파일을 엽니다.
with csv_path.open("r", encoding="utf-8-sig", newline="") as file:
# 첫 행의 열 이름을 키로 사용하는 DictReader를 생성합니다.
reader = csv.DictReader(file)
# 반복 가능한 reader의 모든 행을 리스트로 변환하여 반환합니다.
return list(reader)
def run_batch_linear() -> None:
"""CSV의 모든 티켓을 선형 워크플로우로 일괄 처리합니다."""
# 일괄 처리 화면 제목을 출력합니다.
print_title(f"CSV 선형 워크플로우 일괄 처리 - {CURRENT_PROVIDER}")
# 현재 공급자를 사용하는 선형 그래프를 한 번만 생성합니다.
workflow = build_linear_workflow(CURRENT_PROVIDER)
# CSV에서 실습용 티켓 목록을 읽습니다.
tickets = load_tickets()
# 결과 열의 제목을 일정한 너비로 출력합니다.
print(f"{'티켓':<8}{'분류':<10}{'우선순위':<10}{'담당팀':<14}{'상태'}")
# 표의 제목과 데이터를 구분하는 선을 출력합니다.
print("-" * 78)
# CSV에서 읽은 각 티켓을 하나씩 처리합니다.
for ticket in tickets:
# 현재 티켓 내용을 초기 상태로 전달하여 워크플로우를 실행합니다.
result = workflow.invoke({"content": ticket["content"]})
# 오류가 없으면 정상, 있으면 오류 메시지를 상태 열에 표시합니다.
status = result.get("error") or "정상"
# 한 티켓의 주요 결과를 한 줄로 정렬하여 출력합니다.
print(
f"{ticket['ticket_id']:<8}"
f"{result.get('category', ''):<10}"
f"{result.get('priority', ''):<10}"
f"{result.get('team', ''):<14}"
f"{status}"
)
def run_conditional() -> None:
"""긴급과 일반 티켓이 다른 노드로 이동하는 조건부 그래프를 실행합니다."""
# 조건부 분기 화면 제목을 출력합니다.
print_title(f"조건부 분기 워크플로우 - {CURRENT_PROVIDER}")
# 현재 공급자를 사용하는 조건부 그래프를 생성합니다.
workflow = build_conditional_workflow(CURRENT_PROVIDER)
# 사용자가 직접 입력하지 않으면 긴급 경로 예제를 사용합니다.
content = input("티켓 내용: ").strip() or "배송이 일주일째 안 와요."
# 초기 상태를 전달하여 분류와 우선순위 이후 분기 경로를 실행합니다.
result = workflow.invoke({"content": content})
# 최종 상태 전체를 출력하여 선택된 route와 action을 확인합니다.
print("최종 상태:", result)
def run_failure_demo() -> None:
"""빈 입력과 API 오류가 전체 흐름을 중단시키지 않는지 확인합니다."""
# 실패 처리 실습 화면 제목을 출력합니다.
print_title(f"노드 실패 처리와 상태 검증 - {CURRENT_PROVIDER}")
# 견고화 로직이 포함된 선형 그래프를 생성합니다.
workflow = build_linear_workflow(CURRENT_PROVIDER)
# 정상 입력과 빈 입력을 함께 준비합니다.
samples = [
("NORMAL", "회원 정보 변경 방법을 알려주세요."),
("EMPTY", ""),
]
# 두 입력을 차례로 실행하여 한 건의 오류가 다음 실행에 영향을 주지 않는지 확인합니다.
for ticket_id, content in samples:
# 현재 샘플을 워크플로우에 전달합니다.
result = workflow.invoke({"content": content})
# error 필드가 비어 있으면 정상으로 표시합니다.
status = result.get("error") or "정상"
# 보정된 분류, 우선순위, 담당팀, 오류 기록을 출력합니다.
print(
ticket_id,
result.get("category"),
result.get("priority"),
result.get("team"),
status,
)
def run_provider_comparison() -> None:
"""같은 티켓을 Gemini와 OpenAI에 각각 전달하여 분류 결과를 비교합니다."""
# 두 공급자 비교 화면 제목을 출력합니다.
print_title("Gemini API와 OpenAI API 결과 비교")
# 두 모델에 동일하게 전달할 티켓 내용을 입력합니다.
content = input("티켓 내용: ").strip() or "앱이 자꾸 튕기고 로그인이 풀립니다."
# 비교할 공급자 이름을 고정된 순서로 순회합니다.
for provider in ("gemini", "openai"):
try:
# 현재 비교 대상 공급자의 선형 그래프를 생성합니다.
workflow = build_linear_workflow(provider)
# 동일한 입력으로 워크플로우를 실행합니다.
result = workflow.invoke({"content": content})
# 공급자 이름과 주요 결과를 출력합니다.
print(provider, "→", result)
except SystemExit as exc:
# 해당 공급자의 API 키가 없더라도 다른 공급자 비교는 계속합니다.
print(provider, "→ 설정 오류:", exc)
def pause() -> None:
"""사용자가 결과를 읽은 뒤 메뉴로 돌아가도록 Enter 입력을 기다립니다."""
# 콘솔 출력이 바로 사라지지 않도록 Enter 입력을 받습니다.
input("\nEnter를 누르면 메뉴로 돌아갑니다...")
def print_menu() -> None:
"""실행 가능한 실습 기능만 메뉴로 출력합니다."""
# 프로그램 이름과 현재 선택된 공급자를 출력합니다.
print_title(f"Agent Workflow 콘솔 실습 앱 (현재 LLM: {CURRENT_PROVIDER})")
# 환경변수와 공통 경로 확인 기능을 표시합니다.
print("1. 공통 환경 및 API 키 로드 상태 확인")
# 이후 LLM 실습에서 사용할 공급자 변경 기능을 표시합니다.
print("2. LLM 공급자 선택 (Gemini / OpenAI)")
# PyTorch 텐서 연산 실습을 표시합니다.
print("3. PyTorch 기반 티켓 우선순위 점수 확인")
# LLM 없이 규칙 노드만 개별 실행하는 기능을 표시합니다.
print("4. 규칙 기반 노드 개별 실행")
# 한 건의 티켓을 선형 그래프로 실행하는 기능을 표시합니다.
print("5. StateGraph 선형 워크플로우 단일 처리")
# CSV 전체를 선형 그래프로 일괄 처리하는 기능을 표시합니다.
print("6. CSV 티켓 일괄 처리")
# 우선순위에 따라 실행 노드가 달라지는 기능을 표시합니다.
print("7. add_conditional_edges 조건부 분기")
# 빈 입력과 API 오류 보정 기능을 표시합니다.
print("8. 노드 실패 처리와 상태 검증")
# Gemini와 OpenAI의 동일 입력 결과 비교 기능을 표시합니다.
print("9. Gemini / OpenAI 결과 비교")
# 프로그램 종료 기능을 표시합니다.
print("0. 종료")
def main() -> None:
"""사용자가 종료할 때까지 콘솔 메뉴를 반복 실행합니다."""
# 메뉴 번호와 실행 함수의 연결 관계를 딕셔너리로 정의합니다.
menu_actions: dict[str, Callable[[], None]] = {
"1": print_environment_status,
"2": choose_provider,
"3": run_torch_demo,
"4": run_rule_nodes_demo,
"5": run_single_linear,
"6": run_batch_linear,
"7": run_conditional,
"8": run_failure_demo,
"9": run_provider_comparison,
}
# 사용자가 0번을 선택하기 전까지 메뉴를 계속 표시합니다.
while True:
# 현재 메뉴 항목을 화면에 출력합니다.
print_menu()
# 사용자가 선택한 번호를 문자열로 읽습니다.
selected = input("메뉴 번호를 선택하세요: ").strip()
# 0번이면 반복문을 종료하여 프로그램을 끝냅니다.
if selected == "0":
print("프로그램을 종료합니다.")
break
# 입력한 번호와 연결된 실행 함수를 딕셔너리에서 찾습니다.
action = menu_actions.get(selected)
# 존재하지 않는 번호라면 오류 안내 후 메뉴를 다시 표시합니다.
if action is None:
print("지원하지 않는 메뉴 번호입니다.")
pause()
continue
try:
# 선택한 메뉴의 실제 실습 함수를 실행합니다.
action()
except SystemExit as exc:
# API 키 미설정 같은 설정 오류를 전체 프로그램 종료 없이 안내합니다.
print(exc)
except FileNotFoundError as exc:
# 필요한 CSV가 없을 때 어떤 파일이 누락됐는지 출력합니다.
print("필요한 파일을 찾지 못했습니다:", exc)
except Exception as exc:
# 예상하지 못한 오류도 메뉴 루프가 종료되지 않도록 포착합니다.
print(f"[실행 오류] {type(exc).__name__}: {exc}")
# 실행 결과를 확인한 뒤 메뉴로 돌아갈 수 있도록 대기합니다.
pause()
# 이 파일을 직접 실행할 때만 main 함수를 호출합니다.
if __name__ == "__main__":
# PyCharm에서 code/main.py를 실행하면 콘솔 메뉴가 시작됩니다.
main()
또 다른 프로젝트 langgraph_console_learning_project를 분석해보자 .
basic_graph.py
import typing TypeDict, langraph.graph end, start, stategraph 가져오기
BasicState() 사용자 입력한 문자, 길이, 출력 result 저장
measure_node 길이구하기
format_node 포맷에 맞춰 text표현
def build_basic_graph() stateGraph만들고 add_node, add_edge, builder.compile() 수행.
run_demo() 앞서 작성한 함수 실행해 graph 만들기, basicstate 초기설정 정의 및 invoke.
# -*- coding: utf-8 -*-
"""State, Node, Edge의 기본 구조를 확인하는 선형 그래프 예제입니다."""
# TypedDict는 상태 딕셔너리의 키와 타입을 선언할 때 사용합니다.
from typing import TypedDict
# StateGraph는 상태 기반 그래프를 생성합니다.
# START와 END는 그래프의 시작점과 종료점을 나타냅니다.
from langgraph.graph import END, START, StateGraph
class BasicState(TypedDict):
"""그래프의 모든 노드가 공유하는 상태 구조입니다."""
# 사용자가 입력한 원본 문장을 저장합니다.
text: str
# 입력 문장의 문자 수를 저장합니다.
length: int
# 최종 출력 문장을 저장합니다.
result: str
def measure_node(state: BasicState) -> dict:
"""입력 문장의 문자 수를 계산합니다."""
# 공유 상태에서 원본 문장을 읽습니다.
text = state["text"]
# len 함수를 사용해 문자열 길이를 계산합니다.
length = len(text)
# 변경한 length 값만 반환하면 LangGraph가 기존 상태에 병합합니다.
return {"length": length}
def format_node(state: BasicState) -> dict:
"""원본 문장과 길이를 최종 결과 문자열로 구성합니다."""
# 기존 상태에 유지된 원본 문장을 읽습니다.
text = state["text"]
# 앞 노드가 계산한 길이를 읽습니다.
length = state["length"]
# 사용자에게 보여 줄 최종 문자열을 만듭니다.
result = f"입력 문장: {text} / 문자 수: {length}"
# 새로 만든 result 값만 반환합니다.
return {"result": result}
def build_basic_graph():
"""START → measure → format → END 그래프를 생성합니다."""
# BasicState를 공유 상태로 사용하는 그래프 빌더를 만듭니다.
builder = StateGraph(BasicState)
# measure라는 이름으로 문자 수 계산 노드를 등록합니다.
builder.add_node("measure", measure_node)
# format이라는 이름으로 결과 구성 노드를 등록합니다.
builder.add_node("format", format_node)
# 그래프 시작 후 measure 노드가 실행되도록 연결합니다.
builder.add_edge(START, "measure")
# measure 다음에 format 노드가 실행되도록 연결합니다.
builder.add_edge("measure", "format")
# format 실행 후 그래프가 종료되도록 연결합니다.
builder.add_edge("format", END)
# 정의한 그래프를 실행 가능한 객체로 컴파일합니다.
return builder.compile()
def run_demo() -> None:
"""기본 그래프를 실행하고 최종 상태를 출력합니다."""
# 실행 가능한 기본 그래프를 생성합니다.
graph = build_basic_graph()
# 그래프 실행에 사용할 초기 상태를 준비합니다.
initial_state: BasicState = {
"text": "LangGraph를 학습합니다.",
"length": 0,
"result": "",
}
# invoke로 그래프를 동기 실행합니다.
final_state = graph.invoke(initial_state)
# 모든 노드 실행이 끝난 최종 상태를 출력합니다.
print("최종 상태:", final_state)
# 최종 결과 문자열만 별도로 출력합니다.
print("최종 결과:", final_state["result"])
checkpoint_graph.py
import langchain_core.message aimessage, humanmessage 대화메세지 객체 생성
langraph.checkpoint.memorty inmemorysaver 상태를 메모리에 저장하도록함
langraph.graph end, start, messagestate, stategraph messagestate와 그래프 구성요소를 가져온다 .MessageState는 LangGraph에서 채팅(메시지) 기반 애플리케이션을 만들기 위해 미리 정의해둔 State이고, State는 네가 직접 정의하는 일반적인 상태. LangGraph에서는 대화형 AI를 너무 많이 만들다 보니 아예 메시지 저장용 State를 제공한것. 일반 state라면 class State(TypedDict): messages: list 직접 만들어야하지만 messagestate은 state안에 messages 형태로 이미 들어가있다. { "messages": [ HumanMessage(...), AIMessage(...), ToolMessage(...) ] } 뿐만아니라 이 messages는 자동으로 기존 message, 새로운 message를 합쳐준다. 자동으로 메세지를 누적하면서도 필요한 추가상태를 관리할수있으니 편리하다.
memory_node() messagestate를 받아 len을 구하고 aimessage ai 응답을 생성해 return.
build_checkpoint_graph() InMemorySaver()로 상태를 메모리에 저장할 checkpointer를 만들고 messagestate기반 stategraph를 만든다. add.node연결, add_edge 실행후 그래프 end 종료, builder.compile() 시 checkpoiner를 전달해 상태 저장기능을 활성화한다. builder.compile()에서 checkpoiner를 전달하지않으면 상태가 영구적으로 저장되지않는다 . 그래프가 실행되는 동안에는 노드사이에서 state가 전달되다가 실행이 끝나면 사라지고 다음에 다시 graph.invoke()시 처음부터 시작한다.
run_demo() build_checkpoint_graph() 생성해 config 설정. 이때 대화세션을 구분할 thread_id를 명시한다. graph.invoke() 두번실행 순서대로 메세지 출력
# -*- coding: utf-8 -*-
"""InMemorySaver와 thread_id를 이용한 체크포인트 예제입니다."""
# AIMessage와 HumanMessage는 대화 메시지 객체를 생성합니다.
from langchain_core.messages import AIMessage, HumanMessage
# InMemorySaver는 프로그램 실행 중 상태를 메모리에 저장합니다.
from langgraph.checkpoint.memory import InMemorySaver
# MessagesState와 그래프 구성 요소를 가져옵니다.
from langgraph.graph import END, START, MessagesState, StateGraph
def memory_node(state: MessagesState) -> dict:
"""현재 thread에 저장된 전체 메시지 수를 응답합니다."""
# Checkpointer가 복원한 과거 메시지를 포함해 전체 개수를 계산합니다.
message_count = len(state["messages"])
# 현재 메시지 개수를 설명하는 AI 응답을 생성합니다.
answer = AIMessage(content=f"현재 thread에 저장된 전체 메시지는 {message_count}개입니다.")
# 생성한 AI 응답을 messages에 누적하도록 반환합니다.
return {"messages": [answer]}
def build_checkpoint_graph():
"""메모리 Checkpointer가 적용된 메시지 그래프를 생성합니다."""
# 상태를 메모리에 저장할 Checkpointer를 만듭니다.
memory = InMemorySaver()
# MessagesState 기반 그래프 빌더를 생성합니다.
builder = StateGraph(MessagesState)
# 메모리 확인 노드를 등록합니다.
builder.add_node("memory", memory_node)
# START에서 memory 노드로 연결합니다.
builder.add_edge(START, "memory")
# memory 실행 후 그래프를 종료합니다.
builder.add_edge("memory", END)
# compile 시 checkpointer를 전달해 상태 저장 기능을 활성화합니다.
graph = builder.compile(checkpointer=memory)
# 그래프와 Checkpointer를 함께 반환합니다.
return graph, memory
def run_demo() -> None:
"""같은 thread_id로 두 번 실행해 과거 상태 복원을 확인합니다."""
# 체크포인트 그래프를 생성합니다.
graph, _memory = build_checkpoint_graph()
# 동일한 대화 세션을 구분할 thread_id를 설정합니다.
config = {"configurable": {"thread_id": "langgraph-learning-thread"}}
# 첫 번째 사용자 메시지를 실행합니다.
first_result = graph.invoke(
{"messages": [HumanMessage(content="첫 번째 질문입니다.")]},
config=config,
)
# 첫 번째 실행의 마지막 AI 응답을 출력합니다.
print("첫 번째 실행:", first_result["messages"][-1].content)
# 같은 thread_id로 두 번째 사용자 메시지를 실행합니다.
second_result = graph.invoke(
{"messages": [HumanMessage(content="이전 상태를 기억하나요?")]},
config=config,
)
# 두 번째 실행에서는 이전 메시지가 복원되어 전체 개수가 증가합니다.
print("두 번째 실행:", second_result["messages"][-1].content)
# 최종 메시지 전체를 출력합니다.
print("\n최종 메시지 목록:")
# 최종 상태의 메시지를 순서대로 출력합니다.
for index, message in enumerate(second_result["messages"], start=1):
print(f"{index}. {type(message).__name__}: {message.content}")
conditional_graph.py
import typing, langraph.graph
RouterState 질문, 카테고리, 답변 정의
classify_node() 질문을 state에서 가져와 정책 질문 키워드 정의를 해 for문으로 돌려 category 설정
return.
route_question() 전달받은 state중 category를 return한다 .
policy_node(), sales_node() 질문을 answer에 넣어 반환
build_conditional_graph() stategraph를 만들어 add_node, add_edge, add_conditional_edges() 즉 마지막 add_node의 결과에 따라 policy, sales로 분기하는것. add_edge(), bulider.compile()
run_demo() 분기 그래프를 생성해 questions를 정의하고 for문을 돌려 질문마다 새로운 초기상태 invoke().해 retun result
# -*- coding: utf-8 -*-
"""add_conditional_edges를 이용한 조건부 분기 예제입니다."""
# TypedDict는 상태 딕셔너리 구조를 선언할 때 사용합니다.
from typing import TypedDict
# LangGraph의 그래프 생성 요소를 가져옵니다.
from langgraph.graph import END, START, StateGraph
class RouterState(TypedDict):
"""질문 분류 그래프가 공유하는 상태입니다."""
# 사용자가 입력한 질문을 저장합니다.
question: str
# policy 또는 sales 분류 결과를 저장합니다.
category: str
# 선택된 처리 노드의 답변을 저장합니다.
answer: str
def classify_node(state: RouterState) -> dict:
"""질문을 정책 또는 상품 추천으로 분류합니다."""
# 현재 상태에서 질문을 가져옵니다.
question = state["question"]
# 정책 질문을 판단할 키워드를 정의합니다.
policy_words = ["환불", "교환", "배송", "취소", "적립", "포인트"]
# 정책 키워드가 하나라도 포함되었는지 검사합니다.
is_policy = any(word in question for word in policy_words)
# 정책이면 policy, 아니면 sales를 선택합니다.
category = "policy" if is_policy else "sales"
# 분류 결과만 상태에 반영하도록 반환합니다.
return {"category": category}
def route_question(state: RouterState) -> str:
"""category 값에 따라 다음 노드 이름을 반환합니다."""
# 앞 노드가 저장한 category를 분기 키로 사용합니다.
return state["category"]
def policy_node(state: RouterState) -> dict:
"""정책 질문을 처리하는 전문 노드입니다."""
# 실제 프로젝트에서는 FAQ 또는 Vector DB 검색을 이 위치에 넣을 수 있습니다.
answer = f"정책 담당 노드가 처리했습니다: {state['question']}"
# 생성한 답변을 상태에 저장합니다.
return {"answer": answer}
def sales_node(state: RouterState) -> dict:
"""상품 추천 질문을 처리하는 전문 노드입니다."""
# 실제 프로젝트에서는 상품 DB 검색이나 추천 모델을 이 위치에 넣을 수 있습니다.
answer = f"상품 추천 담당 노드가 처리했습니다: {state['question']}"
# 생성한 답변을 상태에 저장합니다.
return {"answer": answer}
def build_conditional_graph():
"""질문 분류 결과에 따라 서로 다른 노드로 이동하는 그래프를 생성합니다."""
# RouterState를 공유 상태로 사용하는 그래프 빌더를 생성합니다.
builder = StateGraph(RouterState)
# 질문 분류 노드를 등록합니다.
builder.add_node("classify", classify_node)
# 정책 처리 노드를 등록합니다.
builder.add_node("policy", policy_node)
# 상품 추천 처리 노드를 등록합니다.
builder.add_node("sales", sales_node)
# START에서 classify 노드로 연결합니다.
builder.add_edge(START, "classify")
# classify 결과에 따라 policy 또는 sales로 분기합니다.
builder.add_conditional_edges(
"classify",
route_question,
{"policy": "policy", "sales": "sales"},
)
# policy 노드 실행 후 종료합니다.
builder.add_edge("policy", END)
# sales 노드 실행 후 종료합니다.
builder.add_edge("sales", END)
# 그래프를 실행 가능한 객체로 컴파일합니다.
return builder.compile()
def run_demo() -> None:
"""정책 질문과 추천 질문을 각각 실행해 분기 결과를 확인합니다."""
# 조건부 분기 그래프를 생성합니다.
graph = build_conditional_graph()
# 서로 다른 경로를 확인하기 위한 질문 목록을 준비합니다.
questions = [
"환불은 며칠 이내에 가능한가요?",
"가성비 좋은 전자기기를 추천해 주세요.",
]
# 질문을 하나씩 그래프에 전달합니다.
for question in questions:
# 질문마다 새로운 초기 상태를 만듭니다.
result = graph.invoke({"question": question, "category": "", "answer": ""})
# 원본 질문을 출력합니다.
print("\n질문:", question)
# 분류 결과를 출력합니다.
print("분류:", result["category"])
# 최종 답변을 출력합니다.
print("답변:", result["answer"])
error_graph.py
import typing typedict
langgraph.graph의 end, start, stategraph.
ErrorState() 입력값, 결과, 에러 정의
risky_node() try, value를 upper()해 return.
route_after_risky() state에 error값이 있으면 fallback 로 경로지정. 아니면 success로 경로를 지정한다.
success_node result, fallback_node 출력.
build_error_graph() stategraph객체를 만들어 add)node, add_edge, add_conditional_edges, add_edge를 거쳐 builder.compile().
run_demo() build_error_graph()를 실행하고 values를 가져와 for문으로 돌려가며 graph.invoke() 후 결과 print.
# -*- coding: utf-8 -*-
"""노드 예외 처리와 fallback 분기 예제입니다."""
# TypedDict는 오류 처리 상태 구조를 선언합니다.
from typing import TypedDict
# LangGraph 그래프 구성 요소를 가져옵니다.
from langgraph.graph import END, START, StateGraph
class ErrorState(TypedDict):
"""입력값, 처리 결과, 오류 메시지를 공유하는 상태입니다."""
# 처리할 원본 입력값을 저장합니다.
value: str
# 정상 처리 또는 fallback 결과를 저장합니다.
result: str
# 발생한 오류 메시지를 저장합니다.
error: str
def risky_node(state: ErrorState) -> dict:
"""빈 문자열 입력 시 오류 상태를 반환합니다."""
try:
# 상태에서 입력 문자열을 읽습니다.
value = state["value"]
# 공백을 제거한 값이 비어 있으면 예외를 발생시킵니다.
if not value.strip():
raise ValueError("빈 문자열은 처리할 수 없습니다.")
# 정상 입력은 대문자로 변환합니다.
result = value.upper()
# 정상 결과와 빈 오류 메시지를 반환합니다.
return {"result": result, "error": ""}
except Exception as error:
# 예외를 다시 던지지 않고 상태의 error 필드에 기록합니다.
return {"result": "", "error": f"risky_node 오류: {error}"}
def route_after_risky(state: ErrorState) -> str:
"""오류 발생 여부에 따라 success 또는 fallback 경로를 선택합니다."""
# error 값이 존재하면 fallback 경로를 선택합니다.
if state["error"]:
return "fallback"
# 오류가 없으면 success 경로를 선택합니다.
return "success"
def success_node(state: ErrorState) -> dict:
"""정상 처리 결과를 최종 문장으로 만듭니다."""
# 앞 노드의 결과를 포함한 성공 메시지를 생성합니다.
result = f"정상 처리 완료: {state['result']}"
# 최종 결과를 상태에 저장합니다.
return {"result": result}
def fallback_node(state: ErrorState) -> dict:
"""오류 발생 시 안전한 기본 결과를 생성합니다."""
# 오류 내용을 포함하는 fallback 메시지를 만듭니다.
result = f"안전한 기본 처리로 전환했습니다. 원인: {state['error']}"
# fallback 결과를 상태에 저장합니다.
return {"result": result}
def build_error_graph():
"""오류 상태에 따라 success 또는 fallback으로 분기하는 그래프를 생성합니다."""
# ErrorState를 공유 상태로 사용하는 그래프 빌더를 생성합니다.
builder = StateGraph(ErrorState)
# 위험 처리, 정상 처리, fallback 처리 노드를 등록합니다.
builder.add_node("risky", risky_node)
builder.add_node("success", success_node)
builder.add_node("fallback", fallback_node)
# START에서 risky 노드로 연결합니다.
builder.add_edge(START, "risky")
# risky 결과에 따라 success 또는 fallback으로 분기합니다.
builder.add_conditional_edges(
"risky",
route_after_risky,
{"success": "success", "fallback": "fallback"},
)
# 두 경로 모두 처리 후 그래프를 종료합니다.
builder.add_edge("success", END)
builder.add_edge("fallback", END)
# 실행 가능한 그래프로 컴파일합니다.
return builder.compile()
def run_demo() -> None:
"""정상 입력과 빈 입력을 각각 실행합니다."""
# 오류 처리 그래프를 생성합니다.
graph = build_error_graph()
# 정상 입력과 오류 입력을 준비합니다.
values = ["langgraph", ""]
# 각 값을 그래프에 전달합니다.
for value in values:
# 입력마다 새로운 초기 상태를 만듭니다.
result = graph.invoke({"value": value, "result": "", "error": ""})
# 실제 입력값을 repr 형태로 출력합니다.
print(f"\n입력값: {value!r}")
# 최종 결과를 출력합니다.
print("결과:", result["result"])
# 오류가 없으면 '없음'을 출력합니다.
print("오류:", result["error"] or "없음")
loop_graph.py
import typing typedict
langgraph.graph의 end, start, stategraph자겨오기
LoopState 카운트와 메세지 설정
increase_node() 기존 state에서의 count값에 +1, 메세지 출력
should_continue() 3보다 작으면 repeat, 이상이라면 finish.
build_loop_graph() stategraph를 만들고, add_node, add_edge, add_conditional_edges수행후 compile().
run_demo() build_loop_graph() 를 수행한 뒤 loopstate를 0으로 설정해 세팅. for문으로 stram()을 사용해서 노드가 하나 실행될때마다 결과를 바로 반환되게한다. count가 1이면 두번째 실행, 0이면 첫번째 노드실행.. stream()은 invoke()와 다르게 결과만 알려주지않고 작업하는 모습을 실시간으로 보여준다. stream()을 쓰면 노드 하나가 끝날때마다 결과를 확인해 현재 그래프가 어디까지 실행되었는지를 계속 확인할 수 있다. stream()ㅁ은 여러개의 값을 순차적으로 반환하는 함수이고, 이를 for문으로 반환하면 확인할 수 있다. invoke() 최종확인.
# -*- coding: utf-8 -*-
"""조건부 엣지를 이용한 반복과 종료 조건 예제입니다."""
# TypedDict는 반복 그래프 상태 구조를 선언합니다.
from typing import TypedDict
# LangGraph 그래프 구성 요소를 가져옵니다.
from langgraph.graph import END, START, StateGraph
class LoopState(TypedDict):
"""반복 횟수와 메시지를 저장하는 상태입니다."""
# 현재 반복 횟수를 저장합니다.
count: int
# 현재 반복 처리 상태를 저장합니다.
message: str
def increase_node(state: LoopState) -> dict:
"""count 값을 1 증가시킵니다."""
# 기존 count 값에 1을 더합니다.
next_count = state["count"] + 1
# 증가한 횟수를 설명하는 문자열을 만듭니다.
message = f"{next_count}번째 반복 실행이 완료되었습니다."
# 변경된 값을 상태 업데이트 결과로 반환합니다.
return {"count": next_count, "message": message}
def should_continue(state: LoopState) -> str:
"""count가 3 미만이면 반복하고, 아니면 종료합니다."""
# count가 3보다 작으면 repeat 경로를 선택합니다.
if state["count"] < 3:
return "repeat"
# count가 3 이상이면 finish 경로를 선택합니다.
return "finish"
def build_loop_graph():
"""최대 3회 반복하는 그래프를 생성합니다."""
# LoopState를 사용하는 그래프 빌더를 생성합니다.
builder = StateGraph(LoopState)
# 반복 실행할 increase 노드를 등록합니다.
builder.add_node("increase", increase_node)
# START에서 increase로 연결합니다.
builder.add_edge(START, "increase")
# increase 실행 후 반복 또는 종료 경로를 선택합니다.
builder.add_conditional_edges(
"increase",
should_continue,
{"repeat": "increase", "finish": END},
)
# 실행 가능한 그래프로 컴파일합니다.
return builder.compile()
def run_demo() -> None:
"""stream과 invoke로 반복 그래프를 확인합니다."""
# 반복 그래프를 생성합니다.
graph = build_loop_graph()
# count를 0으로 설정한 초기 상태를 준비합니다.
initial_state: LoopState = {"count": 0, "message": ""}
# stream을 사용해 각 노드의 중간 이벤트를 출력합니다.
for event in graph.stream(initial_state):
# 실행된 노드와 해당 노드의 반환값을 확인합니다.
print("중간 이벤트:", event)
# invoke를 사용해 최종 상태를 별도로 확인합니다.
final_state = graph.invoke(initial_state)
# 반복이 count 3에서 종료됐는지 출력합니다.
print("최종 상태:", final_state)
messages_graph.py
import lanchain_core.message, aimessage, humanmessage
langraph.graph end, start, messagestate, stategraph
echo_node() messagestates에서 마지막 메세지를 가져오고 반환
build_messages_graph() messagestate에 add_node echo 추가
add_edge start, end 연결해 builder.compile().
run_demo() build_message_graph() 생성해 invoke() 초기상태로 그래프를 실행, message목록을 for문으로 출력한다.
# -*- coding: utf-8 -*-
"""MessagesState를 이용한 메시지 누적 예제입니다."""
# AIMessage와 HumanMessage는 대화 메시지 객체를 생성할 때 사용합니다.
from langchain_core.messages import AIMessage, HumanMessage
# MessagesState는 messages 필드와 메시지 누적 Reducer가 포함된 기본 상태입니다.
from langgraph.graph import END, START, MessagesState, StateGraph
def echo_node(state: MessagesState) -> dict:
"""가장 최근 사용자 메시지를 읽고 AI 응답을 생성합니다."""
# 현재 메시지 목록의 마지막 메시지를 가져옵니다.
last_message = state["messages"][-1]
# 마지막 메시지 내용을 포함한 응답 문자열을 만듭니다.
answer_text = f"LangGraph 노드가 입력을 확인했습니다: {last_message.content}"
# AIMessage 객체를 반환하면 기존 messages 목록에 자동으로 누적됩니다.
return {"messages": [AIMessage(content=answer_text)]}
def build_messages_graph():
"""사용자 메시지에 응답하는 메시지 그래프를 생성합니다."""
# MessagesState를 공유 상태로 사용하는 그래프 빌더를 생성합니다.
builder = StateGraph(MessagesState)
# echo 노드를 등록합니다.
builder.add_node("echo", echo_node)
# START에서 echo 노드로 연결합니다.
builder.add_edge(START, "echo")
# echo 실행 후 그래프를 종료합니다.
builder.add_edge("echo", END)
# 실행 가능한 그래프로 컴파일합니다.
return builder.compile()
def run_demo() -> None:
"""사용자 메시지와 AI 응답이 누적되는지 확인합니다."""
# 메시지 그래프를 생성합니다.
graph = build_messages_graph()
# 사용자 메시지를 초기 상태로 전달해 그래프를 실행합니다.
result = graph.invoke(
{"messages": [HumanMessage(content="MessagesState의 역할을 확인합니다.")]}
)
# 최종 messages 목록을 순서대로 출력합니다.
for index, message in enumerate(result["messages"], start=1):
# 메시지 객체의 클래스 이름을 확인합니다.
message_type = type(message).__name__
# 메시지 번호, 타입, 내용을 출력합니다.
print(f"{index}. {message_type}: {message.content}")
reducer_graph.py
import operator 기존 리스트와 새 리리스트를 더한다. add() sub() mul() truediv() eq() lt() gt()가 자주사용되며 리스트도 더할수가있다.
typing annotated typedict , langgraph end, start, stategrap
각 노드가 반환란 리스트를 뒤에 이어붙이고 첫, 두, 세번째 노드를 실행한다. stategraph 를 생성해 compile.invoke() for문으로 각 로그에 번호를 붙여 출력한다.
# -*- coding: utf-8 -*-
"""Reducer를 사용해 리스트 값을 누적하는 예제입니다."""
# operator.add는 기존 리스트와 새 리스트를 더하는 Reducer로 사용합니다.
import operator
# Annotated는 상태 필드에 Reducer 정보를 지정합니다.
# TypedDict는 상태 구조를 선언합니다.
from typing import Annotated, TypedDict
# LangGraph 그래프 구성 요소를 가져옵니다.
from langgraph.graph import END, START, StateGraph
class ReducerState(TypedDict):
"""실행 로그를 리스트에 누적하는 상태입니다."""
# 각 노드가 반환한 리스트를 기존 리스트 뒤에 이어 붙입니다.
logs: Annotated[list[str], operator.add]
def first_node(state: ReducerState) -> dict:
"""첫 번째 로그를 추가합니다."""
# 기존 목록을 직접 수정하지 않고 새 항목만 반환합니다.
return {"logs": ["첫 번째 노드 실행"]}
def second_node(state: ReducerState) -> dict:
"""두 번째 로그를 추가합니다."""
# Reducer가 이 리스트를 기존 logs 뒤에 자동으로 결합합니다.
return {"logs": ["두 번째 노드 실행"]}
def third_node(state: ReducerState) -> dict:
"""세 번째 로그를 추가합니다."""
# 세 번째 실행 기록을 반환합니다.
return {"logs": ["세 번째 노드 실행"]}
def build_reducer_graph():
"""세 노드의 로그가 누적되는 그래프를 생성합니다."""
# ReducerState를 사용하는 그래프 빌더를 생성합니다.
builder = StateGraph(ReducerState)
# 로그를 추가하는 노드를 등록합니다.
builder.add_node("first", first_node)
builder.add_node("second", second_node)
builder.add_node("third", third_node)
# 세 노드가 순차적으로 실행되도록 연결합니다.
builder.add_edge(START, "first")
builder.add_edge("first", "second")
builder.add_edge("second", "third")
builder.add_edge("third", END)
# 실행 가능한 그래프로 컴파일합니다.
return builder.compile()
def run_demo() -> None:
"""Reducer가 로그를 누적하는 결과를 출력합니다."""
# Reducer 그래프를 생성합니다.
graph = build_reducer_graph()
# 빈 로그 리스트로 그래프를 실행합니다.
result = graph.invoke({"logs": []})
# 누적된 로그의 제목을 출력합니다.
print("누적 로그:")
# 각 로그에 번호를 붙여 출력합니다.
for index, log in enumerate(result["logs"], start=1):
print(f"{index}. {log}")
utils.py
타이틀 출력.
enter 입력전까지 대기
# -*- coding: utf-8 -*-
"""콘솔 출력에 공통으로 사용하는 보조 함수입니다."""
def print_title(title: str) -> None:
"""실습 제목을 보기 좋게 출력합니다."""
# 각 실습 시작 위치를 구분하기 위해 빈 줄을 출력합니다.
print()
# 일정한 길이의 구분선을 출력합니다.
print("=" * 80)
# 전달받은 제목을 대괄호 형태로 출력합니다.
print(f"[{title}]")
# 제목 아래에도 같은 구분선을 출력합니다.
print("=" * 80)
def pause() -> None:
"""사용자가 결과를 읽을 수 있도록 Enter 입력을 기다립니다."""
# Enter 키 입력 전까지 메인 메뉴로 돌아가지 않도록 대기합니다.
input("\nEnter 키를 누르면 메인 메뉴로 돌아갑니다.")
main.py
import app가져오기
app.utils 가져오기
메뉴출력, 선형그래프를 생성하고 mermaid 문자열 생성, 조건부 분기그래프 생성, mermaid문자열생성 출력.
Mermaid문자열은 그래프를 그리기 위한 문자열. langgraph는 mermaid 문자열을 생성해 그래프 모양을 그려준다.
START
│
▼
A
│
▼
B
│
▼
END START
│
▼
A
┌──┴──┐
True False
│ │
▼ ▼
B C
\ /
▼ ▼
END# -*- coding: utf-8 -*-
"""LangGraph 핵심 개념을 메뉴별로 실행하는 콘솔 애플리케이션입니다."""
# 각 실습 모듈을 app 패키지에서 가져옵니다.
from app import (
basic_graph,
checkpoint_graph,
conditional_graph,
error_graph,
loop_graph,
messages_graph,
reducer_graph,
)
# 공통 콘솔 출력 함수를 가져옵니다.
from app.utils import pause, print_title
def print_menu() -> None:
"""사용자가 선택할 수 있는 실습 메뉴를 출력합니다."""
# 프로그램 전체 제목을 출력합니다.
print_title("LangGraph 핵심 개념 콘솔 학습 앱")
# 실행 가능한 메뉴를 번호와 함께 출력합니다.
print("1. State, Node, Edge 기본 선형 그래프")
print("2. add_conditional_edges 조건부 분기")
print("3. 반복 그래프와 종료 조건")
print("4. Reducer를 이용한 상태 누적")
print("5. MessagesState 메시지 누적")
print("6. InMemorySaver와 thread_id 체크포인트")
print("7. 노드 예외 처리와 fallback")
print("8. Mermaid 그래프 구조 출력")
print("0. 프로그램 종료")
def run_mermaid_demo() -> None:
"""기본 그래프와 조건부 그래프의 Mermaid 정의를 출력합니다."""
# 기본 선형 그래프를 생성합니다.
basic = basic_graph.build_basic_graph()
# 기본 그래프의 Mermaid 문자열을 생성합니다.
basic_mermaid = basic.get_graph().draw_mermaid()
# 기본 그래프 Mermaid 구조를 출력합니다.
print("[기본 선형 그래프 Mermaid]")
print(basic_mermaid)
# 조건부 분기 그래프를 생성합니다.
conditional = conditional_graph.build_conditional_graph()
# 조건부 그래프의 Mermaid 문자열을 생성합니다.
conditional_mermaid = conditional.get_graph().draw_mermaid()
# 조건부 그래프 Mermaid 구조를 출력합니다.
print("\n[조건부 분기 그래프 Mermaid]")
print(conditional_mermaid)
def main() -> None:
"""사용자가 종료할 때까지 메뉴를 반복 실행합니다."""
# 프로그램을 계속 실행하기 위한 무한 반복문입니다.
while True:
# 매 반복마다 메뉴를 출력합니다.
print_menu()
# 사용자가 입력한 메뉴 번호의 앞뒤 공백을 제거합니다.
choice = input("\n실행할 메뉴 번호를 입력하세요: ").strip()
# 0번을 선택하면 프로그램을 종료합니다.
if choice == "0":
print("\nLangGraph 콘솔 학습 앱을 종료합니다.")
break
# 개별 실습 오류가 전체 프로그램 종료로 이어지지 않도록 처리합니다.
try:
# 1번은 기본 선형 그래프를 실행합니다.
if choice == "1":
print_title("1. State, Node, Edge 기본 선형 그래프")
basic_graph.run_demo()
# 2번은 조건부 분기 그래프를 실행합니다.
elif choice == "2":
print_title("2. add_conditional_edges 조건부 분기")
conditional_graph.run_demo()
# 3번은 반복과 종료 조건 예제를 실행합니다.
elif choice == "3":
print_title("3. 반복 그래프와 종료 조건")
loop_graph.run_demo()
# 4번은 Reducer 누적 예제를 실행합니다.
elif choice == "4":
print_title("4. Reducer를 이용한 상태 누적")
reducer_graph.run_demo()
# 5번은 MessagesState 예제를 실행합니다.
elif choice == "5":
print_title("5. MessagesState 메시지 누적")
messages_graph.run_demo()
# 6번은 Checkpointer 예제를 실행합니다.
elif choice == "6":
print_title("6. InMemorySaver와 thread_id 체크포인트")
checkpoint_graph.run_demo()
# 7번은 오류 처리 예제를 실행합니다.
elif choice == "7":
print_title("7. 노드 예외 처리와 fallback")
error_graph.run_demo()
# 8번은 Mermaid 그래프 정의를 출력합니다.
elif choice == "8":
print_title("8. Mermaid 그래프 구조 출력")
run_mermaid_demo()
# 정의되지 않은 번호이면 입력 오류를 안내합니다.
else:
print("\n[입력 오류] 0부터 8 사이의 메뉴 번호를 입력하세요.")
except Exception as error:
# 실습 중 발생한 예외 타입과 내용을 출력합니다.
print(f"\n[실행 오류] {type(error).__name__}: {error}")
# 결과를 읽을 수 있도록 Enter 입력을 기다립니다.
pause()
# 이 파일을 직접 실행한 경우에만 main 함수를 호출합니다.
if __name__ == "__main__":
# 콘솔 메뉴 애플리케이션을 시작합니다.
main()
또다른 프로젝트 multi_agent_supervisor_console_project를 분석해보자.
일반에이전트는 노드가 하나라면 멀티에이전트는 역할을 나누어 각 에이전트는 자기일만 한다. supervisor는 관리자로 router 역할을 한다. 노드가 여러개 supervisor도 하나의 LLM이며 LLM이 분류결과를 반환하면 langraph는 노드로 이동한다.일반 함수에서 단순한 로직이 들어간다면 에이전트는 내부에서 LLM이 추론하고 tool도 호출할 수 있다. 혹은 다른 에이전트를 다시 호출할수도있으니 단순 코드가 아니라 ai한명으로 인지하라.
common.py
import os pathlib dotenv
환경변수설정 client만들기 provider만들어 각각의 chat객체만들기
이 두가지 경우로 embedding 객체만들기
지기접 호출시 현재 구조, 상태간단 문자여로 출력
# -*- coding: utf-8 -*-
"""
common.py — 모든 실습 공통 파일
목적:
- .env 의 API 키를 한 곳에서 로드한다.
- Gemini(주력) / OpenAI(보조) 모델 객체를 일관되게 생성한다.
- 실습 데이터(data/) 경로를 쉽게 찾는다.
각 강의 실습 코드 맨 위에서 다음처럼 불러 씁니다.
from common import get_chat, get_genai_client, DATA
"""
import os
import pathlib
from dotenv import load_dotenv
# 프로젝트 루트
ROOT = pathlib.Path(__file__).resolve().parent.parent
DATA = ROOT / "data"
DOCS = DATA / "docs"
# .env 로드 (루트의 .env 를 읽음)
load_dotenv(ROOT / ".env")
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
GEMINI_EMBED_MODEL = os.getenv("GEMINI_EMBED_MODEL", "models/gemini-embedding-001")
def require_key(name: str) -> str:
"""환경변수 키가 없으면 종료."""
val = os.getenv(name)
if not val or val.startswith("여기에"):
raise SystemExit(
f"[설정 필요] {name} 가 .env 에 없습니다.\n"
f" 1) cp .env.example .env\n"
f" 2) .env 파일을 열어 {name} 값을 채우세요."
)
return val
# ---------- raw SDK (원리 학습용) ----------
def get_genai_client():
"""google-genai 클라이언트 (from google import genai)."""
from google import genai
return genai.Client(api_key=require_key("GOOGLE_API_KEY"))
# ---------- LangChain Chat 모델 (현업용) ----------
def get_chat(provider: str = "gemini", temperature: float = 0.0):
"""LangChain ChatModel 반환. provider: 'gemini'(기본) | 'openai'."""
if provider == "gemini":
require_key("GOOGLE_API_KEY")
from langchain_google_genai import ChatGoogleGenerativeAI
return ChatGoogleGenerativeAI(model=GEMINI_MODEL, temperature=temperature)
elif provider == "openai":
require_key("OPENAI_API_KEY")
from langchain_openai import ChatOpenAI
return ChatOpenAI(model="gpt-4o-mini", temperature=temperature)
raise ValueError(f"알 수 없는 provider: {provider}")
def get_embeddings(provider: str = "gemini"):
"""LangChain Embeddings 반환."""
if provider == "gemini":
require_key("GOOGLE_API_KEY")
from langchain_google_genai import GoogleGenerativeAIEmbeddings
# gemini-embedding-001 기본 출력은 3072차원. 실습 저장·속도를 위해 768로 고정.
# (768/1536/3072 중 선택 가능)
return GoogleGenerativeAIEmbeddings(
model=GEMINI_EMBED_MODEL, output_dimensionality=768)
elif provider == "openai":
require_key("OPENAI_API_KEY")
from langchain_openai import OpenAIEmbeddings
return OpenAIEmbeddings(model="text-embedding-3-small")
raise ValueError(f"알 수 없는 provider: {provider}")
if __name__ == "__main__":
print("ROOT :", ROOT)
print("DATA :", DATA, "(존재:", DATA.exists(), ")")
print("GEMINI_MODEL :", GEMINI_MODEL)
print("키 로드 상태 — GOOGLE_API_KEY:", bool(os.getenv("GOOGLE_API_KEY")),
"/ OPENAI_API_KEY:", bool(os.getenv("OPENAI_API_KEY")))
data_repository.py
csv, dataclass, path, data import
@dataclass(frozen=true) 데이터를 저장하는 클래스, 생성된 객체를 수정할 수없도록 읽기모드로 설정한다.
product, faqitem, 필드 정의
file있는지 확인, csv 파일 존재확인. 인코딩 후 파일을 열어 read. procucts에 append.
daq도 마찬가지., 전체 상품 데이터를 읽어 stock이 이있는 상품만 선택하고 검색어가 포함되어야한다.
상품이 없으면 재고상품을 대상으로 추천 후보를 만든다. 평점 높은, 가격 낮은 순서로 정렬해 가성비 좋은 결과를 ranked.
안내.
faq 데이터를 읽어 검색어가 포함되는지 확인 if not matched tokens로 개별 단어를 사용해 ㅏㄴ번더 검색
마지막에 상담원 연결 추천.
# -*- coding: utf-8 -*-
"""상품 및 FAQ CSV 데이터를 읽고 검색하는 저장소 모듈입니다."""
from __future__ import annotations
import csv
from dataclasses import dataclass
from pathlib import Path
# 같은 code 폴더의 common.py가 제공하는 DATA 경로를 가져옵니다.
from common import DATA
@dataclass(frozen=True)
class Product:
"""상품 CSV 한 행을 표현하는 변경 불가능한 데이터 객체입니다."""
product_id: str
product_name: str
category: str
price: int
stock: int
rating: float
@dataclass(frozen=True)
class FaqItem:
"""FAQ CSV 한 행을 표현하는 변경 불가능한 데이터 객체입니다."""
question: str
answer: str
keywords: tuple[str, ...]
def _require_file(path: Path) -> None:
"""실습 데이터 파일이 없으면 원인을 알 수 있는 예외를 발생시킵니다."""
# 지정된 경로가 실제 파일인지 검사합니다.
if not path.is_file():
# 누락된 파일의 절대 경로를 포함하여 문제 해결에 필요한 정보를 제공합니다.
raise FileNotFoundError(f"실습 데이터 파일을 찾을 수 없습니다: {path}")
def load_products() -> list[Product]:
"""products.csv를 읽어 Product 객체 목록으로 반환합니다."""
# common.py의 DATA 경로를 기준으로 상품 CSV 파일 경로를 구성합니다.
path = DATA / "products.csv"
# 파일이 실제로 존재하는지 먼저 확인합니다.
_require_file(path)
# 읽은 상품을 저장할 빈 리스트를 생성합니다.
products: list[Product] = []
# Excel에서 저장한 CSV도 읽을 수 있도록 utf-8-sig 인코딩으로 파일을 엽니다.
with path.open("r", encoding="utf-8-sig", newline="") as file:
# 첫 행의 열 이름을 키로 사용하는 DictReader를 생성합니다.
reader = csv.DictReader(file)
# CSV의 각 행을 순서대로 읽습니다.
for row in reader:
# 문자열로 읽힌 숫자를 실제 int/float로 변환하여 Product 객체를 생성합니다.
products.append(
Product(
product_id=row["product_id"],
product_name=row["product_name"],
category=row["category"],
price=int(row["price"]),
stock=int(row["stock"]),
rating=float(row["rating"]),
)
)
# 완성된 상품 목록을 호출한 쪽에 반환합니다.
return products
def load_faq() -> list[FaqItem]:
"""faq.csv를 읽어 FaqItem 객체 목록으로 반환합니다."""
# common.py의 DATA 경로를 기준으로 FAQ CSV 파일 경로를 구성합니다.
path = DATA / "faq.csv"
# 파일 존재 여부를 검사합니다.
_require_file(path)
# 읽은 FAQ를 저장할 빈 리스트를 생성합니다.
faq_items: list[FaqItem] = []
# BOM이 있는 CSV도 안전하게 처리하도록 utf-8-sig로 파일을 엽니다.
with path.open("r", encoding="utf-8-sig", newline="") as file:
# 열 이름을 사용해 값을 읽는 DictReader를 생성합니다.
reader = csv.DictReader(file)
# 각 FAQ 행을 순서대로 처리합니다.
for row in reader:
# 쉼표로 저장된 키워드를 분리하고 공백을 제거한 튜플로 변환합니다.
keywords = tuple(word.strip() for word in row["keywords"].split(",") if word.strip())
# FAQ 데이터 객체를 만들어 목록에 추가합니다.
faq_items.append(
FaqItem(
question=row["question"],
answer=row["answer"],
keywords=keywords,
)
)
# 완성된 FAQ 목록을 반환합니다.
return faq_items
def search_products(category: str, limit: int = 3) -> str:
"""카테고리와 관련된 재고 보유 상품을 평점순으로 최대 limit개 추천합니다."""
# 사용자가 입력한 검색어를 비교하기 쉬운 소문자 문자열로 정규화합니다.
normalized = category.strip().lower()
# 전체 상품 데이터를 읽습니다.
products = load_products()
# 카테고리나 상품명에 검색어가 포함되고 재고가 있는 상품만 선택합니다.
matched = [
product
for product in products
if product.stock > 0
and (
normalized in product.category.lower()
or normalized in product.product_name.lower()
or product.category.lower() in normalized
)
]
# 직접 일치하는 상품이 없으면 전체 재고 상품을 대상으로 추천 후보를 만듭니다.
if not matched:
matched = [product for product in products if product.stock > 0]
# 평점은 높은 순서, 가격은 낮은 순서로 정렬하여 가성비 좋은 결과를 앞에 둡니다.
ranked = sorted(matched, key=lambda product: (-product.rating, product.price))[:limit]
# 추천 가능한 상품이 하나도 없으면 안내 문장을 반환합니다.
if not ranked:
return "현재 추천 가능한 재고 상품이 없습니다."
# 사람이 읽기 쉬운 추천 결과 문자열을 만들기 위한 리스트를 생성합니다.
lines = [f"'{category}' 관련 추천 상품입니다."]
# 정렬된 상품 정보를 한 줄씩 추가합니다.
for product in ranked:
lines.append(
f"- {product.product_name} | {product.price:,}원 | "
f"평점 {product.rating:.1f} | 재고 {product.stock}개"
)
# 여러 줄을 줄바꿈 문자로 결합하여 최종 도구 결과로 반환합니다.
return "\n".join(lines)
def search_faq(keyword: str, limit: int = 2) -> str:
"""질문·답변·키워드에서 관련 FAQ를 찾아 최대 limit개 반환합니다."""
# 검색어 앞뒤 공백을 제거하고 소문자로 변환합니다.
normalized = keyword.strip().lower()
# 전체 FAQ 데이터를 읽습니다.
faq_items = load_faq()
# 각 FAQ의 질문, 답변, 키워드에 검색어가 포함되는지 검사합니다.
matched = [
item
for item in faq_items
if normalized in item.question.lower()
or normalized in item.answer.lower()
or any(normalized in tag.lower() or tag.lower() in normalized for tag in item.keywords)
]
# 완전한 검색어로 찾지 못하면 질문 속 개별 단어를 사용해 한 번 더 검색합니다.
if not matched:
tokens = [token for token in normalized.replace("?", " ").split() if len(token) >= 2]
matched = [
item
for item in faq_items
if any(
token in item.question.lower()
or token in item.answer.lower()
or any(token in tag.lower() for tag in item.keywords)
for token in tokens
)
]
# 관련 FAQ가 없으면 근거를 찾지 못했다는 명확한 메시지를 반환합니다.
if not matched:
return "관련 FAQ 근거를 찾지 못했습니다. 상담원 연결이 필요합니다."
# FAQ 결과를 사람이 읽기 쉬운 Q/A 형식으로 변환합니다.
lines: list[str] = []
# 최대 limit개 항목만 출력합니다.
for item in matched[:limit]:
lines.append(f"Q. {item.question}\nA. {item.answer}")
# 각 FAQ 항목 사이를 빈 줄로 구분하여 반환합니다.
return "\n\n".join(lines)
message_utils.py
import typing, annotation
extract_text 문자열이면 그대로 반환, getattr해 content 속성가져오기 for문을 돌려 text 키가 있으면 수집.
# -*- coding: utf-8 -*-
"""LLM 응답 객체를 안전하게 문자열로 변환하는 공통 유틸리티 모듈입니다."""
from __future__ import annotations
from typing import Any
def extract_text(response: Any) -> str:
"""OpenAI/Gemini LangChain 응답에서 최종 문자열을 안전하게 추출합니다.
LangChain 채팅 모델의 일반적인 반환값은 AIMessage이며 ``content`` 속성을 가집니다.
다만 라이브러리 버전이나 테스트용 모의 객체에 따라 문자열, 리스트 또는 ``text`` 속성이
전달될 수 있으므로 여러 형태를 순서대로 검사하여 호환성을 높입니다.
"""
# 응답이 이미 문자열이면 별도 변환 없이 앞뒤 공백만 제거하여 반환합니다.
if isinstance(response, str):
return response.strip()
# 일반적인 LangChain AIMessage 객체의 content 속성을 가져옵니다.
content = getattr(response, "content", None)
# content가 문자열이면 그대로 정리하여 반환합니다.
if isinstance(content, str):
return content.strip()
# 일부 모델은 content를 텍스트 블록 리스트로 반환할 수 있으므로 리스트를 처리합니다.
if isinstance(content, list):
# 여러 블록에서 텍스트만 수집하기 위한 빈 리스트를 생성합니다.
text_parts: list[str] = []
# content 리스트 안의 블록을 하나씩 순회합니다.
for block in content:
# 블록이 딕셔너리이고 text 키가 있으면 해당 텍스트를 수집합니다.
if isinstance(block, dict) and isinstance(block.get("text"), str):
text_parts.append(block["text"])
# 블록이 단순 문자열이면 해당 문자열을 그대로 수집합니다.
elif isinstance(block, str):
text_parts.append(block)
# 수집한 텍스트 블록을 줄바꿈으로 합친 후 공백을 제거하여 반환합니다.
if text_parts:
return "\n".join(text_parts).strip()
# 구형 또는 사용자 정의 객체가 text 속성을 제공하는 경우를 처리합니다.
text = getattr(response, "text", None)
# text 속성이 문자열이면 앞뒤 공백을 제거하여 반환합니다.
if isinstance(text, str):
return text.strip()
# 마지막 수단으로 객체 전체를 문자열로 변환하여 빈 결과를 방지합니다.
return str(response).strip()
agents.py
import annotations, dataclasses, tkyping, common, data_repository message_utils
agentresult, sepeciallistagent, 설정, prompt설정해 llminvoke(), extract_text()를 통해 실제 txt만 추출해 return.
prompt지정해 llm.invoke() 최종답변을 요구.
빈질문은 안내, _extract_tool_input 핵심검색어를 추출해 tool_function 도구 실행 _compost_answer 답변생성 agentresult() 로 묶어 반환speciallistagent 생성
# -*- coding: utf-8 -*-
"""추천 및 정책 역할을 분리한 전문 에이전트 모듈입니다."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any, Callable
# 제공된 common.py의 get_chat 함수를 사용하여 공급자별 LLM을 생성합니다.
from common import get_chat
# CSV 데이터를 검색하는 Python 도구 함수를 가져옵니다.
from data_repository import search_faq, search_products
# 여러 형태의 LLM 응답을 문자열로 바꾸는 공통 함수를 가져옵니다.
from message_utils import extract_text
@dataclass(frozen=True)
class AgentResult:
"""전문 에이전트 실행 결과와 사용 근거를 함께 보관하는 데이터 객체입니다."""
agent_name: str
tool_name: str
tool_input: str
evidence: str
answer: str
class SpecialistAgent:
"""역할별 프롬프트와 전용 도구 하나를 가진 간단하고 명시적인 전문 에이전트입니다."""
def __init__(
self,
*,
name: str,
role_prompt: str,
tool_name: str,
tool_function: Callable[[str], str],
provider: str,
) -> None:
# 에이전트 이름을 인스턴스 변수에 저장합니다.
self.name = name
# 해당 에이전트가 지켜야 할 역할 지시문을 저장합니다.
self.role_prompt = role_prompt
# 실행할 전용 도구 이름을 저장합니다.
self.tool_name = tool_name
# 실제 CSV 검색을 수행하는 Python 함수를 저장합니다.
self.tool_function = tool_function
# common.py의 get_chat을 사용해 선택된 공급자의 채팅 모델을 생성합니다.
self.llm = get_chat(provider=provider, temperature=0.0)
def _extract_tool_input(self, question: str) -> str:
"""LLM을 사용해 사용자 질문에서 도구에 전달할 짧은 검색어를 추출합니다."""
# 도구마다 적합한 입력을 한 줄로 추출하도록 제한적인 프롬프트를 구성합니다.
prompt = (
f"역할: {self.role_prompt}\n"
f"사용 도구: {self.tool_name}\n"
"다음 고객 질문에서 도구 검색에 사용할 가장 핵심적인 한국어 검색어만 출력하라. "
"설명, 따옴표, 문장부호 없이 한 단어나 짧은 구만 출력한다.\n"
f"고객 질문: {question}\n"
"검색어:"
)
# 선택된 LLM에 검색어 추출 프롬프트를 전달합니다.
response = self.llm.invoke(prompt)
# 모델 응답 객체에서 실제 텍스트를 안전하게 추출합니다.
keyword = extract_text(response).strip(" \"'`.,")
# 모델이 빈 문자열을 반환하면 원래 질문을 검색어로 사용해 실행을 계속합니다.
return keyword or question
def _compose_answer(self, question: str, evidence: str) -> str:
"""도구가 찾은 근거만 사용하여 최종 고객 답변을 생성합니다."""
# 모델이 근거 밖 내용을 만들지 않도록 역할과 근거 제한을 명시합니다.
prompt = (
f"당신은 {self.role_prompt}\n"
"아래 도구 실행 결과만 근거로 사용하여 고객에게 한국어로 답하라. "
"근거에 없는 내용을 추측하지 말고, 핵심 답변 뒤에 '[근거: 내부 CSV]'를 표시하라.\n\n"
f"고객 질문:\n{question}\n\n"
f"도구 실행 결과:\n{evidence}\n\n"
"최종 답변:"
)
# LLM에 최종 답변 생성을 요청합니다.
response = self.llm.invoke(prompt)
# 모델 반환 객체에서 텍스트를 추출하여 반환합니다.
return extract_text(response)
def run(self, question: str) -> AgentResult:
"""검색어 추출 → 전용 도구 실행 → 근거 기반 답변 생성 순서로 처리합니다."""
# 빈 질문은 외부 API를 호출하기 전에 즉시 거부하여 불필요한 비용을 방지합니다.
if not question.strip():
raise ValueError("질문이 비어 있습니다. 내용을 입력해 주세요.")
# LLM으로 전용 도구에 전달할 핵심 검색어를 추출합니다.
tool_input = self._extract_tool_input(question)
# 이 에이전트에 허용된 전용 Python 도구만 실행합니다.
evidence = self.tool_function(tool_input)
# 도구가 반환한 근거를 바탕으로 최종 자연어 답변을 생성합니다.
answer = self._compose_answer(question, evidence)
# 실행 추적에 필요한 모든 정보를 AgentResult로 묶어 반환합니다.
return AgentResult(
agent_name=self.name,
tool_name=self.tool_name,
tool_input=tool_input,
evidence=evidence,
answer=answer,
)
def build_specialists(provider: str) -> tuple[SpecialistAgent, SpecialistAgent]:
"""동일한 LLM 공급자를 사용하되 역할과 도구가 다른 두 전문 에이전트를 생성합니다."""
# 상품 추천 업무만 담당하는 sales 전문 에이전트를 생성합니다.
sales_agent = SpecialistAgent(
name="sales",
role_prompt=(
"승승장구몰의 상품 추천 전문 상담원이다. 상품의 카테고리, 가격, 평점, 재고를 "
"근거로 추천하고 정책 질문에는 답하지 않는다."
),
tool_name="search_products",
tool_function=search_products,
provider=provider,
)
# 환불·배송·교환 등 정책 업무만 담당하는 policy 전문 에이전트를 생성합니다.
policy_agent = SpecialistAgent(
name="policy",
role_prompt=(
"승승장구몰의 정책 및 FAQ 전문 상담원이다. 환불, 교환, 배송, 취소, 적립 정책을 "
"FAQ 근거로만 안내하고 상품 추천은 하지 않는다."
),
tool_name="search_faq",
tool_function=search_faq,
provider=provider,
)
# Supervisor가 선택해 사용할 수 있도록 두 에이전트를 튜플로 반환합니다.
return sales_agent, policy_agent
router.py
dataclasses, typing any, common, message_utils import
정책 키워드 정의, 상품 단어 정의,
@dataclass 라우팅 대상 판단 방식 설명 정의
질문의 공백을 제거해서 키워드가포함되었는지 검색, 키워드가있으면 즉시 라우팅 명확하지않으면 unkonwn 반환
prompt정의해 llm.invoke() 호출 소문자 문자열로 변환해 policy 문자열이 포함되면 정책 에이전트 선택. 규칙이 unkown이 아니면 decition 결과 그대로 반환하고 애매한 질문에서는 llm 라우터를 호출. 한번 더 호출
get chat
# -*- coding: utf-8 -*-
"""규칙, LLM, 하이브리드 Supervisor 라우터를 구현한 모듈입니다."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Any
# 제공된 common.py에서 채팅 모델 생성 함수를 가져옵니다.
from common import get_chat
# LLM 응답의 content를 안전하게 문자열로 변환하는 함수를 가져옵니다.
from message_utils import extract_text
# 명시적인 정책 질문을 빠르게 찾기 위한 키워드 튜플입니다.
POLICY_WORDS: tuple[str, ...] = (
"환불",
"교환",
"배송",
"취소",
"적립",
"포인트",
"무료배송",
"회원",
"등급",
"반품",
)
# 명시적인 상품 추천 질문을 찾기 위한 키워드 튜플입니다.
SALES_WORDS: tuple[str, ...] = (
"추천",
"상품",
"제품",
"가성비",
"인기",
"선물",
"전자기기",
"패션",
"식품",
)
@dataclass(frozen=True)
class RouteDecision:
"""라우팅 대상과 판단 방식 및 설명을 함께 전달하는 데이터 객체입니다."""
target: str
method: str
reason: str
llm_calls: int
def route_rule(question: str) -> RouteDecision:
"""키워드만 사용하여 빠르고 비용 없이 policy 또는 sales로 분류합니다."""
# 입력 질문의 앞뒤 공백을 제거합니다.
normalized = question.strip()
# 정책 키워드가 하나라도 포함되어 있는지 검사합니다.
policy_hits = [word for word in POLICY_WORDS if word in normalized]
# 정책 키워드가 있으면 정책 에이전트로 즉시 라우팅합니다.
if policy_hits:
return RouteDecision(
target="policy",
method="rule",
reason=f"정책 키워드 감지: {', '.join(policy_hits)}",
llm_calls=0,
)
# 추천 키워드가 하나라도 포함되어 있는지 검사합니다.
sales_hits = [word for word in SALES_WORDS if word in normalized]
# 추천 키워드가 있으면 추천 에이전트로 즉시 라우팅합니다.
if sales_hits:
return RouteDecision(
target="sales",
method="rule",
reason=f"추천 키워드 감지: {', '.join(sales_hits)}",
llm_calls=0,
)
# 어느 쪽도 명확하지 않으면 애매함을 나타내는 unknown을 반환합니다.
return RouteDecision(
target="unknown",
method="rule",
reason="명시적인 정책 또는 추천 키워드를 찾지 못함",
llm_calls=0,
)
def route_llm(llm: Any, question: str) -> RouteDecision:
"""LLM의 의미 이해를 사용하여 질문을 policy 또는 sales로 분류합니다."""
# 출력 형식을 두 라벨로 제한하는 분류 프롬프트를 구성합니다.
prompt = (
"고객 질문을 다음 두 라벨 중 하나로 분류하라.\n"
"- policy: 환불, 취소, 반품, 교환, 배송 기간, 적립, 회원 정책 문의\n"
"- sales: 상품 선택, 추천, 가격대, 카테고리, 인기 상품 문의\n"
"반드시 policy 또는 sales 한 단어만 출력한다.\n"
f"질문: {question}\n"
"라벨:"
)
# 분류 전용 LLM을 한 번 호출합니다.
response = llm.invoke(prompt)
# 응답을 소문자 문자열로 변환합니다.
answer = extract_text(response).lower()
# policy 문자열이 포함되면 정책 에이전트를 선택합니다.
if "policy" in answer:
target = "policy"
# 그 외에는 안전한 두 번째 라벨인 sales로 정규화합니다.
else:
target = "sales"
# 호출 횟수 1회와 원본 분류 응답을 설명에 포함해 반환합니다.
return RouteDecision(
target=target,
method="llm",
reason=f"LLM 분류 응답: {answer!r}",
llm_calls=1,
)
def route_hybrid(llm: Any, question: str) -> RouteDecision:
"""명확한 질문은 규칙으로, 애매한 질문만 LLM으로 처리합니다."""
# 먼저 비용이 없는 규칙 라우터를 실행합니다.
rule_decision = route_rule(question)
# 규칙이 policy 또는 sales를 명확히 결정했으면 그 결과를 그대로 반환합니다.
if rule_decision.target != "unknown":
return RouteDecision(
target=rule_decision.target,
method="hybrid-rule",
reason=rule_decision.reason,
llm_calls=0,
)
# 규칙으로 판단할 수 없는 애매한 질문에 대해서만 LLM 라우터를 호출합니다.
llm_decision = route_llm(llm, question)
# 하이브리드 경로를 명확히 표시한 결과를 반환합니다.
return RouteDecision(
target=llm_decision.target,
method="hybrid-llm",
reason=f"규칙 판단 불가 → {llm_decision.reason}",
llm_calls=llm_decision.llm_calls,
)
def build_router_llm(provider: str) -> Any:
"""Supervisor 분류에 사용할 temperature 0 채팅 모델을 생성합니다."""
# common.py의 get_chat을 통해 Gemini 또는 OpenAI 모델을 동일한 인터페이스로 생성합니다.
return get_chat(provider=provider, temperature=0.0)
supervisor.py
import dataclasses, typing, agents, roupter
추천, 정책 전문 에이전트 생성.
딕셔너리 구성 규칙모드에서 키워드 라우터 실행
unknown일때는 saled로 보내기
llm 모드에서는 모든 질문을 llm으로 분류한다 .하이브리드모드에서는 애매할때만 llm 사용
질문을 decide(), agent() 객체 가져오기 run() 겨로가반환
# -*- coding: utf-8 -*-
"""Supervisor가 질문을 전문 에이전트에 위임하는 중앙 제어 모듈입니다."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Literal
# 역할별 전문 에이전트와 결과 객체를 가져옵니다.
from agents import AgentResult, SpecialistAgent, build_specialists
# 세 가지 라우터와 결과 객체를 가져옵니다.
from router import RouteDecision, build_router_llm, route_hybrid, route_llm, route_rule
# 메뉴에서 허용할 라우터 방식의 타입을 명시합니다.
RouterMode = Literal["rule", "llm", "hybrid"]
@dataclass(frozen=True)
class SupervisorResult:
"""Supervisor 판단과 전문 에이전트 결과를 하나로 묶는 실행 결과입니다."""
question: str
route: RouteDecision
agent_result: AgentResult
class Supervisor:
"""중앙에서 질문을 분류하고 한 명의 전문 에이전트에게 위임합니다."""
def __init__(self, provider: str) -> None:
# 라우팅에 사용할 LLM을 생성합니다.
self.router_llm = build_router_llm(provider)
# 같은 공급자를 사용하는 추천·정책 전문 에이전트를 생성합니다.
sales_agent, policy_agent = build_specialists(provider)
# 라우팅 결과 문자열로 전문 에이전트를 바로 찾을 수 있도록 딕셔너리를 구성합니다.
self.agents: dict[str, SpecialistAgent] = {
"sales": sales_agent,
"policy": policy_agent,
}
def decide(self, question: str, mode: RouterMode) -> RouteDecision:
"""선택한 모드에 따라 라우팅 결정을 생성합니다."""
# 규칙 모드에서는 키워드 라우터를 실행합니다.
if mode == "rule":
decision = route_rule(question)
# 순수 규칙 라우터가 애매함을 반환하면 HTML 예제의 기본 동작처럼 sales로 보냅니다.
if decision.target == "unknown":
return RouteDecision(
target="sales",
method="rule-default",
reason=f"{decision.reason}; 기본 대상 sales 적용",
llm_calls=0,
)
# 명확한 규칙 결과는 그대로 반환합니다.
return decision
# LLM 모드에서는 모든 질문을 LLM으로 분류합니다.
if mode == "llm":
return route_llm(self.router_llm, question)
# 하이브리드 모드에서는 규칙 우선, 애매할 때만 LLM을 사용합니다.
if mode == "hybrid":
return route_hybrid(self.router_llm, question)
# 허용하지 않은 문자열이 들어오면 즉시 오류를 발생시킵니다.
raise ValueError(f"지원하지 않는 라우터 모드입니다: {mode}")
def run(self, question: str, mode: RouterMode = "hybrid") -> SupervisorResult:
"""라우팅 결정을 내리고 선택된 전문 에이전트에 질문을 위임합니다."""
# 빈 질문은 LLM 호출 전에 차단합니다.
if not question.strip():
raise ValueError("질문이 비어 있습니다. 내용을 입력해 주세요.")
# 선택한 라우팅 방식으로 담당 전문 에이전트를 결정합니다.
decision = self.decide(question, mode)
# 결정된 대상 이름으로 전문 에이전트 객체를 가져옵니다.
agent = self.agents[decision.target]
# 원래 사용자 질문을 변경하지 않고 전문 에이전트에 전달합니다.
agent_result = agent.run(question)
# 중앙 라우팅 정보와 전문 에이전트 실행 결과를 함께 반환합니다.
return SupervisorResult(
question=question,
route=decision,
agent_result=agent_result,
)
torch_evaluation.py
testset 문자열 정의 ,
평가데이터 정의,
for문으로 돌려 rule를 얻어 애매한 질문은 sales 를 기본값으로 사용해 append
정답라벨을 정수인덱스 텐서로 변환, 예측도. 불리어 텐서 생성해 int 합산, accuracy 구해 return.
예측갯수가 테스트셋 크기와 ㅇ같은지 검사하고 텐서변환후 일치여부 계산, 정수변환, 저오학도 계싼해 반환
# -*- coding: utf-8 -*-
"""PyTorch 텐서로 라우터 정확도와 비용 지표를 계산하는 평가 모듈입니다."""
from __future__ import annotations
from dataclasses import dataclass
from typing import Callable
import torch
# 규칙 라우터 함수를 가져옵니다.
from router import route_rule
# 정답 라벨이 포함된 라우팅 테스트 데이터입니다.
TESTSET: tuple[tuple[str, str], ...] = (
("환불은 며칠 안에 신청해야 해?", "policy"),
("무료배송 기준이 어떻게 돼?", "policy"),
("포인트 적립은 언제 돼?", "policy"),
("교환하려면 어떻게 해?", "policy"),
("전자기기 추천 좀 해줘", "sales"),
("패션 인기상품 뭐 있어?", "sales"),
("선물용으로 좋은 거 골라줘", "sales"),
("가성비 좋은 거 없을까?", "sales"),
("주문한 거 무를 수 있어?", "policy"),
("이거 며칠이면 받아볼 수 있어?", "policy"),
)
# 문자열 라벨을 텐서 정수로 바꾸기 위한 매핑입니다.
LABEL_TO_INDEX: dict[str, int] = {"sales": 0, "policy": 1}
@dataclass(frozen=True)
class EvaluationResult:
"""라우터 평가 결과를 구조적으로 보관합니다."""
name: str
accuracy: float
correct: int
total: int
llm_calls: int
predictions: tuple[str, ...]
def evaluate_rule_router() -> EvaluationResult:
"""규칙 라우터를 테스트셋에 적용하고 PyTorch로 정확도를 계산합니다."""
# 모델 예측 라벨을 저장할 빈 리스트를 생성합니다.
predictions: list[str] = []
# 테스트 질문을 하나씩 규칙 라우터에 전달합니다.
for question, _gold in TESTSET:
# 규칙 라우터의 결정을 얻습니다.
decision = route_rule(question)
# HTML 예제와 동일하게 애매한 질문은 sales를 기본값으로 사용합니다.
predicted = decision.target if decision.target != "unknown" else "sales"
# 평가를 위해 예측 라벨을 목록에 추가합니다.
predictions.append(predicted)
# 정답 라벨을 정수 인덱스 텐서로 변환합니다.
gold_tensor = torch.tensor(
[LABEL_TO_INDEX[gold] for _question, gold in TESTSET],
dtype=torch.long,
)
# 예측 라벨을 정수 인덱스 텐서로 변환합니다.
prediction_tensor = torch.tensor(
[LABEL_TO_INDEX[prediction] for prediction in predictions],
dtype=torch.long,
)
# 각 위치에서 예측과 정답이 같은지 나타내는 불리언 텐서를 생성합니다.
correct_tensor = prediction_tensor.eq(gold_tensor)
# True의 개수를 합산하여 맞힌 질문 수를 계산합니다.
correct = int(correct_tensor.sum().item())
# 불리언 텐서를 float로 바꾼 뒤 평균을 구해 정확도를 계산합니다.
accuracy = float(correct_tensor.float().mean().item())
# 구조화된 평가 결과를 반환합니다.
return EvaluationResult(
name="규칙 라우터",
accuracy=accuracy,
correct=correct,
total=len(TESTSET),
llm_calls=0,
predictions=tuple(predictions),
)
def evaluate_predictions(
*,
name: str,
predictions: list[str],
llm_calls: int,
) -> EvaluationResult:
"""외부에서 얻은 LLM/하이브리드 예측 목록을 PyTorch 텐서로 평가합니다."""
# 예측 개수가 테스트셋 크기와 같은지 검사합니다.
if len(predictions) != len(TESTSET):
raise ValueError("예측 개수와 테스트셋 크기가 일치하지 않습니다.")
# 정답 문자열을 정수 텐서로 변환합니다.
gold_tensor = torch.tensor(
[LABEL_TO_INDEX[gold] for _question, gold in TESTSET],
dtype=torch.long,
)
# 예측 문자열을 정수 텐서로 변환합니다.
prediction_tensor = torch.tensor(
[LABEL_TO_INDEX[prediction] for prediction in predictions],
dtype=torch.long,
)
# 정답과 예측의 위치별 일치 여부를 계산합니다.
correct_tensor = prediction_tensor.eq(gold_tensor)
# 일치한 항목 개수를 Python 정수로 변환합니다.
correct = int(correct_tensor.sum().item())
# 전체 평균으로 정확도를 계산합니다.
accuracy = float(correct_tensor.float().mean().item())
# 계산한 결과를 EvaluationResult 객체로 반환합니다.
return EvaluationResult(
name=name,
accuracy=accuracy,
correct=correct,
total=len(TESTSET),
llm_calls=llm_calls,
predictions=tuple(predictions),
)
main.py
import os, sys, path,
path 지정
현재 폴더를 code_dir로 지정
경로보정
common, datarespository, supervicor, 가져오기
torch_evaluation의 testset, evaluate_predictions, evaluate_rule_router 가져오기
gemini 공급자 설정
title print, meny 보여주기, 환경 보여주기프로바이더 선택하기 검색어를 입력받아 search_faq() LLM 호출없이 csv 검색 도구 자체로 확인하기
api 키 및 모델 초기화가 필요한 시점에 supervisor 생성. 질문을 받아 supervisor.agent() 선택한 에이전트를 가져와 agent.run()
현재 공급자에 맞는 supercisor와 질문, run() 단계별로 출력한다.
규칙 평가함수를 호출해 출력.
비교에도 LLM호출이 필요함으로 supervisor를 생성해 결과를 계산해 return testset 질문을 for문으로 돌려 supervisor가 decide() 하도록한뒤 append. 명확한 질문은 규칙, 애매한 질문은 LLM으로 분류한다.
LLM 평가하이브리드도 평가. for문으로 출력.
메뉴출력
# -*- coding: utf-8 -*-
"""PyCharm에서 바로 실행하는 멀티에이전트 Supervisor 콘솔 애플리케이션입니다."""
from __future__ import annotations
import os
import sys
from pathlib import Path
# 현재 main.py가 있는 code 폴더를 Python 모듈 검색 경로에 추가합니다.
CODE_DIR = Path(__file__).resolve().parent
# 직접 실행과 PyCharm 실행 구성 모두에서 같은 모듈을 찾도록 경로를 보정합니다.
if str(CODE_DIR) not in sys.path:
sys.path.insert(0, str(CODE_DIR))
# 제공된 common.py의 공통 경로 및 모델 설정을 가져옵니다.
from common import DATA, GEMINI_MODEL, ROOT
# CSV 도구를 직접 실행하기 위한 함수를 가져옵니다.
from data_repository import search_faq, search_products
# 중앙 Supervisor 클래스를 가져옵니다.
from supervisor import Supervisor
# PyTorch 기반 라우팅 평가 함수와 테스트셋을 가져옵니다.
from torch_evaluation import TESTSET, evaluate_predictions, evaluate_rule_router
# 현재 콘솔 세션에서 사용할 기본 LLM 공급자입니다.
CURRENT_PROVIDER = "gemini"
def print_title(title: str) -> None:
"""메뉴 실행 결과를 구분하기 위한 제목 선을 출력합니다."""
# 가독성을 위해 빈 줄과 구분선을 출력합니다.
print("\n" + "=" * 78)
# 전달받은 제목을 출력합니다.
print(title)
# 제목 아래 구분선을 출력합니다.
print("=" * 78)
def show_menu() -> None:
"""HTML 설명 조회 항목 없이 실행 실습 메뉴만 출력합니다."""
# 현재 선택된 LLM 공급자를 메뉴 상단에 표시합니다.
print(f"\n[현재 LLM 공급자: {CURRENT_PROVIDER}]")
# 앱에서 직접 실행할 수 있는 실습 기능만 출력합니다.
print("1. 공통 환경 및 API 키 로드 상태 확인")
print("2. LLM 공급자 선택 (Gemini / OpenAI)")
print("3. 추천 도구 단독 실행 (API 호출 없음)")
print("4. 정책 FAQ 도구 단독 실행 (API 호출 없음)")
print("5. 추천 전문 에이전트 실행")
print("6. 정책 전문 에이전트 실행")
print("7. Supervisor 규칙 라우터 실행")
print("8. Supervisor LLM 라우터 실행")
print("9. Supervisor 하이브리드 라우터 실행")
print("10. PyTorch 규칙 라우터 정확도 평가")
print("11. 규칙/LLM/하이브리드 라우터 비교")
print("0. 종료")
def show_environment() -> None:
"""common.py가 계산한 경로와 환경변수 로드 상태를 출력합니다."""
# 결과 영역 제목을 출력합니다.
print_title("공통 환경 확인")
# 프로젝트 루트 경로를 출력합니다.
print("ROOT:", ROOT)
# 데이터 폴더 경로와 존재 여부를 출력합니다.
print("DATA:", DATA, "(존재:", DATA.exists(), ")")
# 제공된 common.py가 선택한 Gemini 모델명을 출력합니다.
print("GEMINI_MODEL:", GEMINI_MODEL)
# 실제 키 값은 노출하지 않고 설정 여부만 출력합니다.
print("GOOGLE_API_KEY 설정:", bool(os.getenv("GOOGLE_API_KEY")))
# OpenAI 키도 값이 아닌 설정 여부만 출력합니다.
print("OPENAI_API_KEY 설정:", bool(os.getenv("OPENAI_API_KEY")))
def select_provider() -> None:
"""현재 콘솔 세션에서 사용할 LLM 공급자를 변경합니다."""
# 함수 안에서 전역 공급자 값을 수정하겠다고 선언합니다.
global CURRENT_PROVIDER
# 사용자가 선택할 수 있는 공급자 목록을 출력합니다.
print("\n1. Gemini")
print("2. OpenAI")
# 사용자 선택 문자열을 입력받고 공백을 제거합니다.
choice = input("선택: ").strip()
# 1번을 선택하면 common.py의 기본 공급자인 gemini를 설정합니다.
if choice == "1":
CURRENT_PROVIDER = "gemini"
# 2번을 선택하면 openai 공급자를 설정합니다.
elif choice == "2":
CURRENT_PROVIDER = "openai"
# 그 외 입력은 설정을 변경하지 않고 안내합니다.
else:
print("잘못된 선택입니다. 기존 설정을 유지합니다.")
return
# 변경된 공급자를 사용자에게 확인시킵니다.
print(f"LLM 공급자를 '{CURRENT_PROVIDER}'로 변경했습니다.")
def run_tool(tool_type: str) -> None:
"""LLM 호출 없이 CSV 검색 도구 자체의 동작을 확인합니다."""
# 추천 도구를 실행하는 경우 카테고리를 입력받습니다.
if tool_type == "sales":
keyword = input("상품 카테고리 또는 검색어: ").strip()
print_title("search_products 실행 결과")
print(search_products(keyword))
return
# 정책 도구를 실행하는 경우 정책 검색어를 입력받습니다.
keyword = input("정책 검색어: ").strip()
print_title("search_faq 실행 결과")
print(search_faq(keyword))
def run_specialist(target: str) -> None:
"""Supervisor 라우팅 없이 선택한 전문 에이전트를 직접 실행합니다."""
# API 키 및 모델 초기화가 필요한 시점에 Supervisor 객체를 생성합니다.
supervisor = Supervisor(CURRENT_PROVIDER)
# 사용자 질문을 입력받습니다.
question = input("고객 질문: ").strip()
# Supervisor 내부 딕셔너리에서 선택한 전문 에이전트를 가져옵니다.
agent = supervisor.agents[target]
# 전문 에이전트의 검색어 추출, 도구 실행, 답변 생성을 수행합니다.
result = agent.run(question)
# 실행 추적 정보를 보기 쉽게 출력합니다.
print_title(f"{target} 전문 에이전트 실행 결과")
print("에이전트:", result.agent_name)
print("사용 도구:", result.tool_name)
print("도구 입력:", result.tool_input)
print("\n[도구 근거]\n" + result.evidence)
print("\n[최종 답변]\n" + result.answer)
def run_supervisor(mode: str) -> None:
"""선택한 라우팅 방식으로 Supervisor 전체 흐름을 실행합니다."""
# 현재 공급자에 맞는 Supervisor와 두 전문 에이전트를 생성합니다.
supervisor = Supervisor(CURRENT_PROVIDER)
# 라우팅할 고객 질문을 입력받습니다.
question = input("고객 질문: ").strip()
# Supervisor가 질문을 분류하고 담당 에이전트에 위임하도록 실행합니다.
result = supervisor.run(question, mode=mode) # type: ignore[arg-type]
# 중앙 라우팅과 전문 에이전트 실행 내용을 단계별로 출력합니다.
print_title(f"Supervisor {mode} 라우팅 결과")
print("라우팅 대상:", result.route.target)
print("판단 방식:", result.route.method)
print("판단 근거:", result.route.reason)
print("라우팅 LLM 호출:", result.route.llm_calls, "회")
print("전문 에이전트:", result.agent_result.agent_name)
print("사용 도구:", result.agent_result.tool_name)
print("도구 입력:", result.agent_result.tool_input)
print("\n[도구 근거]\n" + result.agent_result.evidence)
print("\n[최종 답변]\n" + result.agent_result.answer)
def evaluate_rule() -> None:
"""외부 API 없이 규칙 라우터의 정확도를 PyTorch로 계산합니다."""
# 규칙 라우터 평가 함수를 호출합니다.
result = evaluate_rule_router()
# 평가 개요를 출력합니다.
print_title("PyTorch 규칙 라우터 평가")
print(f"정확도: {result.correct}/{result.total} ({result.accuracy * 100:.1f}%)")
print("LLM 호출:", result.llm_calls, "회")
# 각 질문의 정답과 예측을 나란히 출력합니다.
for (question, gold), predicted in zip(TESTSET, result.predictions):
mark = "O" if gold == predicted else "X"
print(f"[{mark}] {question} | 정답={gold} | 예측={predicted}")
def compare_routers() -> None:
"""동일한 테스트셋으로 규칙, LLM, 하이브리드 라우터를 비교합니다."""
# 비교에는 실제 LLM 호출이 필요하므로 현재 공급자의 Supervisor를 생성합니다.
supervisor = Supervisor(CURRENT_PROVIDER)
# API 호출이 없는 규칙 라우터 결과를 먼저 계산합니다.
rule_result = evaluate_rule_router()
# LLM 라우터 예측과 호출 횟수를 저장할 변수를 준비합니다.
llm_predictions: list[str] = []
llm_calls = 0
# 하이브리드 라우터 예측과 호출 횟수를 저장할 변수를 준비합니다.
hybrid_predictions: list[str] = []
hybrid_calls = 0
# 같은 테스트 질문을 두 라우터에 각각 전달합니다.
for question, _gold in TESTSET:
# 모든 질문을 LLM으로 분류합니다.
llm_decision = supervisor.decide(question, "llm")
llm_predictions.append(llm_decision.target)
llm_calls += llm_decision.llm_calls
# 명확한 질문은 규칙, 애매한 질문은 LLM으로 분류합니다.
hybrid_decision = supervisor.decide(question, "hybrid")
hybrid_predictions.append(hybrid_decision.target)
hybrid_calls += hybrid_decision.llm_calls
# LLM 라우터 예측을 PyTorch 텐서로 평가합니다.
llm_result = evaluate_predictions(
name="LLM 라우터",
predictions=llm_predictions,
llm_calls=llm_calls,
)
# 하이브리드 라우터 예측도 같은 방식으로 평가합니다.
hybrid_result = evaluate_predictions(
name="하이브리드 라우터",
predictions=hybrid_predictions,
llm_calls=hybrid_calls,
)
# 세 라우터의 정확도와 호출 횟수를 한눈에 비교하여 출력합니다.
print_title("규칙 / LLM / 하이브리드 라우터 비교")
for result in (rule_result, llm_result, hybrid_result):
print(
f"{result.name:<14} | 정확도 {result.correct}/{result.total} "
f"({result.accuracy * 100:5.1f}%) | LLM 호출 {result.llm_calls}회"
)
def main() -> None:
"""사용자가 종료할 때까지 콘솔 메뉴를 반복 실행합니다."""
# 프로그램 시작 메시지를 출력합니다.
print("멀티에이전트 Supervisor 콘솔 실습 앱")
print("HTML 설명 조회 메뉴 없이 실행 실습만 제공합니다.")
# 사용자가 0을 입력할 때까지 메뉴를 반복합니다.
while True:
# 현재 메뉴를 출력합니다.
show_menu()
# 메뉴 번호를 문자열로 입력받습니다.
choice = input("메뉴 선택: ").strip()
# 각 메뉴 번호에 맞는 함수를 실행합니다.
try:
if choice == "1":
show_environment()
elif choice == "2":
select_provider()
elif choice == "3":
run_tool("sales")
elif choice == "4":
run_tool("policy")
elif choice == "5":
run_specialist("sales")
elif choice == "6":
run_specialist("policy")
elif choice == "7":
run_supervisor("rule")
elif choice == "8":
run_supervisor("llm")
elif choice == "9":
run_supervisor("hybrid")
elif choice == "10":
evaluate_rule()
elif choice == "11":
compare_routers()
elif choice == "0":
print("프로그램을 종료합니다.")
break
else:
print("0~11 사이의 메뉴 번호를 입력해 주세요.")
# 설정 누락, API 오류, 입력 오류가 발생해도 콘솔 프로그램 전체가 종료되지 않게 처리합니다.
except Exception as error:
print(f"\n[실행 오류] {type(error).__name__}: {error}")
# 이 파일을 직접 실행했을 때만 main 함수를 호출합니다.
if __name__ == "__main__":
main()
