Role based agent
Role-based Agent라는것은 새로운 기술이 아닌 그냥 설계 패턴
새로운 객체나 라이브러리 이름이 아니라, "역할별로 에이전트를 나누는 아키텍처"를 부르는 용어
어제 supervisor는 질문을 한명에게 위임했으나 현실업
무는 보통 여러단계로 쪼개야한다. planner설계자가 무엇을 할지를 정해 할일 목록을 만들고 worker가 각 항목을 어떻게 할지 처리한다.
하나의 LLM에게 모든 역할을 맡기는 대신, 각각의 역할(Role)에 특화된 에이전트들을 만들어 협업시키는 방식, 모두 GPT-5.5를 사용한다고 해도 역할이 다르기 때문에 프롬프트도 다르다. 모델은 같지만 역할(Role)과 지시(System Prompt)가 다르므로 행동이 달라지는것. 하나의 LLM에게 여러가지일을 한번에 시키면 프롬프트가 길어지고 책임이 섞여 품질이 떨어질수있다. 역할을 나누어 각 에이전트의 관심사를 분리한다. supervisor는 어떤 agent를 호출할지, 순서를 결정해 합쳐 사용자에게 반환한다. LangGraph에서 각 Agent를 하나의 그래프나 노드로 만들며 supervisor가 작업을 넘긴다 .
이는 Tool과 헷갈릴수있는데 Tool은 특정기능수행으로 LLM이 호출하고 스스로 판단하지않으나 Agent는 tool, memory planner가 들어있을수있어 스스로 생각하고 Tool을 여러번 사용하고 결과를 종합하여 최종답을 만든다.
함수는 정해진 로직만 수행하나 agent는 내부적으로 여러 단계의 추론과 의사결정을 수행할 수 있다.
예를 들어 csv가 있다고 하자. 이것을 planner에게 넘긴다. 보통 Pydantic를 사용해 planner에게 할일을 부여한다. 이때 실행하지말라고 명령하며 작업 목록만 만들도록한다.
PLANNER_SYSTEM = """
너는 Planner이다.
작업을 여러 개의 독립적인 작업으로 나누어라.
규칙
- 실행하지 마라.
- 분석만 하라.
- 겹치지 않게 나누어라.
- 명령형 문장으로 작성하라.
"""
with_structured_output로 실행하면 invoke결과가 리스트로 떨어진다 .
subtasks로 가져와 각 worker에게 전달해 하나씩 처리하게된다.
LLM 일반출력에서는 형식을 잘 안지킬수있ㅈ만 class와 with_structured_output를 사용하면 스키마에 맞는 객체를 생성하는것.
Planner를 사용하면 서로 의존성이 없어 동시에 실행할 수 있다. 병렬화
순차실행과 병렬실행모두 가능
실행은 가장 오래걸리는 작업시간정도만 기다리면 된다.
이때 만약 시장조사 - 조사결과분석 처럼 앞 결과가 다음단계 입력이라면 순차 실행해야한다.
results = []
for task in plan.subtasks:
result = worker.invoke(task)
results.append(result)from concurrent.futures import ThreadPoolExecutor
def run_task(task):
return worker.invoke(task)
with ThreadPoolExecutor(max_workers=3) as executor:
results = list(executor.map(run_task, plan.subtasks))
이렇게 결과를 받았다면 마지막 역할은 취합이다. Aggregation
해당 기능으로 다시 결과로 하나의 보고서를 만들라는등의 LLM을 넘길 수도 있다.
단순취합시 LLM이 필요없으니 통합할경우 LLM이 더잘한다.
return {
"research": result1,
"analysis": result2,
"copy": result3
}from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
(
"system",
"""
당신은 결과를 취합하는 전문가이다.
여러 Worker의 결과를
중복 없이
자연스럽게
하나의 최종 답변으로 작성하라.
"""
),
("human", "{results}")
])
synthesizer = prompt | llm
결국 planner, worker, synthesizer는 특별한 라이브러리 객체가 아니라 역할을 부여한 LLM 체인(또는 Agent)이며, 이름도 개발자가 자유롭게 정해서 사용하는 변수이다.
이때 for문등을 활용해 실패처리 및 재시도를 명령할 수도있다 .
MAX_RETRY = 3
for task in plan.subtasks:
for attempt in range(MAX_RETRY):
try:
result = worker.invoke(task)
break
except Exception:
if attempt == MAX_RETRY - 1:
result = "실패"
실무에서는 add_node할때 retry 재시도정책을 정의한다.
RetryPolicy는 LangGraph가 제공하는 클래스로 from langgraph.types import RetryPolicy를 해 사용한다.
max_attempts=3, 최대 몇번 시도할
initial_interval=1.0,첫 재시도 후 몇초 기다릴지
backoff_factor=2.0,재 시도 간격을 얼마나 늘릴지
max_interval=30.0, 몇 초까지만 기다릴지
retry_on=... 어떤 예외일때만 재시도할지 TimeoutError만 재시도하고 valueerror는 바로 실패처리할때 등
RetryPolicy(
max_attempts=3,
initial_interval=1.0,
backoff_factor=2.0,
max_interval=30.0,
retry_on=...
)from typing import TypedDict
from langgraph.graph import StateGraph
from langgraph.types import RetryPolicy
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4.1")
# Graph State
class State(TypedDict):
task: str
result: str
status: str
retry_count: int
# Worker Node
def worker_node(state: State):
response = llm.invoke(state["task"])
return {
"result": response.content,
"status": "SUCCESS",
}
builder = StateGraph(State)
builder.add_node(
"worker",
worker_node,
retry_policy=RetryPolicy(
max_attempts=3
)
)
builder.set_entry_point("worker")
graph = builder.compile()
샘플 프로젝트 planner_worker_console_project 를 분석해보자.
Worker Role Agent 콘솔 프로젝트는 환경변수와 캠페인 csv를 확인해 planner 구조화출력으로 하위작업을 분해해 worker 단일 작업을 실행하고 결과를 취합한다. 이때 오류시 재시도, 스킵, 부분 실패 기록을한다.
common
import os, path, dotenv
환경변수지정, key확인provicer 확인해 openai혹은 gemini chatai 객체 만들어 반환
# -*- coding: utf-8 -*-
"""API 키, 모델 생성, 공통 경로를 관리하는 모듈입니다."""
# 운영체제 환경변수를 읽기 위해 os 모듈을 가져옵니다.
import os
# 파일과 폴더 경로를 운영체제에 독립적으로 처리하기 위해 Path를 가져옵니다.
from pathlib import Path
# .env 파일의 값을 환경변수로 등록하기 위해 load_dotenv를 가져옵니다.
from dotenv import load_dotenv
# 현재 파일(code/common.py)의 상위 상위 폴더를 프로젝트 루트로 지정합니다.
ROOT = Path(__file__).resolve().parent.parent
# 프로젝트의 실습 데이터 폴더 경로를 지정합니다.
DATA = ROOT / "data"
# 실행 결과를 저장할 폴더 경로를 지정합니다.
OUTPUTS = ROOT / "outputs"
# 프로젝트 루트의 .env 파일을 읽어 API 키와 모델 설정을 로드합니다.
load_dotenv(ROOT / ".env")
# 환경변수에 OpenAI 모델이 없으면 교육용 기본 모델명을 사용합니다.
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
# 환경변수에 Gemini 모델이 없으면 교육용 기본 모델명을 사용합니다.
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
def require_key(name: str) -> str:
"""필수 API 키가 실제 값으로 설정되었는지 검사합니다."""
# 지정한 이름의 환경변수 값을 읽습니다.
value = os.getenv(name, "").strip()
# 값이 없거나 예제 문구가 남아 있으면 명확한 설정 오류를 발생시킵니다.
if not value or value.startswith("여기에"):
raise RuntimeError(f"[설정 필요] .env 파일에 {name}의 실제 값을 입력하세요.")
# 검증된 API 키 문자열을 호출한 위치에 반환합니다.
return value
def get_chat(provider: str, temperature: float = 0.0):
"""선택한 provider에 맞는 LangChain 채팅 모델을 생성합니다."""
# 사용자가 입력한 공급자 이름의 앞뒤 공백을 제거하고 소문자로 통일합니다.
normalized = provider.strip().lower()
# OpenAI가 선택되었는지 확인합니다.
if normalized == "openai":
# OpenAI API 키가 설정되었는지 먼저 검사합니다.
require_key("OPENAI_API_KEY")
# OpenAI 채팅 모델 클래스를 필요한 시점에만 가져옵니다.
from langchain_openai import ChatOpenAI
# .env의 모델명과 전달받은 temperature로 모델 객체를 반환합니다.
return ChatOpenAI(model=OPENAI_MODEL, temperature=temperature)
# Gemini가 선택되었는지 확인합니다.
if normalized == "gemini":
# Google API 키가 설정되었는지 먼저 검사합니다.
require_key("GOOGLE_API_KEY")
# Gemini 채팅 모델 클래스를 필요한 시점에만 가져옵니다.
from langchain_google_genai import ChatGoogleGenerativeAI
# .env의 모델명과 전달받은 temperature로 모델 객체를 반환합니다.
return ChatGoogleGenerativeAI(model=GEMINI_MODEL, temperature=temperature)
# 지원하지 않는 공급자 이름이면 허용값을 포함한 오류를 발생시킵니다.
raise ValueError("provider는 'openai' 또는 'gemini'여야 합니다.")
def print_environment_status() -> None:
"""프로젝트 경로, 데이터, API 키 설정 여부를 화면에 표시합니다."""
# 프로젝트 루트 경로를 출력합니다.
print(f"프로젝트 루트: {ROOT}")
# 데이터 폴더의 존재 여부를 출력합니다.
print(f"데이터 폴더: {DATA} / 존재={DATA.exists()}")
# OpenAI 키는 실제 값을 노출하지 않고 설정 여부만 출력합니다.
print(f"OPENAI_API_KEY 설정: {bool(os.getenv('OPENAI_API_KEY'))}")
# Google 키도 실제 값을 노출하지 않고 설정 여부만 출력합니다.
print(f"GOOGLE_API_KEY 설정: {bool(os.getenv('GOOGLE_API_KEY'))}")
# 현재 사용할 OpenAI 모델명을 출력합니다.
print(f"OPENAI_MODEL: {OPENAI_MODEL}")
# 현재 사용할 Gemini 모델명을 출력합니다.
print(f"GEMINI_MODEL: {GEMINI_MODEL}")
role_agent
import pandas, pydantic의 basemodel, feild, common
planner와 worker용 프롬프트 지정하고 이때 planner는 직접 실행하지말라고 명명한다.
Plan class를 만들어 list형 subtasks field를 만든다.
파일경로를 구성해 파일이 없으면 error, 있으면 read해 return
이를 읽어 itertyples() 각 행을 typle 형태로 반환해 문자열로 id: title형으로 바꿔 출력한다 .
수령한 매개변수 campain_id와 일치하는 matched 행을 가져와 문자열로 재구성해 return한다 .
with_structured_output() langchain이 제공하는 만들어놓은 plan 스키마를 사용해 LLM에게 ~ 형식으로 답하라는 프롬프르를 자동으로 추가하도록 한다 .이 with_structured_output()은 모든 객체에서 사용할 수 있는 함수가 아닌 chatmodel에서 제공하는 메서드다.
langchain_core humanmessage와 systemmessage 객체를 만들어 invoke(), 결과 return.
work의 경우에도 humanmessage와 systemmessage 객체를 만들어 invoke(), 결과 return.
combine_report의 경우 outputsㄹㄹ for문으로 돌려 문자열로 append, 반환한다.
run_campain에서는 llm_plan, llm_work 객체를 만들어 get_chat(), 계획은 temperature를 0으로 한다.
수령한 txt와 llm_pan으로 plan() 수행해 subtasks를 얻어내고 for문에 돌려 outputs를 정리해 append해 combine_report() 해 ㄱreturn한다.
# -*- coding: utf-8 -*-
"""Planner, Worker, 오케스트레이터의 핵심 실행 코드입니다."""
# 표 형식 CSV 데이터를 읽기 위해 pandas를 가져옵니다.
import pandas as pd
# 구조화 출력 스키마를 정의하기 위해 Pydantic 구성요소를 가져옵니다.
from pydantic import BaseModel, Field
# 공통 데이터 경로와 채팅 모델 생성 함수를 가져옵니다.
from app.common import DATA, get_chat
# Planner가 계획만 수행하도록 역할과 출력 범위를 고정합니다.
PLANNER_SYSTEM = (
"너는 마케팅 캠페인 기획 설계자(Planner)다. "
"브리프를 보고 실행에 필요한 하위 작업을 3~4개로 쪼개라. "
"작업은 겹치지 않는 한 문장 명령형으로 작성하고 직접 실행하지 마라."
)
# Worker가 한 번에 하나의 작업만 수행하도록 역할과 출력 범위를 고정합니다.
WORKER_SYSTEM = (
"너는 마케팅 실무 작업자(Worker)다. "
"브리프 맥락에서 지정된 하나의 작업만 구체적으로 수행하라. "
"결과는 5줄 이내의 실무용 내용으로 작성하라."
)
class Plan(BaseModel):
"""Planner의 응답을 문자열 목록으로 고정하는 구조화 출력 스키마입니다."""
# LLM이 3~4개의 하위 작업을 문자열 리스트에 채우도록 필드를 선언합니다.
subtasks: list[str] = Field(description="캠페인 실행에 필요한 하위 작업 3~4개")
def load_campaign_table() -> pd.DataFrame:
"""marketing_brief.csv 전체를 DataFrame으로 읽습니다."""
# 캠페인 CSV 파일의 전체 경로를 구성합니다.
csv_path = DATA / "marketing_brief.csv"
# 파일이 없으면 사용자가 원인을 바로 알 수 있도록 오류를 발생시킵니다.
if not csv_path.exists():
raise FileNotFoundError(f"캠페인 데이터가 없습니다: {csv_path}")
# UTF-8 CSV를 DataFrame으로 읽어 반환합니다.
return pd.read_csv(csv_path)
def list_campaigns() -> None:
"""선택 가능한 캠페인 목록을 콘솔에 출력합니다."""
# 전체 캠페인 표를 읽습니다.
table = load_campaign_table()
# 각 행을 순서대로 반복합니다.
for row in table.itertuples(index=False):
# 캠페인 ID와 제목을 한 줄로 출력합니다.
print(f"- {row.campaign_id}: {row.title}")
def load_brief_text(campaign_id: str = "CMP02") -> str:
"""캠페인 ID에 해당하는 행을 LLM 입력용 한 줄 텍스트로 변환합니다."""
# 전체 캠페인 표를 읽습니다.
table = load_campaign_table()
# campaign_id가 일치하는 행만 필터링합니다.
matched = table[table["campaign_id"] == campaign_id]
# 일치하는 캠페인이 없으면 사용 가능한 ID를 포함한 오류를 발생시킵니다.
if matched.empty:
available = ", ".join(table["campaign_id"].astype(str).tolist())
raise ValueError(f"캠페인 ID를 찾을 수 없습니다: {campaign_id} / 사용 가능: {available}")
# 첫 번째 일치 행을 Series로 꺼냅니다.
campaign = matched.iloc[0]
# 각 컬럼을 설명형 한 줄 텍스트로 조합해 반환합니다.
return (
f"캠페인명: {campaign['title']} / 카테고리: {campaign['category']} / "
f"타깃: {campaign['target']} / 예산: {int(campaign['budget']):,}원 / "
f"핵심혜택: {campaign['key_offer']}"
)
def plan(llm, brief_text: str, planner_prompt: str = PLANNER_SYSTEM) -> list[str]:
"""브리프를 구조화된 하위 작업 목록으로 분해합니다."""
# 메시지 클래스를 실제 LLM 호출 시점에만 가져옵니다.
from langchain_core.messages import HumanMessage, SystemMessage
# 일반 채팅 모델을 Plan 스키마의 구조화 출력 모델로 감쌉니다.
planner = llm.with_structured_output(Plan)
# Planner 역할 지시와 캠페인 브리프를 모델에 전달합니다.
result = planner.invoke([SystemMessage(planner_prompt), HumanMessage(f"[캠페인 브리프]\n{brief_text}")])
# Pydantic 객체에서 하위 작업 문자열 리스트를 꺼내 반환합니다.
return result.subtasks
def work(llm, brief_text: str, subtask: str, worker_prompt: str = WORKER_SYSTEM) -> str:
"""브리프 맥락에서 지정된 하위 작업 하나만 수행합니다."""
# 메시지 클래스를 실제 LLM 호출 시점에만 가져옵니다.
from langchain_core.messages import HumanMessage, SystemMessage
# Worker 역할과 전체 맥락 및 단일 작업을 모델에 전달합니다.
result = llm.invoke([
SystemMessage(worker_prompt),
HumanMessage(f"[캠페인 브리프]\n{brief_text}\n\n[수행할 작업]\n{subtask}"),
])
# 모델 응답 객체의 본문을 문자열로 변환해 반환합니다.
return str(result.content)
def combine_report(brief_text: str, outputs: list[tuple[str, str]]) -> str:
"""Worker 산출물을 Markdown 형식의 실행안으로 취합합니다."""
# 문서 제목과 원본 브리프를 첫 번째 섹션으로 생성합니다.
sections = [f"# 캠페인 실행안\n\n> {brief_text}\n"]
# 작업명과 산출물을 순서 번호와 함께 반복합니다.
for index, (task, output) in enumerate(outputs, start=1):
# 각 작업을 별도의 Markdown 섹션으로 추가합니다.
sections.append(f"## {index}. {task}\n{output}\n")
# 섹션 사이에 줄바꿈을 넣어 하나의 문자열로 반환합니다.
return "\n".join(sections)
def run_campaign(provider: str, brief_text: str) -> str:
"""선택 provider로 계획, 실행, 취합 전체 과정을 수행합니다."""
# 계획은 일관성이 중요하므로 temperature를 0으로 설정합니다.
llm_plan = get_chat(provider=provider, temperature=0.0)
# 실행은 표현 다양성이 필요할 수 있으므로 temperature를 0.5로 설정합니다.
llm_work = get_chat(provider=provider, temperature=0.5)
# Planner로 하위 작업 목록을 생성합니다.
subtasks = plan(llm_plan, brief_text)
# 작업별 산출물을 저장할 빈 리스트를 준비합니다.
outputs: list[tuple[str, str]] = []
# Planner가 만든 작업을 하나씩 순회합니다.
for index, subtask in enumerate(subtasks, start=1):
# 현재 수행 중인 작업을 콘솔에 표시합니다.
print(f"[{index}/{len(subtasks)}] Worker 수행: {subtask}")
# 한 작업의 결과를 작업명과 함께 리스트에 저장합니다.
outputs.append((subtask, work(llm_work, brief_text, subtask)))
# 모든 산출물을 하나의 실행안으로 취합해 반환합니다.
return combine_report(brief_text, outputs)
retry_agent
trackback 예외발생시 상세원인을 출력하기 위한.
작성한 common, role_agent, retry_agent, reviewer_exercise, yaml_exercise를 가져온다.
재시도 횟수를 2회로 지정하고
for문을 돌려 work() 수행이 잘되면 return except일때는 게속 순회하되 max_retry를 attempt에 반영해 기록한다. 모두 실패하면 error_text를 반환하자.
llm_plan, llm_work get_chat() 모델을 만들어 plan() 작업목록을 생성하고 이 작업 subtasks를 순회하여 work_with_retry() 정의한 함수로 3번 반복하게 된다. 성공할경우 ouputs에 apend(), 실패의 경우 failed.append()를 하자. 이 outputs를 combine)report 합쳐 retirn.
failed의 경우 failed를 for문으로 돌려 실패 기록을 포함한 보고서를 반환한다.
# -*- coding: utf-8 -*-
"""Worker 실패 시 재시도하고 부분 실패를 허용하는 실습 코드입니다."""
# 공통 모델 생성 함수를 가져옵니다.
from app.common import get_chat
# 핵심 Planner/Worker/취합 함수를 재사용합니다.
from app.role_agent import combine_report, plan, work
# 최초 호출 이후 허용할 추가 재시도 횟수를 2회로 지정합니다.
MAX_RETRY = 2
def work_with_retry(llm, brief_text: str, subtask: str, index: int, total: int, force_fail: bool = False) -> tuple[bool, str]:
"""Worker를 최대 3회 시도하고 성공 여부와 결과를 반환합니다."""
# 최초 1회와 재시도 2회를 합친 총 3회를 순회합니다.
for attempt in range(1, MAX_RETRY + 2):
try:
# 작업 번호와 현재 시도 횟수를 출력합니다.
print(f"작업 {index}/{total}, 시도 {attempt}/{MAX_RETRY + 1}")
# 메뉴 시연에서 선택한 작업은 실제 API 호출 전에 모의 오류를 발생시킵니다.
if force_fail:
raise TimeoutError("재시도 동작 확인을 위한 모의 오류")
# 모의 실패 대상이 아니면 실제 Worker를 호출합니다.
return True, work(llm, brief_text, subtask)
except Exception as error:
# 오류 형식과 메시지를 기록합니다.
error_text = f"{type(error).__name__}: {error}"
# 현재 오류를 사용자에게 출력합니다.
print(f" 실패: {error_text}")
# 마지막 시도가 아니면 다음 반복에서 재시도합니다.
if attempt <= MAX_RETRY:
print(" 재시도합니다.")
else:
# 모든 시도가 실패하면 False와 오류 문자열을 반환합니다.
return False, error_text
# 논리상 도달하지 않지만 타입 검사기를 위해 기본 실패값을 반환합니다.
return False, "알 수 없는 오류"
def run_campaign_robust(provider: str, brief_text: str, fail_indices: set[int] | None = None) -> str:
"""실패 작업을 기록하고 나머지 작업을 끝까지 실행합니다."""
# 전달값이 없으면 빈 집합으로 바꾸어 포함 여부 검사를 단순화합니다.
fail_indices = fail_indices or set()
# 계획용 모델을 생성합니다.
llm_plan = get_chat(provider=provider, temperature=0.0)
# 실행용 모델을 생성합니다.
llm_work = get_chat(provider=provider, temperature=0.5)
# Planner로 작업 목록을 생성합니다.
subtasks = plan(llm_plan, brief_text)
# 성공한 작업 결과를 저장할 리스트를 만듭니다.
outputs: list[tuple[str, str]] = []
# 실패한 작업 정보를 저장할 리스트를 만듭니다.
failed: list[tuple[int, str, str]] = []
# 모든 하위 작업을 순회합니다.
for index, subtask in enumerate(subtasks, start=1):
# 선택한 작업 번호이면 force_fail을 True로 전달합니다.
ok, information = work_with_retry(llm_work, brief_text, subtask, index, len(subtasks), index in fail_indices)
# 성공 여부에 따라 결과 저장 위치를 나눕니다.
if ok:
outputs.append((subtask, information))
else:
failed.append((index, subtask, information))
# 성공한 결과를 기본 보고서로 취합합니다.
report = combine_report(brief_text, outputs)
# 실패 작업이 있으면 미완료 섹션을 추가합니다.
if failed:
report += "\n## 미완료 작업\n"
# 실패한 각 작업의 번호, 내용, 오류를 한 줄씩 추가합니다.
for index, subtask, error in failed:
report += f"- 원래 작업 {index}: {subtask} — {error}\n"
# 부분 성공 결과와 실패 기록을 포함한 보고서를 반환합니다.
return report
reviewer_exercise
import common, role_agent를 가져온다 .
reviewer의 프롬프트 서리정
humanmessage, systemmessage를 가져와 llm.invoke() 역할지시와 systemmessage를 첨부해 전달하고 결과 return.
llm_plan, llm_work, llm)review get_chat() 객체를 만들어 각각 수행하고 combine, review를 수행하고 return.
# -*- coding: utf-8 -*-
"""실습문제 1: Planner → Worker 뒤에 Reviewer를 추가합니다."""
# provider별 모델 생성 함수를 가져옵니다.
from app.common import get_chat
# 기존 계획, 실행, 취합 함수를 재사용합니다.
from app.role_agent import combine_report, plan, work
# Reviewer가 새 실행안을 만들지 않고 누락과 개선점만 제시하도록 지시합니다.
REVIEWER_SYSTEM = (
"너는 마케팅 캠페인 검토자(Reviewer)다. "
"실행안 전체를 읽고 누락되었거나 보완할 점을 구체적으로 3가지 이상 제시하라. "
"새 실행안을 작성하지 말고 검토 의견만 작성하라."
)
def review(llm, report: str, reviewer_prompt: str = REVIEWER_SYSTEM) -> str:
"""완성된 실행안 전체를 검토해 개선 의견을 반환합니다."""
# 메시지 클래스를 실제 검토 호출 시점에만 가져옵니다.
from langchain_core.messages import HumanMessage, SystemMessage
# Reviewer 역할 지시와 검토 대상 보고서를 모델에 전달합니다.
result = llm.invoke([SystemMessage(reviewer_prompt), HumanMessage(f"[검토할 캠페인 실행안]\n{report}")])
# 응답 본문을 문자열로 반환합니다.
return str(result.content)
def run_campaign_with_review(provider: str, brief_text: str) -> str:
"""계획, 실행, 취합, 검토의 네 단계를 순서대로 수행합니다."""
# 계획용 모델을 재현성이 높은 설정으로 생성합니다.
llm_plan = get_chat(provider=provider, temperature=0.0)
# 실행용 모델을 약간 창의적인 설정으로 생성합니다.
llm_work = get_chat(provider=provider, temperature=0.5)
# 검토용 모델을 일관성이 높은 설정으로 생성합니다.
llm_review = get_chat(provider=provider, temperature=0.0)
# Planner로 하위 작업을 생성합니다.
subtasks = plan(llm_plan, brief_text)
# 각 하위 작업을 Worker로 실행해 튜플 리스트로 저장합니다.
outputs = [(subtask, work(llm_work, brief_text, subtask)) for subtask in subtasks]
# Worker 결과를 하나의 보고서로 취합합니다.
report = combine_report(brief_text, outputs)
# 취합된 전체 보고서를 Reviewer에게 전달합니다.
opinion = review(llm_review, report)
# 원본 보고서 끝에 검토 의견 섹션을 붙여 반환합니다.
return report + f"\n## 검토 의견\n{opinion}\n"
yaml_exercise
import yaml, common과 role_agent, reviewer_exervcise 가져오기
프로그램 설정을 사람이 읽기 쉽게 적어놓은 파일인 yaml 파일 경로를 확인해 safe_load()로 파일을 읽는다. 임의 객체로 읽어 각 요소가 있는지 확인한다. 참고로 yaml파일은 아래처럼 구성되었다.
# 역할 프롬프트를 코드 밖에서 관리합니다.
planner: |
너는 마케팅 캠페인 '기획 설계자(Planner)'다.
주어진 캠페인 브리프를 보고 실행에 필요한 하위 작업을 3~4개로 쪼개라.
작업은 서로 겹치지 않게 한 문장 명령형으로 작성하고, 직접 실행하지 마라.
worker: |
너는 마케팅 '실무 작업자(Worker)'다.
캠페인 브리프의 맥락에서 지정된 하나의 작업만 구체적으로 수행하라.
결과는 5줄 이내로 실무에 바로 사용할 수 있게 작성하라.
reviewer: |
너는 마케팅 캠페인 '검토자(Reviewer)'다.
실행안 전체에서 누락되었거나 보완할 점을 구체적으로 3가지 이상 제시하라.
새 실행안을 작성하지 말고 검토 의견만 작성하라.
이를 사전형식으로 return한다.
prompt를 읽어 llm_plan, work, review get_chat를 만들어 plan형 스키마로 강제하기위해 with_structured_output()을 사용한다. invoke() 실행한다.
work는 list형에 for문을 순회해 invoke해 할당하고 output에 () 튜플로 저장한다 . 튜플로 저장하는 이유는 작업명과 결과를 한쌍으로 묶기 위해서이다 .이를 combine_report() 해 review() 프롬프트로 검토해 합쳐 반환한다.
# -*- coding: utf-8 -*-
"""실습문제 2: 역할 프롬프트를 YAML 파일로 외부화합니다."""
# YAML 파일을 안전하게 읽기 위해 yaml 모듈을 가져옵니다.
import yaml
# provider별 모델과 데이터 경로를 가져옵니다.
from app.common import DATA, get_chat
# 기존 스키마, 취합, 작업 실행 함수를 재사용합니다.
from app.role_agent import Plan, combine_report
# Reviewer 함수도 재사용하되 YAML 프롬프트를 전달합니다.
from app.reviewer_exercise import review
def load_prompts() -> dict[str, str]:
"""data/role_prompts.yaml의 역할별 프롬프트를 읽습니다."""
# YAML 파일 경로를 구성합니다.
prompt_path = DATA / "role_prompts.yaml"
# UTF-8 텍스트 모드로 YAML 파일을 엽니다.
with prompt_path.open(mode="r", encoding="utf-8") as file:
# 임의 객체 생성을 막는 safe_load로 딕셔너리를 읽습니다.
loaded = yaml.safe_load(file)
# 필수 역할 키가 모두 있는지 검사합니다.
required = {"planner", "worker", "reviewer"}
# 누락 키의 집합을 계산합니다.
missing = required - set(loaded or {})
# 하나라도 빠졌으면 설정 오류를 발생시킵니다.
if missing:
raise ValueError(f"YAML에 역할 프롬프트가 누락되었습니다: {sorted(missing)}")
# 타입 검사기가 값을 문자열로 인식하도록 새 딕셔너리로 변환합니다.
return {key: str(value) for key, value in loaded.items()}
def run_yaml_campaign(provider: str, brief_text: str) -> str:
"""YAML에서 읽은 프롬프트로 계획, 실행, 검토를 수행합니다."""
# 메시지 클래스를 실제 LLM 호출 시점에만 가져옵니다.
from langchain_core.messages import HumanMessage, SystemMessage
# 실행할 때마다 YAML을 다시 읽어 파일 수정 내용이 즉시 반영되게 합니다.
prompts = load_prompts()
# 역할별로 필요한 세 모델 객체를 생성합니다.
llm_plan = get_chat(provider=provider, temperature=0.0)
llm_work = get_chat(provider=provider, temperature=0.5)
llm_review = get_chat(provider=provider, temperature=0.0)
# Planner 출력을 Plan 스키마로 강제합니다.
planner = llm_plan.with_structured_output(Plan)
# YAML의 planner 프롬프트로 하위 작업을 생성합니다.
plan_result = planner.invoke([SystemMessage(prompts["planner"]), HumanMessage(f"[캠페인 브리프]\n{brief_text}")])
# Worker 결과를 저장할 리스트를 준비합니다.
outputs: list[tuple[str, str]] = []
# 생성된 각 하위 작업을 순회합니다.
for subtask in plan_result.subtasks:
# YAML의 worker 프롬프트로 한 작업을 실행합니다.
result = llm_work.invoke([
SystemMessage(prompts["worker"]),
HumanMessage(f"[브리프]\n{brief_text}\n\n[작업]\n{subtask}"),
])
# 작업명과 결과 본문을 튜플로 저장합니다.
outputs.append((subtask, str(result.content)))
# Worker 결과를 실행안으로 취합합니다.
report = combine_report(brief_text, outputs)
# YAML의 reviewer 프롬프트로 전체 실행안을 검토합니다.
opinion = review(llm_review, report, prompts["reviewer"])
# 실행안과 검토 의견을 합쳐 반환합니다.
return report + f"\n## 검토 의견\n{opinion}\n"
console_app.py
import traceback, common, role_agent, retry_agent, reviewer_exervise, yaml_exercise
프로바이더에 맞게 메뉴를 출력하고 id를 입력받아 해당 캡페인 목록을 출력한다.
menu도 게속해서 while출력하되 0이면 종료 각 메뉴에 맞게 메뉴를 처리한다 .
# -*- coding: utf-8 -*-
"""OpenAI와 Gemini를 선택해 각 실습을 실행하는 콘솔 메뉴입니다."""
# 예외 발생 시 상세 원인을 출력하기 위해 traceback을 가져옵니다.
import traceback
# 환경 상태 출력 함수를 가져옵니다.
from app.common import print_environment_status
# 기본 Planner/Worker 기능을 가져옵니다.
from app.role_agent import list_campaigns, load_brief_text, plan, run_campaign, work
# 재시도 실습 함수를 가져옵니다.
from app.retry_agent import run_campaign_robust
# Reviewer 실습문제 완성 함수를 가져옵니다.
from app.reviewer_exercise import run_campaign_with_review
# YAML 프롬프트 실습문제 완성 함수를 가져옵니다.
from app.yaml_exercise import run_yaml_campaign
# provider별 채팅 모델 생성을 위해 공통 함수를 가져옵니다.
from app.common import get_chat
def select_provider() -> str | None:
"""OpenAI 또는 Gemini를 선택하고 종료 요청이면 None을 반환합니다."""
# 공급자 선택 메뉴를 출력합니다.
print("\n[LLM 선택] 1. OpenAI 2. Gemini 0. 종료")
# 사용자 입력의 앞뒤 공백을 제거합니다.
choice = input("선택: " ).strip()
# 메뉴 번호를 내부 provider 문자열로 변환합니다.
providers = {"1": "openai", "2": "gemini"}
# 0을 입력하면 상위 반복문을 종료하도록 None을 반환합니다.
if choice == "0":
return None
# 존재하지 않는 번호이면 None과 구분되는 빈 문자열을 반환합니다.
return providers.get(choice, "")
def select_campaign() -> str:
"""캠페인 목록을 보여주고 선택한 브리프 문자열을 반환합니다."""
# 데이터에 들어 있는 캠페인 목록을 출력합니다.
list_campaigns()
# 기본값 CMP02를 안내하며 캠페인 ID를 입력받습니다.
campaign_id = input("캠페인 ID(Enter=CMP02): " ).strip() or "CMP02"
# 선택한 ID의 브리프를 문자열로 변환해 반환합니다.
return load_brief_text(campaign_id)
def run_provider_menu(provider: str) -> None:
"""선택한 provider를 유지하면서 하위 실습 메뉴를 반복 실행합니다."""
# 뒤로 가기를 선택할 때까지 하위 메뉴를 계속 표시합니다.
while True:
# HTML 요약 기능을 제외한 실행 확인 메뉴만 출력합니다.
print(f"\n[{provider.upper()} 실습 메뉴]")
print("1. 환경·데이터 설정 확인")
print("2. Planner 작업 분해")
print("3. Worker 단일 작업 실행")
print("4. Planner → Worker → 취합 전체 실행")
print("5. Worker 실패 재시도·스킵 실습")
print("6. 실습문제 1: Reviewer 단계 추가")
print("7. 실습문제 2: YAML 프롬프트 외부화")
print("0. LLM 다시 선택")
# 실행할 메뉴 번호를 입력받습니다.
choice = input("선택: " ).strip()
# 0이면 현재 provider의 하위 메뉴를 종료합니다.
if choice == "0":
return
try:
# 환경 및 데이터 경로 확인 메뉴를 처리합니다.
if choice == "1":
print_environment_status()
list_campaigns()
# Planner만 단독으로 실행합니다.
elif choice == "2":
brief = select_campaign()
llm = get_chat(provider=provider, temperature=0.0)
for index, subtask in enumerate(plan(llm, brief), start=1):
print(f"{index}. {subtask}")
# Worker에 사용자가 직접 입력한 작업 하나를 전달합니다.
elif choice == "3":
brief = select_campaign()
subtask = input("Worker가 수행할 한 작업: " ).strip() or "타깃 고객의 핵심 니즈를 분석하라"
llm = get_chat(provider=provider, temperature=0.5)
print(work(llm, brief, subtask))
# 전체 기본 오케스트레이션을 실행합니다.
elif choice == "4":
print(run_campaign(provider, select_campaign()))
# 두 번째 작업을 고의로 실패시켜 재시도와 부분 실패를 확인합니다.
elif choice == "5":
print(run_campaign_robust(provider, select_campaign(), fail_indices={2}))
# Reviewer 추가 실습문제 완성 코드를 실행합니다.
elif choice == "6":
print(run_campaign_with_review(provider, select_campaign()))
# YAML 프롬프트 외부화 실습문제 완성 코드를 실행합니다.
elif choice == "7":
print(run_yaml_campaign(provider, select_campaign()))
# 정의되지 않은 메뉴 번호를 안내합니다.
else:
print("메뉴 번호를 다시 입력하세요.")
except Exception as error:
# 사용자가 수정할 수 있도록 오류 종류와 메시지를 출력합니다.
print(f"[실행 오류] {type(error).__name__}: {error}")
# 개발 단계에서 상세 호출 위치까지 확인할 수 있게 traceback을 출력합니다.
traceback.print_exc()
def main() -> None:
"""프로그램 시작점에서 LLM 선택 메뉴를 반복 실행합니다."""
# 프로그램의 목적을 한 번 출력합니다.
print("Planner/Worker Role Agent 콘솔 실습 프로젝트")
# 사용자가 종료할 때까지 공급자 선택을 반복합니다.
while True:
# OpenAI 또는 Gemini 선택 결과를 받습니다.
provider = select_provider()
# None은 명시적인 종료 요청입니다.
if provider is None:
print("프로그램을 종료합니다.")
return
# 빈 문자열은 잘못된 메뉴 번호입니다.
if not provider:
print("1, 2, 0 중에서 선택하세요.")
continue
# 선택한 공급자의 하위 실습 메뉴를 실행합니다.
run_provider_menu(provider)
# 이 파일을 직접 실행했을 때만 main 함수를 호출합니다.
if __name__ == "__main__":
main()
main.py
app.console에 main 함수를 가져와 실행한다.
# -*- coding: utf-8 -*-
"""PyCharm에서 이 파일을 실행하면 콘솔 메뉴가 시작됩니다."""
# 실제 메뉴의 main 함수를 가져옵니다.
from app.console_app import main
# 이 파일을 직접 실행한 경우에만 콘솔 프로그램을 시작합니다.
if __name__ == "__main__":
main()
다른 프로젝트 ai_team_console_project도 분석해보자.
langgraph ai team pycharm 콘솔 프로젝트는 경쟁사 csv 데이터를 로드해 researcher, analyst, writer노드가 있고 check해 조건부 품질을 검토하고 중간산출물을 저장한다.
reviewer에 finalize 노드를 추가하고,
네단계의 결과를 output에 저장하는 것도 추가로 확인해보자.
common_bridge.py
import pathlib, sys, common
모든 모듈이 명시적으로 사용할 이름을 정의한다.
# -*- coding: utf-8 -*-
"""제공된 code/common.py를 수정하지 않고 불러오기 위한 연결 모듈입니다."""
# pathlib는 프로젝트의 절대 경로를 안전하게 계산할 때 사용합니다.
import pathlib
# sys는 제공된 common.py가 있는 code 폴더를 import 검색 경로에 넣을 때 사용합니다.
import sys
# 현재 파일(code_app/common_bridge.py)의 부모의 부모가 프로젝트 루트입니다.
PROJECT_ROOT = pathlib.Path(__file__).resolve().parent.parent
# 사용자가 제공한 common.py를 그대로 둔 폴더의 경로입니다.
COMMON_DIR = PROJECT_ROOT / "code"
# 같은 경로가 중복 등록되지 않도록 확인한 다음 Python 모듈 검색 경로에 추가합니다.
if str(COMMON_DIR) not in sys.path:
# 목록 앞쪽에 넣어 프로젝트의 common.py가 우선 검색되게 합니다.
sys.path.insert(0, str(COMMON_DIR))
# 제공된 공통 모듈의 경로 상수와 모델 생성 함수를 그대로 가져옵니다.
from common import DATA, ROOT, get_chat # noqa: E402
# 다른 모듈이 명시적으로 사용할 공개 이름을 정의합니다.
__all__ = ["DATA", "ROOT", "get_chat", "PROJECT_ROOT"]
ai_team.py
collections.abc를 import 해 함수만 받을 수 있음을 명시하도록한다. 안써도 프로그램은 돌아가지만 ide나 타입검사기 mypy등이 같은 형태인지 확인해주기때문에 자동완성, 타입오류 검출, 코드 가독성이 훨씬 좋아진다. collections.abc에는 callable외에도 iterable, iterator, mapping, sequence등이 많이 사용되고 langgraph나 langchain코드에서는 callable이 가장 자주 등장한다. Node = Callable[[State], State]라 하면 node리스트에 state를 받아 state를 반환하는 노드 함수만 넣을 수 있다는 의미이다.
typing typeddict langgraph에서 공유할 상태 딕셔너리 키와 타입을 선언, state딕셔너리에 어떤 키가 있고 각 키의 타입이 무엇인지를 정의하는 클래스이다. 쓰지않고 사용할 수 있으나 typeddict를 사용하면 ide가 각 형을 알고있어 아무키나 가능했던 것들이 형에 맞춰 에러검출하기가 좋아진다. langgraph에서는 여러 노드가 하나의 state를 공유하기에 state구조를 미리 정의하여 docements와 같이 오타를 내면 ide가 미리 알려주고, answer에 문자열 대신 숫자를 넣으면 타입경고도 받을 수 있다. 설계도와 다름이 없다. 각 노드는 이 state를 입력받아 수정한뒤 다시 sttate를 반환하며 안전하고 일관되게 데이터를 공유한다.
pandas, langchain_core.message, langrph,.graph
code_app.common_bridge 가져오기
typedict로 teamstate를 생성한다.
csv를 읽고 각 행을 llm이 읽기 좋도록 자연어로 정레해 return한다 .
모델객체를 생성한다. get_chat()
system 프롬프트를 정의해 llm.invoke()해 output을 리턴한다.
analyst도.
writer도.
이 세개를 함께 return한다.
각 모델 공급자를 사용하는 노드를 생성한다. create_nodes()
stategraph() 를 생성한다.
add_node() 각 노드 추가.
graph.add_edge() 경로 지정 compile() 해 return.
build_team() 위 코드를 거쳐 raw에 텍스트를 넣고 나머지를 비워주다.
teamstate - initial.copy() 초기상태 복사
for문을 순회하여 team.stream() current_state를 final_state에 할당하고 각 단계가 존재하면 단계 완료라고 표시한다.
final_state를 return한다 .
# -*- coding: utf-8 -*-
"""ai_team.py를 기반으로 OpenAI/Gemini 선택이 가능하도록 확장한 핵심 모듈입니다."""
# Callable은 노드 함수의 타입을 명확히 표현할 때 사용합니다.
from collections.abc import Callable
# TypedDict는 LangGraph에서 공유할 상태 딕셔너리의 키와 타입을 선언합니다.
from typing import TypedDict
# pandas는 제공된 competitor_data.csv 파일을 읽습니다.
import pandas as pd
# HumanMessage와 SystemMessage는 LLM에 역할 지시와 실제 입력을 구분해 전달합니다.
from langchain_core.messages import HumanMessage, SystemMessage
# START와 END는 그래프의 시작·종료점이고 StateGraph는 노드와 엣지를 선언합니다.
from langgraph.graph import END, START, StateGraph
# 제공된 common.py의 DATA와 get_chat을 수정 없이 사용합니다.
from code_app.common_bridge import DATA, get_chat
class TeamState(TypedDict):
"""Researcher, Analyst, Writer가 함께 사용하는 공유 상태의 기본 구조입니다."""
# raw는 최초에는 CSV 원문 텍스트이고 Researcher 실행 후에는 정리된 사실이 됩니다.
raw: str
# analysis는 Analyst가 정리된 사실을 바탕으로 만든 경쟁사 분석입니다.
analysis: str
# report는 Writer가 분석 결과로 작성한 최종 마크다운 보고서입니다.
report: str
def load_competitor_text() -> str:
"""제공된 경쟁사 CSV를 읽어 LLM이 이해하기 쉬운 불릿 텍스트로 변환합니다."""
# common.py의 DATA 경로 아래에서 원본 CSV 파일을 읽습니다.
csv_path = DATA / "competitor_data.csv"
# 필수 데이터가 없을 때 원인을 바로 알 수 있도록 명확한 예외를 발생시킵니다.
if not csv_path.exists():
raise FileNotFoundError(f"경쟁사 데이터 파일을 찾을 수 없습니다: {csv_path}")
# CSV의 각 열을 유지한 채 pandas DataFrame으로 불러옵니다.
dataframe = pd.read_csv(csv_path)
# 각 행을 LLM이 읽기 쉬운 한 줄 설명으로 변환해 리스트에 저장합니다.
lines = [
# itertuples의 각 행에서 회사명, 주력 분야, 평점, 강점, 약점을 꺼냅니다.
f"- {row.company}: 주력={row.main_category}, 평점={row.rating}, "
f"강점={row.strength}, 약점={row.weakness}"
for row in dataframe.itertuples()
]
# 제목과 모든 경쟁사 행을 줄바꿈으로 연결해 하나의 입력 문자열로 반환합니다.
return "승승장구몰 경쟁사 현황:\n" + "\n".join(lines)
def create_nodes(provider: str, temperature: float = 0.2) -> tuple[Callable, Callable, Callable]:
"""선택한 provider의 LLM을 사용하는 Researcher·Analyst·Writer 노드를 생성합니다."""
# common.py의 get_chat에 openai 또는 gemini를 전달해 해당 모델 객체를 생성합니다.
llm = get_chat(provider=provider, temperature=temperature)
def researcher(state: TeamState) -> dict[str, str]:
"""원천 데이터를 해석하지 않고 사실 중심으로 정리합니다."""
# 시스템 메시지로 Researcher가 분석이나 추측을 하지 않도록 역할을 제한합니다.
system = SystemMessage(
"너는 리서처다. 주어진 데이터를 해석 없이 사실만 항목별로 정리하라. "
"회사명, 주력 카테고리, 평점, 강점, 약점을 빠뜨리지 마라."
)
# 최초 상태의 raw에 들어 있는 경쟁사 원천 텍스트를 사용자 메시지로 전달합니다.
human = HumanMessage(f"[경쟁사 원천 데이터]\n{state['raw']}")
# LLM을 한 번 호출하고 반환 메시지의 본문을 문자열로 꺼냅니다.
output = llm.invoke([system, human]).content
# LangGraph가 기존 상태에 병합할 raw 칸만 반환합니다.
return {"raw": str(output)}
def analyst(state: TeamState) -> dict[str, str]:
"""Researcher가 정리한 사실만 사용해 경쟁사별 강점·약점·기회를 분석합니다."""
# 시스템 메시지로 분석 범위와 반드시 포함할 분석 항목을 지정합니다.
system = SystemMessage(
"너는 시장 분석가다. 정리된 사실만 근거로 경쟁사별 강점과 약점을 비교하고, "
"승승장구몰의 시장 기회와 대응 전략을 구체적으로 분석하라."
)
# 앞 단계가 갱신한 raw 값만 Analyst의 입력으로 전달합니다.
human = HumanMessage(f"[Researcher가 정리한 사실]\n{state['raw']}")
# 분석 모델을 호출해 메시지 본문을 가져옵니다.
output = llm.invoke([system, human]).content
# 공유 상태에서 analysis 칸만 갱신하도록 부분 딕셔너리를 반환합니다.
return {"analysis": str(output)}
def writer(state: TeamState) -> dict[str, str]:
"""Analyst의 분석만 사용해 임원 보고용 마크다운 리포트를 작성합니다."""
# 시스템 메시지로 보고서 역할과 문서 구성을 고정합니다.
system = SystemMessage(
"너는 임원 보고서 작성자다. 분석 결과를 근거로 마크다운 보고서를 작성하라. "
"제목, 핵심 요약, 경쟁사별 비교, 기회, 실행 제언 순서를 지켜라."
)
# 앞 단계가 만든 analysis 값만 Writer 입력으로 전달해 역할 간 결합을 낮춥니다.
human = HumanMessage(f"[Analyst 분석 결과]\n{state['analysis']}")
# Writer 역할의 LLM 호출 결과에서 텍스트 본문을 꺼냅니다.
output = llm.invoke([system, human]).content
# 최종 산출물인 report 칸만 갱신하도록 반환합니다.
return {"report": str(output)}
# 세 노드 함수를 그래프 조립 코드가 사용할 수 있도록 튜플로 반환합니다.
return researcher, analyst, writer
def build_team(provider: str):
"""START → Researcher → Analyst → Writer → END 순서의 StateGraph를 만듭니다."""
# 선택한 모델 공급자를 사용하는 세 노드를 생성합니다.
researcher, analyst, writer = create_nodes(provider)
# TeamState의 구조를 공유 상태 계약으로 사용하는 그래프 빌더를 생성합니다.
graph = StateGraph(TeamState)
# 각 Python 함수를 식별 가능한 노드 이름과 함께 그래프에 등록합니다.
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
# 그래프 시작점에서 Researcher로 이동하도록 첫 엣지를 선언합니다.
graph.add_edge(START, "researcher")
# Researcher 완료 후 Analyst가 실행되도록 연결합니다.
graph.add_edge("researcher", "analyst")
# Analyst 완료 후 Writer가 실행되도록 연결합니다.
graph.add_edge("analyst", "writer")
# Writer가 보고서를 만들면 그래프를 종료하도록 연결합니다.
graph.add_edge("writer", END)
# 선언한 그래프를 실제 invoke/stream이 가능한 실행 객체로 컴파일합니다.
return graph.compile()
def run_basic_team(provider: str, show_progress: bool = True) -> TeamState:
"""기본 AI 팀을 한 번 실행하고 누적된 최종 상태를 반환합니다."""
# provider에 맞는 기본 파이프라인을 컴파일합니다.
team = build_team(provider)
# raw에는 원천 텍스트를 넣고 아직 생성되지 않은 나머지 칸은 빈 문자열로 둡니다.
initial: TeamState = {"raw": load_competitor_text(), "analysis": "", "report": ""}
# 그래프가 아직 실행되지 않았을 때도 반환할 상태가 있도록 초기 상태를 복사합니다.
final_state: TeamState = initial.copy()
# values 모드는 각 노드 후의 누적 상태를 반환하므로 진행과 최종 결과를 한 번에 얻습니다.
for current_state in team.stream(initial, stream_mode="values"):
# 마지막 반복의 누적 상태가 전체 파이프라인의 최종 상태가 됩니다.
final_state = current_state
# 메뉴 실행일 때만 어떤 단계가 완료됐는지 화면에 표시합니다.
if show_progress:
completed = []
# 분석 결과가 존재하면 Analyst 단계 완료로 표시합니다.
if current_state.get("analysis"):
completed.append("analysis")
# 보고서가 존재하면 Writer 단계 완료로 표시합니다.
if current_state.get("report"):
completed.append("report")
# 완료 목록이 비어 있으면 Researcher가 동작 중인 시작 단계로 표시합니다.
print(f" [진행] {', '.join(completed) if completed else 'researcher'}")
# API를 다시 호출하지 않고 stream에서 얻은 마지막 상태를 반환합니다.
return final_state
conditional_review
ai_team을 가져온다.
실행한 횟수와 check 판정을 저장할 변수를 정의한다. 200자보다 짧고 강점, 약점, 기회중 표현이 없으면 부실 거짓이면 충분이라고 판정해 return.
verdict가 부실이고 max_retries도 적었을때 redo를 return.
노드 세개를 생성해 researcher를 호출해 현재 raw를 정돈한다.
현재 횟수에 1을 더한다.
그래프 생성. compile()
조건부 그래프 생성 필드 초기화, 각 노드 후 누적상태 받아 return.
# -*- coding: utf-8 -*-
"""분석 품질을 검사하고 부족하면 재조사·재분석하는 조건부 StateGraph 실습입니다."""
# TypedDict 확장을 위해 기존 TeamState를 상속합니다.
from code_app.ai_team import TeamState as BaseTeamState
# 기본 데이터 로더와 선택형 노드 생성 함수를 재사용합니다.
from code_app.ai_team import create_nodes, load_competitor_text
# StateGraph와 시작·종료 상수를 사용해 분기와 루프를 선언합니다.
from langgraph.graph import END, START, StateGraph
# 분석 재시도는 최대 두 번까지만 허용해 무한루프를 막습니다.
MAX_RETRIES = 2
class ConditionalTeamState(BaseTeamState):
"""기본 상태에 품질 판정과 재시도 횟수를 추가한 상태입니다."""
# retries는 재조사 노드를 실행한 횟수입니다.
retries: int
# verdict는 check 노드가 만든 '충분' 또는 '부실' 판정입니다.
verdict: str
def check(state: ConditionalTeamState) -> dict[str, str]:
"""분석 길이와 핵심 키워드를 기준으로 품질을 빠르게 검사합니다."""
# 상태에 analysis가 없을 가능성까지 고려해 기본값을 빈 문자열로 가져옵니다.
analysis = state.get("analysis", "")
# 200자보다 짧으면 충분한 비교 분석이 아닐 가능성이 높다고 판단합니다.
too_short = len(analysis) < 200
# 강점·약점·기회 중 어느 표현도 없으면 핵심 분석 요소가 빠진 것으로 봅니다.
missing_core = not any(keyword in analysis for keyword in ("강점", "약점", "기회"))
# 두 조건 중 하나라도 참이면 부실, 모두 거짓이면 충분으로 판정합니다.
verdict = "부실" if too_short or missing_core else "충분"
# 분기 판단 근거를 콘솔에서 확인할 수 있도록 출력합니다.
print(f" [검수] 길이={len(analysis)}자 / 핵심키워드={'없음' if missing_core else '있음'} / {verdict}")
# verdict 칸만 갱신하도록 부분 상태를 반환합니다.
return {"verdict": verdict}
def route_after_check(state: ConditionalTeamState) -> str:
"""품질과 재시도 횟수에 따라 redo 또는 ok 경로 이름을 반환합니다."""
# 분석이 부실하고 아직 최대 횟수에 도달하지 않았을 때만 재조사합니다.
if state["verdict"] == "부실" and state["retries"] < MAX_RETRIES:
return "redo"
# 충분하거나 재시도 한도에 도달했으면 Writer로 진행합니다.
return "ok"
def build_conditional_team(provider: str):
"""check 결과에 따라 redo 루프 또는 writer로 분기하는 그래프를 만듭니다."""
# 선택한 provider용 기본 노드 세 개를 생성합니다.
researcher, analyst, writer = create_nodes(provider)
def redo(state: ConditionalTeamState) -> dict[str, object]:
"""기존 사실을 더 구체적으로 다시 정리하고 재시도 횟수를 증가시킵니다."""
# 기본 Researcher를 다시 호출해 현재 raw를 더 정돈합니다.
updated_raw = researcher(state)["raw"]
# 현재 재시도 횟수에 1을 더해 다음 검사에서 상한을 판단하게 합니다.
next_retry = state["retries"] + 1
# 현재 몇 번째 재조사인지 사용자에게 알립니다.
print(f" [재조사] {next_retry}/{MAX_RETRIES}회")
# raw와 retries 두 칸만 동시에 갱신합니다.
return {"raw": updated_raw, "retries": next_retry}
# 확장 상태를 사용하는 그래프를 생성합니다.
graph = StateGraph(ConditionalTeamState)
# 기본 세 노드와 품질 검사·재조사 노드를 등록합니다.
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("check", check)
graph.add_node("redo", redo)
graph.add_node("writer", writer)
# 시작부터 분석 검사 전까지의 직선 흐름을 선언합니다.
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "check")
# 라우터 반환값 redo/ok를 실제 redo/writer 노드에 매핑합니다.
graph.add_conditional_edges("check", route_after_check, {"redo": "redo", "ok": "writer"})
# 재조사가 끝나면 Analyst로 되돌아가는 루프 엣지를 선언합니다.
graph.add_edge("redo", "analyst")
# Writer 완료 후 그래프를 종료합니다.
graph.add_edge("writer", END)
# 실행 가능한 그래프로 컴파일해 반환합니다.
return graph.compile()
def run_conditional_team(provider: str) -> ConditionalTeamState:
"""조건부 품질 재검토 그래프를 실행하고 최종 상태를 반환합니다."""
# provider별 조건부 그래프를 생성합니다.
team = build_conditional_team(provider)
# 확장 상태의 모든 필드를 초기화합니다.
initial: ConditionalTeamState = {
"raw": load_competitor_text(), "analysis": "", "report": "", "retries": 0, "verdict": ""
}
# stream 마지막 값을 보관할 변수를 초기화합니다.
final_state = initial.copy()
# 그래프를 한 번만 실행하면서 각 노드 후의 누적 상태를 받습니다.
for current_state in team.stream(initial, stream_mode="values"):
final_state = current_state
# 재실행 없이 최종 누적 상태를 반환합니다.
return final_state
artiface_save
import datetime, path, ai_team, common_bridge
langgraph.graph 가져오기
repoets 파일경로 생성파일세팅 datetime을 활용해 이름겹치지않도록 함. write_text() 하되 encoding을 지정하고 print로 안내한뒤 return path.
노드 함수 생성 researcher()를 실행해 상태를 받고, 정리된 사실을 txt 형식으로 저장.
analyst일때도. writer일때도.
그래프 생성.
graph.compile()
정의한 그래프생성을 해 teamstate를 초기화하고 final_state도 초기화해
team.stream() 최종상태 반환.
# -*- coding: utf-8 -*-
"""각 노드의 중간 산출물을 reports 폴더에 저장하는 래퍼 노드 실습입니다."""
# datetime은 실행마다 고유한 타임스탬프 파일명을 만들 때 사용합니다.
import datetime
# pathlib.Path는 텍스트 파일 저장 경로를 다룹니다.
from pathlib import Path
# 기본 상태와 노드 생성 함수, 데이터 로더를 재사용합니다.
from code_app.ai_team import TeamState, create_nodes, load_competitor_text
# 프로젝트 루트는 reports 폴더 위치를 정할 때 사용합니다.
from code_app.common_bridge import PROJECT_ROOT
# StateGraph로 저장 래퍼 노드의 실행 순서를 선언합니다.
from langgraph.graph import END, START, StateGraph
# 모든 실행 산출물을 프로젝트의 reports 폴더에 모읍니다.
REPORTS_DIR = PROJECT_ROOT / "reports"
def save_artifact(name: str, content: str, extension: str = "md") -> Path:
"""한 산출물을 UTF-8 파일로 저장하고 생성된 경로를 반환합니다."""
# 폴더가 없으면 상위 폴더까지 만들고 이미 존재하면 오류 없이 넘어갑니다.
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
# 마이크로초까지 포함해 같은 초에 여러 파일을 만들어도 이름이 겹치지 않게 합니다.
stamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S_%f")
# 단계명과 확장자를 조합해 실제 저장 경로를 만듭니다.
output_path = REPORTS_DIR / f"{stamp}_{name}.{extension}"
# 한글이 깨지지 않도록 UTF-8 인코딩을 지정해 파일을 기록합니다.
output_path.write_text(content, encoding="utf-8")
# 사용자가 저장 결과를 바로 확인할 수 있도록 경로를 출력합니다.
print(f" [저장] {name} → {output_path}")
# 테스트나 후속 처리에서 사용할 수 있도록 경로를 반환합니다.
return output_path
def build_saving_team(provider: str):
"""기본 세 노드를 감싸 실행 직후 각 산출물을 파일로 저장합니다."""
# 선택한 provider를 사용하는 원래 노드 함수를 생성합니다.
researcher, analyst, writer = create_nodes(provider)
def researcher_save(state: TeamState) -> dict[str, str]:
"""Researcher 결과를 텍스트 파일로 저장합니다."""
# 원래 노드를 실행해 부분 상태를 받습니다.
update = researcher(state)
# 정리된 사실을 txt 형식으로 저장합니다.
save_artifact("01_research", update["raw"], "txt")
# LangGraph 병합을 위해 원래 부분 상태를 그대로 반환합니다.
return update
def analyst_save(state: TeamState) -> dict[str, str]:
"""Analyst 결과를 마크다운 파일로 저장합니다."""
# 원래 Analyst 노드를 실행합니다.
update = analyst(state)
# 분석 결과를 md 파일로 저장합니다.
save_artifact("02_analysis", update["analysis"])
# 상태 병합을 위해 분석 업데이트를 반환합니다.
return update
def writer_save(state: TeamState) -> dict[str, str]:
"""Writer 최종 보고서를 마크다운 파일로 저장합니다."""
# 원래 Writer 노드를 실행합니다.
update = writer(state)
# 최종 보고서를 md 파일로 저장합니다.
save_artifact("03_report", update["report"])
# 최종 report 업데이트를 반환합니다.
return update
# 기본 상태를 사용하는 그래프를 생성합니다.
graph = StateGraph(TeamState)
# 원본 노드 대신 저장 기능이 추가된 래퍼 노드를 등록합니다.
graph.add_node("researcher", researcher_save)
graph.add_node("analyst", analyst_save)
graph.add_node("writer", writer_save)
# 기본 파이프라인과 같은 순서로 노드를 연결합니다.
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", END)
# 저장 기능이 포함된 실행 그래프를 반환합니다.
return graph.compile()
def run_saving_team(provider: str) -> TeamState:
"""중간 산출물 저장 그래프를 한 번 실행합니다."""
# 선택한 provider의 저장 그래프를 생성합니다.
team = build_saving_team(provider)
# 기본 공유 상태를 초기화합니다.
initial: TeamState = {"raw": load_competitor_text(), "analysis": "", "report": ""}
# 최종 상태 보관 변수를 초기화합니다.
final_state = initial.copy()
# API 호출을 중복하지 않도록 stream을 한 번만 순회합니다.
for current_state in team.stream(initial, stream_mode="values"):
final_state = current_state
# 화면 출력과 테스트에 사용할 최종 상태를 반환합니다.
return final_state
exercist_reviewer
import ai_team, common_bridge,
langchain_coremessage, langgrap.graph 가져오기
review와 final 변수지정
노드 그대로 사용
get_chat() 모델생성
reviewer() 생성 프롬프트를 적용하고 humanmessage를 writer가 만든것을 부여해 llm.invoke()
finalize 생성. 마찬가지로 편집자.라는 프롬프트를 적용하고 human에는 검토의견을 부여해 llm.invoke()
그래프 생성
앞서 작성한 확장 그래프를 생성하고
확장 상태에 필요한 칸 모두 초기화.
final_state 마지막상태를 보관하며
invoke대신 stream을 실행
# -*- coding: utf-8 -*-
"""실습문제 1 해답: Writer 뒤에 Reviewer와 Finalize 노드를 추가합니다."""
# 기본 상태를 확장하고 기본 노드 팩토리와 데이터 로더를 재사용합니다.
from code_app.ai_team import TeamState as BaseTeamState
from code_app.ai_team import create_nodes, load_competitor_text
# 제공된 common.py로 선택한 모델 공급자의 LLM을 생성합니다.
from code_app.common_bridge import get_chat
# LLM 메시지의 역할과 입력을 구분합니다.
from langchain_core.messages import HumanMessage, SystemMessage
# 다섯 노드의 순서를 StateGraph로 선언합니다.
from langgraph.graph import END, START, StateGraph
class ReviewTeamState(BaseTeamState):
"""기본 상태에 검토 의견과 최종 수정본 칸을 추가합니다."""
# review는 Reviewer가 작성한 사실성·가독성·구성 개선 의견입니다.
review: str
# final은 검토 의견을 반영해 Finalize 노드가 작성한 최종본입니다.
final: str
def build_review_team(provider: str):
"""Researcher → Analyst → Writer → Reviewer → Finalize 그래프를 구성합니다."""
# 기본 세 역할은 베이스 코드의 provider 선택형 노드를 그대로 사용합니다.
researcher, analyst, writer = create_nodes(provider)
# 검토와 최종 수정은 일관성을 높이기 위해 낮은 temperature로 별도 모델을 만듭니다.
llm = get_chat(provider=provider, temperature=0.2)
def reviewer(state: ReviewTeamState) -> dict[str, str]:
"""초안의 사실성·가독성·구성을 검토하고 개선점만 작성합니다."""
# Reviewer가 보고서를 직접 다시 쓰지 않도록 역할 경계를 명확히 지정합니다.
system = SystemMessage(
"너는 보고서 검토자다. 리포트의 사실성, 가독성, 구성을 평가하고 "
"개선할 점을 3가지 이내로 구체적으로 지적하라. 보고서를 다시 쓰지는 마라."
)
# Writer가 만든 report만 검토 대상으로 전달합니다.
human = HumanMessage(f"[검토할 리포트]\n{state['report']}")
# 검토 모델을 실행해 텍스트 의견을 가져옵니다.
output = llm.invoke([system, human]).content
# 사용자에게 Reviewer 노드 완료를 알립니다.
print(" [reviewer] 검토 의견 작성 완료")
# review 칸만 갱신하도록 반환합니다.
return {"review": str(output)}
def finalize(state: ReviewTeamState) -> dict[str, str]:
"""기존 리포트에 검토 의견을 반영해 최종 수정본을 작성합니다."""
# Finalize가 원문과 검토 의견을 모두 사용하도록 시스템 역할을 설정합니다.
system = SystemMessage(
"너는 최종 보고서 편집자다. 기존 리포트의 사실을 유지하면서 검토 의견을 반영해 "
"더 명확하고 실행 가능한 최종 마크다운 보고서를 작성하라."
)
# 초안과 검토 의견을 구분된 섹션으로 함께 전달합니다.
human = HumanMessage(f"[기존 리포트]\n{state['report']}\n\n[검토 의견]\n{state['review']}")
# 최종 수정 모델을 호출합니다.
output = llm.invoke([system, human]).content
# 사용자에게 Finalize 노드 완료를 알립니다.
print(" [finalize] 최종본 작성 완료")
# final 칸만 갱신하도록 반환합니다.
return {"final": str(output)}
# 확장된 ReviewTeamState를 사용하는 그래프를 생성합니다.
graph = StateGraph(ReviewTeamState)
# 기본 세 노드와 새로 추가한 두 노드를 등록합니다.
graph.add_node("researcher", researcher)
graph.add_node("analyst", analyst)
graph.add_node("writer", writer)
graph.add_node("reviewer", reviewer)
graph.add_node("finalize", finalize)
# 사람 조직의 초안→검토→반영 과정과 같은 순서로 엣지를 연결합니다.
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", "reviewer")
graph.add_edge("reviewer", "finalize")
graph.add_edge("finalize", END)
# 실행 가능한 확장 그래프로 컴파일합니다.
return graph.compile()
def run_review_exercise(provider: str) -> ReviewTeamState:
"""실습문제 1 완성 그래프를 실행하고 최종 상태를 반환합니다."""
# provider별 Reviewer 확장 그래프를 생성합니다.
team = build_review_team(provider)
# 확장 상태에 필요한 다섯 칸을 모두 초기화합니다.
initial: ReviewTeamState = {
"raw": load_competitor_text(), "analysis": "", "report": "", "review": "", "final": ""
}
# 마지막 누적 상태를 보관합니다.
final_state = initial.copy()
# stream을 한 번만 실행해 중복 비용을 방지합니다.
for current_state in team.stream(initial, stream_mode="values"):
final_state = current_state
# 검토 의견과 최종본이 포함된 최종 상태를 반환합니다.
return final_state
exercise_save_all
output 폴더 생성하기
researcher, analyst, writer, reviewer inivoke()해 final_save() 종합해 한번더 incoke. graph 생성, compile() 리턴
이를 .steam()로 실행하고 return.
# -*- coding: utf-8 -*-
"""실습문제 2 해답: Research·Analysis·Report·Final 네 산출물을 output 폴더에 저장합니다."""
# 프로젝트 루트는 output 폴더의 절대 경로를 정할 때 사용합니다.
from code_app.common_bridge import PROJECT_ROOT
# 실습문제 1의 상태와 그래프 구성을 재사용할 수 있는 구성 요소를 가져옵니다.
from code_app.ai_team import create_nodes, load_competitor_text
from code_app.exercise_reviewer import ReviewTeamState
# Reviewer와 Finalize를 이 파일 안에서 provider별로 생성하기 위해 메시지 클래스를 사용합니다.
from code_app.common_bridge import get_chat
from langchain_core.messages import HumanMessage, SystemMessage
# 모든 저장 래퍼 노드를 StateGraph에 연결합니다.
from langgraph.graph import END, START, StateGraph
# 실습문제 요구사항대로 고정 파일 네 개를 output 폴더에 생성합니다.
OUTPUT_DIR = PROJECT_ROOT / "output"
def save(name: str, content: str) -> None:
"""output 폴더 아래에 지정한 이름으로 UTF-8 텍스트를 저장합니다."""
# output 폴더가 없으면 자동으로 생성합니다.
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
# 프로젝트 루트와 파일명을 결합해 저장 경로를 만듭니다.
output_path = OUTPUT_DIR / name
# 한글과 마크다운이 깨지지 않도록 UTF-8로 기록합니다.
output_path.write_text(content, encoding="utf-8")
# 파일명과 글자 수를 표시해 저장 성공을 확인하게 합니다.
print(f" [저장] {name} ({len(content)}자)")
def build_save_all_team(provider: str):
"""실습문제 1의 전체 흐름에 네 단계 파일 저장을 결합합니다."""
# 선택한 provider로 Researcher·Analyst·Writer를 생성합니다.
researcher, analyst, writer = create_nodes(provider)
# Reviewer와 Finalize도 같은 provider를 사용하도록 모델을 생성합니다.
llm = get_chat(provider=provider, temperature=0.2)
def researcher_save(state: ReviewTeamState) -> dict[str, str]:
"""정리된 사실을 research.md로 저장합니다."""
update = researcher(state)
save("research.md", update["raw"])
return update
def analyst_save(state: ReviewTeamState) -> dict[str, str]:
"""분석 결과를 analysis.md로 저장합니다."""
update = analyst(state)
save("analysis.md", update["analysis"])
return update
def writer_save(state: ReviewTeamState) -> dict[str, str]:
"""Writer 초안을 report.md로 저장합니다."""
update = writer(state)
save("report.md", update["report"])
return update
def reviewer(state: ReviewTeamState) -> dict[str, str]:
"""Writer 초안의 개선 의견을 review 칸에 저장합니다."""
output = llm.invoke([
SystemMessage("너는 보고서 검토자다. 사실성·가독성·구성 개선점을 3가지 이내로 제시하라."),
HumanMessage(f"[검토할 리포트]\n{state['report']}"),
]).content
return {"review": str(output)}
def finalize_save(state: ReviewTeamState) -> dict[str, str]:
"""검토 의견을 반영한 최종본을 27_final.md로 저장합니다."""
output = llm.invoke([
SystemMessage("너는 최종 편집자다. 리포트에 검토 의견을 반영해 최종 마크다운 문서를 작성하라."),
HumanMessage(f"[리포트]\n{state['report']}\n\n[검토 의견]\n{state['review']}"),
]).content
update = {"final": str(output)}
save("27_final.md", update["final"])
return update
# Reviewer 확장 상태를 사용하는 그래프를 생성합니다.
graph = StateGraph(ReviewTeamState)
# 각 단계에서 저장이 필요한 노드는 래퍼 함수로 등록합니다.
graph.add_node("researcher", researcher_save)
graph.add_node("analyst", analyst_save)
graph.add_node("writer", writer_save)
graph.add_node("reviewer", reviewer)
graph.add_node("finalize", finalize_save)
# 실습문제 1과 동일한 직렬 흐름을 선언합니다.
graph.add_edge(START, "researcher")
graph.add_edge("researcher", "analyst")
graph.add_edge("analyst", "writer")
graph.add_edge("writer", "reviewer")
graph.add_edge("reviewer", "finalize")
graph.add_edge("finalize", END)
# 실행 가능한 저장 그래프로 컴파일합니다.
return graph.compile()
def run_save_all_exercise(provider: str) -> ReviewTeamState:
"""실습문제 2 그래프를 실행하고 네 파일과 최종 상태를 만듭니다."""
# 선택한 provider의 저장 그래프를 생성합니다.
team = build_save_all_team(provider)
# 모든 상태 칸을 초기화합니다.
initial: ReviewTeamState = {
"raw": load_competitor_text(), "analysis": "", "report": "", "review": "", "final": ""
}
# 마지막 상태를 담을 변수를 초기화합니다.
final_state = initial.copy()
# 그래프를 한 번만 실행해 모든 파일과 최종 결과를 생성합니다.
for current_state in team.stream(initial, stream_mode="values"):
final_state = current_state
# 생성된 output 폴더 위치를 알려줍니다.
print(f"\n완료 — {OUTPUT_DIR} 폴더를 확인하세요.")
# 메뉴에서 최종본을 출력할 수 있도록 상태를 반환합니다.
return final_state
console_meny.py
import ai_team, conditional_review, artiface_save, exercist_reviewer, exercise_save_all
메뉴출력,
각 초이스에 맞게 state 설정해 함수실행,
# -*- coding: utf-8 -*-
"""OpenAI와 Gemini를 동일한 실습 메뉴에서 선택·실행하는 콘솔 UI입니다."""
# 기본 파이프라인 실행 함수를 가져옵니다.
from code_app.ai_team import create_nodes, load_competitor_text, run_basic_team
# 조건부 분기, 중간 저장, 두 실습문제 실행 함수를 가져옵니다.
from code_app.conditional_review import run_conditional_team
from code_app.artifact_save import REPORTS_DIR, run_saving_team
from code_app.exercise_reviewer import run_review_exercise
from code_app.exercise_save_all import OUTPUT_DIR, run_save_all_exercise
def print_separator(title: str = "") -> None:
"""메뉴와 결과 영역을 구분하는 제목선을 출력합니다."""
# 고정 폭의 구분선을 출력합니다.
print("\n" + "=" * 72)
# 제목이 전달된 경우 구분선 아래에 제목을 표시합니다.
if title:
print(title)
print("-" * 72)
def select_provider() -> str | None:
"""상위 메뉴에서 OpenAI 또는 Gemini provider를 선택합니다."""
# 사용자가 볼 모델 공급자 선택 메뉴를 출력합니다.
print_separator("LangGraph AI Team — LLM 선택")
print("1. OpenAI 실행 메뉴")
print("2. Gemini 실행 메뉴")
print("0. 프로그램 종료")
# 앞뒤 공백을 제거한 사용자 입력을 받습니다.
choice = input("선택: ").strip()
# 1번은 common.py가 지원하는 openai 문자열로 변환합니다.
if choice == "1":
return "openai"
# 2번은 common.py가 지원하는 gemini 문자열로 변환합니다.
if choice == "2":
return "gemini"
# 0번은 종료를 의미하도록 None을 반환합니다.
if choice == "0":
return None
# 정의되지 않은 입력은 안내 후 상위 메뉴를 다시 보여줍니다.
print("[입력 오류] 0, 1, 2 중 하나를 입력하세요.")
return ""
def run_provider_menu(provider: str) -> None:
"""선택된 provider에서 실행할 제27강 실습 항목을 반복해서 보여줍니다."""
# 메뉴 제목에 표시할 공급자 이름을 사람이 읽기 좋은 형태로 변환합니다.
provider_label = "OpenAI" if provider == "openai" else "Gemini"
# 사용자가 이전 메뉴를 선택하기 전까지 하위 메뉴를 반복합니다.
while True:
print_separator(f"{provider_label} — StateGraph AI Team 실행 메뉴")
print("1. 경쟁사 CSV 데이터 로드 확인")
print("2. Researcher 단일 노드 실행")
print("3. Analyst 단일 노드 실행")
print("4. Writer 단일 노드 실행")
print("5. 기본 StateGraph 전체 실행")
print("6. 조건부 분기 품질 재검토 실행")
print("7. 중간 산출물 reports 저장 실행")
print("8. 실습문제 1 — Reviewer/Finalize 노드")
print("9. 실습문제 2 — 단계별 output 파일 저장")
print("0. LLM 선택 메뉴로 돌아가기")
# 실행할 번호를 문자열로 받습니다.
choice = input("선택: ").strip()
# 0번이면 현재 하위 메뉴 반복을 끝냅니다.
if choice == "0":
return
try:
# 데이터 로드 메뉴는 API를 호출하지 않고 제공 CSV만 확인합니다.
if choice == "1":
print_separator("경쟁사 데이터")
print(load_competitor_text())
# Researcher 단일 노드는 원천 데이터를 정리한 결과만 출력합니다.
elif choice == "2":
researcher, _, _ = create_nodes(provider)
state = {"raw": load_competitor_text(), "analysis": "", "report": ""}
print_separator("Researcher 결과")
print(researcher(state)["raw"])
# Analyst 단일 실행을 위해 Researcher 결과를 먼저 만든 후 분석합니다.
elif choice == "3":
researcher, analyst, _ = create_nodes(provider)
state = {"raw": load_competitor_text(), "analysis": "", "report": ""}
state.update(researcher(state))
print_separator("Analyst 결과")
print(analyst(state)["analysis"])
# Writer 단일 실행은 앞 단계 산출물이 필요하므로 Researcher와 Analyst를 순서대로 실행합니다.
elif choice == "4":
researcher, analyst, writer = create_nodes(provider)
state = {"raw": load_competitor_text(), "analysis": "", "report": ""}
state.update(researcher(state))
state.update(analyst(state))
print_separator("Writer 결과")
print(writer(state)["report"])
# 기본 StateGraph 전체 실행 결과의 report를 출력합니다.
elif choice == "5":
final = run_basic_team(provider)
print_separator("기본 AI Team 최종 리포트")
print(final["report"])
# 조건부 분기 결과와 재시도 횟수·최종 판정을 함께 출력합니다.
elif choice == "6":
final = run_conditional_team(provider)
print_separator(f"조건부 그래프 결과 — 판정={final['verdict']}, 재시도={final['retries']}")
print(final["report"])
# 저장 래퍼 그래프를 실행하고 실제 reports 경로를 표시합니다.
elif choice == "7":
final = run_saving_team(provider)
print_separator(f"저장 완료: {REPORTS_DIR}")
print(final["report"])
# 실습문제 1은 review와 final을 모두 확인하게 합니다.
elif choice == "8":
final = run_review_exercise(provider)
print_separator("Reviewer 검토 의견")
print(final["review"])
print_separator("검토 반영 최종본")
print(final["final"])
# 실습문제 2는 네 파일 저장 후 최종본과 output 경로를 표시합니다.
elif choice == "9":
final = run_save_all_exercise(provider)
print_separator(f"단계별 저장 완료: {OUTPUT_DIR}")
print(final["final"])
# 메뉴 범위 밖의 입력은 오류 메시지를 출력합니다.
else:
print("[입력 오류] 0부터 9 사이의 번호를 입력하세요.")
# API 키 누락, 네트워크, 할당량, 모델 오류 등 실행 예외를 메뉴 전체 종료 없이 처리합니다.
except Exception as error:
print_separator("실행 오류")
print(f"{type(error).__name__}: {error}")
def run_console() -> None:
"""프로그램 종료가 선택될 때까지 provider 선택 메뉴를 반복합니다."""
# 상위 메뉴를 계속 반복합니다.
while True:
# OpenAI/Gemini 중 사용할 provider를 선택받습니다.
provider = select_provider()
# None은 사용자가 0번 종료를 선택했다는 의미입니다.
if provider is None:
print("프로그램을 종료합니다.")
return
# 빈 문자열은 잘못된 입력이므로 하위 메뉴로 진입하지 않습니다.
if not provider:
continue
# 선택한 provider의 실습 메뉴를 실행합니다.
run_provider_menu(provider)
main()
import code_app.console_menu해 run_console()
# -*- coding: utf-8 -*-
"""PyCharm에서 직접 실행하는 제27강 AI Team 콘솔 프로젝트 시작 파일입니다."""
# code 폴더 안의 콘솔 메뉴 함수를 가져옵니다.
from code_app.console_menu import run_console
# 이 파일을 직접 실행했을 때만 콘솔 메뉴를 시작합니다.
if __name__ == "__main__":
# 사용자가 OpenAI 또는 Gemini를 선택하고 실습 메뉴를 실행하도록 합니다.
run_console()
지금까지의 학습 과정을 정리해보자.
1. LLM 하나만 사용하는 기본을 수행해봤었다.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4.1")
result = llm.invoke("파이썬이 뭐야?")
print(result.content)
2. 좀더 효과적으로 하기위해 프롬프트를 추가해봤었다.
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
("system", "당신은 파이썬 전문가입니다."),
("human", "{question}")
])
chain = prompt | llm
result = chain.invoke({
"question": "리스트가 뭐야?"
})
3. structured output으로 json으로 답하게했다 .
from pydantic import BaseModel
class Plan(BaseModel):
subtasks: list[str]
planner = llm.with_structured_output(Plan)
result = planner.invoke("쇼핑몰을 홍보하는 방법")
4. agent, 즉 LLM이 tool도 사용할 수 있도록 한다 . 검색이 필요하다고 판단이 들면 tool을 사용하는것 .
agent = create_react_agent(
llm,
tools=[search_tool]
)
agent.invoke({
"messages":[
HumanMessage("오늘 서울 날씨")
]
})
5. planner에게 리스트를 만들게 한뒤 for문으로 돌려봤었다.
Planner
│
▼
[
"시장조사",
"광고작성"
]for task in subtasks:
result = llm.invoke(task)
6. Synthesizer, 결과를 합쳤다.
outputs = [
("시장조사", "..."),
("광고", "..."),
]
answer = llm.invoke(outputs)
7. 위의 과정들을 자동으로 연결하도록 langgraph를 만들어 연결했다.
builder.add_node(
"planner",
planner_node
)
builder.add_node(
"worker",
worker_node
)
builder.add_node(
"synthesizer",
synthesizer_node
)builder.add_edge(
"planner",
"worker"
)
builder.add_edge(
"worker",
"synthesizer"
)
8. 필요에따라 rag 래그를 추가했다 .
llm.invoke(f"""
참고문서
{docs}
""")
9. 병렬로 worker를 부렸다.
from concurrent.futures import ThreadPoolExecutor
tasks = [
"시장 조사",
"경쟁사 분석",
"광고 작성"
]
def worker(task):
return llm.invoke(task).content
with ThreadPoolExecutor(max_workers=3) as executor:
results = list(
executor.map(worker, tasks)
)
print(results)from langgraph.types import Send
def planner(state):
return [
Send(
"worker",
{"task": task}
)
for task in state["tasks"]
]
20. reviewer를 추가했다.
def reviewer(state):
review = llm.invoke(f"""
다음 광고를 검토하세요.
{state["draft"]}
부족한 점만 말하세요.
""")
state["review"] = review.content
return statedef revise(state):
new_text = llm.invoke(f"""
초안
{state["draft"]}
검토
{state["review"]}
수정하세요.
""")
state["draft"] = new_text.content
return state
아래의 프로세스를 이해하자.
User
│
▼
Planner LLM
(할 일을 분해)
│
┌────────────┼────────────┐
▼ ▼ ▼
Worker1 Worker2 Worker3
(시장조사) (경쟁사) (광고)
└────────────┼────────────┘
▼
Reviewer LLM
(품질 검사)
┌───────┴────────┐
│ │
부족 충분
│ │
▼ ▼
Worker 재실행 Synthesizer
▼
최종 결과 생성'Personal > SK 네트웍스 AI 캠프' 카테고리의 다른 글
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day48_AI Agent 구현 실습 1 (0) | 2026.07.20 |
|---|---|
| [SK네트웍스 Family AI 캠프] 32기 12주차 회고: Day44 ~ Day47 (0) | 2026.07.16 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day46_LangGraph 기반 멀티 에이전트 구조와 응용 (1) | 2026.07.15 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day45_MCP(Model Context Protocol)의 이해와 서버 구축 (1) | 2026.07.15 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day44_Vector DB와 LLM을 결합한 RAG 아키텍처 (0) | 2026.07.13 |
