final project 대본
FastAPI 기반 ReAct + RAG + A2A + LangGraph + MCP 통합 AI Agent 프로젝트
프로젝트는 REST API 와 웹 기반 사용자 인터페이스를 제공하며, OpenAI 와 Google Gemini 를모두 지원하여 사용 환경에 따라 원하는 대형 언어 모델을 선택하여 사용할 수 있다.
시스템의 핵심은 LangChain 과 LangGraph 를 활용한 AI Agent 이며, 사용자의 질문을 분석하여적절한 처리 방식을 선택한다.
일반적인 질문은 LLM 이 직접 처리하고, 데이터 조회가 필요한 경우에는 다양한 Tool 을 호출하며, 정책과 같은 문서 기반 질문은 RAG(Retrieval-AugmentedGeneration)를 통해 근거를 검색한 후 답변을 생성한다.
또한 복합적인 질문은 ReAct(Reasoning + Acting) 패턴을 적용하여 질문 분석, 도구 선택, 실행결과 관찰, 추가 도구 호출 및 최종 답변 생성 과정을 반복 수행함으로써 보다 정확한 응답을제공한다.
프로젝트는 A2A(Agent-to-Agent) 구조를 적용하여 예로 주문 조회, 재고 조회, FAQ 검색, 정책 검색,교환 신청 등의 기능을 각각 독립적인 전문 에이전트로 구성한다.
각 에이전트는 자신의 역할과 기능을 Agent Card 형태로 제공하며, 필요한 경우 다른 에이전트와 협력하여 하나의 질문을 처리한다. 이러한 구조는 기능의 독립성과 확장성을 높여 새로운 업무 기능을 손쉽게 추가할 수 있도록 지원한다.
정책 검색 기능은 RAG 기반으로 구현되어 있으며, PDF 형태의 정책 문서를 자동으로 수집하고 문서를 적절한 크기로 분할한 후 임베딩 벡터를 생성하여 FAISS Vector DB 에 저장한다. 사용자가 정책과 관련된 질문을 입력하면 Retriever 가 가장 유사한 문서를 검색하고, 검색된근거(Context)를 LLM 에 전달하여 출처 기반의 신뢰성 높은 답변을 생성한다. 정책 문서에 존재하지 않는 내용은 임의로 생성하지 않고 확인할 수 없음을 안내하도록 구현하여 Hallucination 을 최소화하였다.
프로젝트는 MCP(Model Context Protocol)를 적용하여 예로 주문 조회, 재고 조회, FAQ 검색, 교환 신청 등의 기능을 표준화된 Tool 형태로 제공한다. MCP Server 는 외부 기능을 일관된 인터페이스로 노출하며, AI Agent 는 필요한 기능을 자동으로 발견하고 호출하여 결과를 활용할 수 있다. 이를 통해 시스템은 외부 서비스와의 연계성을 높이고 향후 새로운 도구를 손쉽게 추가할 수 있는 확장 가능한 구조를 제공한다.
사용자의 대화 이력은 Thread Memory(InMemorySaver)를 이용하여 스레드 단위로 관리된다.이를 통해 이전 대화 내용을 기억하고 멀티턴(Multi-turn) 대화를 자연스럽게 이어갈 수 있으며,동일한 사용자가 연속적으로 질문하는 경우에도 이전 문맥을 유지하여 일관성 있는 답변을 제공한다.FastAPI 는 프로젝트의 전체 API 서버 역할을 수행하며, REST API 와 Swagger(OpenAPI)를 함께 제공하여 개발자와 사용자가 손쉽게 기능을 테스트할 수 있도록 구성하였다. 웹 UI 에서는 OpenAI와 Gemini 모델 선택, 질문 입력, 응답 결과 확인, 처리 시간 확인, LangGraph 실행 경로 확인 등의 기능을 제공하며, API 를 이용하는 외부 시스템과도 쉽게 연동할 수 있다. 프로젝트에서 사용하는 데이터는 주문 정보, 재고 정보, FAQ 데이터, 정책 PDF 문서 등으로 구성되어 있으며, CSV 데이터는 Tool 을 통해 직접 조회하고 정책 문서는 RAG 를 이용하여 검색한다. 이러한 구조는 정형 데이터와 비정형 문서를 동시에 활용할 수 있는 하이브리드 AI시스템을 구현하는 데 목적이 있다.
전체 시스템은 FastAPI, LangChain, LangGraph, FAISS, OpenAI API, Gemini API, Pydantic, Uvicorn등의 최신 기술을 활용하여 구축되었으며, ReAct 기반 추론, RAG 기반 문서 검색, A2A 기반협업 에이전트, MCP 기반 Tool 호출을 하나의 통합 플랫폼으로 구현하였다.이를 통해 다양한 업무 환경에서 정확하고 신뢰성 높은 AI 서비스를 제공할 수 있으며, 향후 데이터베이스, Vector DB, 외부 API, 다양한 전문 에이전트를 추가하여 더욱 확장 가능한 AI플랫폼으로 발전시킬 수 있도록 설계하였다.
통합아키텍처 integrated architecture
사용자 요청, AI 추론, 문서검색, Tool 실행, 여러 Agent 협업을 하나의 흐름으로 연결하는 구조
사용자
│
FastAPI(API Server)
│
LangGraph (Workflow)
│
┌──────────────┐
│ │
LLM RAG
│ │
MCP VectorDB
│ │
Tool 문서검색
RAG
https://standout.tistory.com/1876
RAG(Retrieval-Augmented Generation)와 LLM의 Hallucination(환각): 외부 문서를 검색한 후 검색 결과를 바탕으
RAG(Retrieval-Augmented Generation)기업은 범용 언어모델을 그대로 사용하는 것이 아니라 사내 문서, 정보, 고객데이터 등을 추가해 기업전용 AI시스템을 구축하며 LLM은 이때 존재하지않는 정보를 생성
standout.tistory.com
Memory
AI가 이전 대화를 기억하는 기능
Short-term Memory, Long-term Memory(DB 저장), Conversation Buffer(대화전부저장), Summary Memory가 있다.
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import StateGraph
memory = InMemorySaver()
graph = builder.compile(
checkpointer=memory
)
config = {
"configurable": {
"thread_id": "user1"
}
}
graph.invoke(
{"messages": [("user", "내 이름은 홍길동")]},
config=config
)
graph.invoke(
{"messages": [("user", "내 이름이 뭐야?")]},
config=config
)import sqlite3
conn = sqlite3.connect("memory.db")
cur = conn.cursor()
cur.execute("""
CREATE TABLE IF NOT EXISTS memory(
user_id TEXT,
memory TEXT
)
""")
cur.execute(
"INSERT INTO memory VALUES (?, ?)",
("user1", "좋아하는 음식은 피자")
)
conn.commit()
cur.execute(
"SELECT memory FROM memory WHERE user_id=?",
("user1",)
)
print(cur.fetchone())from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()
memory.save_context(
{"input": "안녕"},
{"output": "안녕하세요"}
)
memory.save_context(
{"input": "내 이름은 홍길동"},
{"output": "기억하겠습니다"}
)
print(memory.load_memory_variables({}))from langchain.memory import ConversationSummaryMemory
from langchain_openai import ChatOpenAI
llm = ChatOpenAI()
memory = ConversationSummaryMemory(
llm=llm
)
memory.save_context(
{"input":"나는 AI를 공부한다."},
{"output":"좋습니다."}
)
print(memory.load_memory_variables({}))
Multi-Agent
여러 AI가 역할을 나눠 협업한다.
설계 Design
AI 설계는 Workflow, Prompt, Memory, Tool, Agent, Data Flow를 정의하는 것.
배포 Deployment
FastAPI - Docker - Nginx - Cloud - 사용자
배포시에 API SSI Logging Monitoring, Scaling 을 고려해야한다.
InMemorySaver
LangGraph에서 사용하는 Memory 저장방식
운영에서는 Redis, SQLite, Postgres 로 교체한다.
멀티턴 Multi-turn Scenario
여러번 이어지는 대화 이전대화를 기억하는 스무고개
FastAPI
Python 기반의 고성능 웹 API 프레임워크 REST API를 빠르게 구축 할 수 있고 비동기 Async 처리를 지원해 AI 서비스의 API 서버로 많이 사용된다. Swagger 문서를 자동생성하고 Pydantic과 연계해 데이터 검증도 제공한다 .
LangChain
LLM을 활용한 애플리케이션 개발 프레임워크
Prompt, Memory, RAG, Tool Calling, Agent등을 쉽게 구현할 수 있도록 다양한 컴포넌트를 제공한다.
LangGraph
LangChain 위에서 동작하는 Ai Workflow엔진 Node와 Edge를 이용해 복잡한 AI 흐름을 구현한다.
Loop, 조건분기, 재시도, 멀티에이전트등이 가능하다.
def search(state):
if state["retry"] < 3:
return "search"
return "answer"
builder.add_edge("search", "judge")
builder.add_conditional_edges(
"judge",
should_retry,
{
"retry": "search",
"finish": "answer"
}
)def router(state):
if state["category"] == "math":
return "math_agent"
return "search_agent"
builder.add_conditional_edges(
"router",
router
)from langgraph.types import RetryPolicy
builder.add_node(
"search",
search_tool,
retry_policy=RetryPolicy(
max_attempts=3
)
)builder.add_node("planner", planner)
builder.add_node("research", research)
builder.add_node("writer", writer)
builder.add_edge("planner", "research")
builder.add_edge("research", "writer")
FAISS Facebook AI Similarity Search
Vector DB중 하나 문서를 vector화해 저장한 뒤 질문과 가장 가까운 문서를 검색한다. RAG에서 많이 사용된다.
Pydantic
Python 데이터 검증 라이브러리
FasiAPI에서 거의 필수. 잘못된 데이터가 들어오면 자동으로 오류를 반환한다.
class User(BaseModel):
name:str
age:int
Uvicorn
ASGI 서버, FastAPI를 실제 실행시키는 서버 이를 거쳐 HTTP 요청이 처리된다.
ReAct 기반 추론 Reason + Act
Tool Calling의 핵심 기법으로 LLM이 생각 - 검색 - 생각 - 계산 - 답변 등의 생각 - 행동을 반복한다.
RAG 기반 문서 검색
질문이 들어오면 문서를 검색해 검색결과를 Prompt에 넣어 LLM이 답변한다.
A2A 기반 협업 에이전트 Agent - to - Agent
여러 전문 Agent가 역할을 분담하여 협업하는 구조.
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(
model="gpt-4o-mini"
)
def planner_agent(state):
response = llm.invoke(
f"""
사용자의 요청을 분석하고
작업 계획을 작성하세요.
요청:
{state['input']}
"""
)
return {
"plan": response.content
}
def researcher_agent(state):
response = llm.invoke(
f"""
아래 계획에 필요한 자료를 조사하세요.
계획:
{state['plan']}
"""
)
return {
"research": response.content
}
def writer_agent(state):
response = llm.invoke(
f"""
조사 결과를 기반으로
최종 답변을 작성하세요.
자료:
{state['research']}
"""
)
return {
"answer": response.content
}from langgraph.graph import StateGraph
builder = StateGraph(dict)
builder.add_node(
"planner",
planner_agent
)
builder.add_node(
"researcher",
researcher_agent
)
builder.add_node(
"writer",
writer_agent
)
builder.set_entry_point(
"planner"
)
builder.add_edge(
"planner",
"researcher"
)
builder.add_edge(
"researcher",
"writer"
)
graph = builder.compile()
MCP 기반 Tool 호출 Model Context Protocol
LLM이 외부 도구나 시스템과 표준화된 방식으로 통신하기 위한 프로토콜
MCP를 통해 데이터베이스 조회, 파일읽기, 일정 조회, 외부 API 호출 등 다양한 Tool을 일관된 인터페이스로 사용할 수 있고 도구가 추가되더라도 LLM 자체를 크게 수정하지않고 확장 할 수 있다.
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("Weather Server")
@mcp.tool()
def get_weather(city: str) -> str:
"""
특정 도시의 날씨 조회
"""
weather_data = {
"서울": "맑음 25도",
"부산": "비 22도"
}
return weather_data.get(
city,
"정보 없음"
)
if __name__ == "__main__":
mcp.run()
예시 final project final_system_fastapi_react_rag_a2a_langgraph_mcp를 분석해보자.
현 프로젝트는 FastAPI웹 UI에 REST API, Swagger에서 내부 LangGrapth StateGraph로 ReAct, RAG, A2A, MCP 로 나뉘어 구성된다.
데이터는 data.zip 실데이터와 정책 pdf를 사용했다.
FastAPI 엔드포인트은 아래와같다.
| 메서드 | URL | 기능 |
| ---- | ---------------------------- | --------------- |
| GET | `/` | 웹 채팅 UI |
| GET | `/docs` | Swagger |
| GET | `/redoc` | ReDoc |
| GET | `/api/v1/health` | 서버 및 API 키 상태 |
| POST | `/api/v1/chat` | 전체 통합 워크플로우 |
| POST | `/api/v1/tools/call` | 개별 MCP 호환 도구 실행 |
| GET | `/api/v1/a2a/agents` | A2A 에이전트 카드 |
| POST | `/api/v1/a2a/message` | 전문 에이전트 위임 |
| GET | `/api/v1/data/files` | data.zip 데이터 목록 |
| POST | `/api/v1/rag/reset` | FAISS 캐시 초기화 |
| POST | `/api/v1/exercises/exchange` | 교환신청 실습 해답 |
| GET | `/api/v1/exercises/memory` | 메모리 격리 실습 |
분석해보자.
settings.py
import functools lru_cache 설정 객체를 한번만 생성해 반복 로딩을 막기
pathlib, pydantic, pydantic_settings BaseSettings, SettingConfigDict .env 와 시스템 환경변수를 Pydantic모델로 읽는다.
typing Literal 허용가능한 공급자 문자열을 제한한다. provider: Literal["openai", "gemini"]
env 파일에서 읽어 할당할 변수세팅.
SettingsConfigDict() 환경변수를 읽되 알수없는 키는 exta=ignore 무시하도록 한다.
@lru_cache(maxsize = 1) 설정으로 객체를 한번만 생성해 반복되지않도록 한다.
# -*- coding: utf-8 -*-
"""애플리케이션 환경설정과 경로를 한 곳에서 관리하는 모듈입니다."""
# functools.lru_cache는 설정 객체를 한 번만 생성하여 반복 로딩을 막습니다.
from functools import lru_cache
# pathlib.Path는 운영체제에 독립적인 파일 경로를 안전하게 다룹니다.
from pathlib import Path
# typing.Literal은 허용 가능한 공급자 문자열을 제한합니다.
from typing import Literal
# pydantic의 Field는 환경변수 기본값과 설명을 정의합니다.
from pydantic import Field
# pydantic-settings는 .env와 시스템 환경변수를 Pydantic 모델로 읽습니다.
from pydantic_settings import BaseSettings, SettingsConfigDict
# 현재 파일의 상위 경로를 기준으로 프로젝트 루트 경로를 계산합니다.
PROJECT_ROOT = Path(__file__).resolve().parents[2]
# data.zip에서 추출한 데이터 파일이 위치하는 경로를 정의합니다.
DATA_DIR = PROJECT_ROOT / "data"
# 정책 PDF 문서가 위치하는 하위 폴더 경로를 정의합니다.
DOCS_DIR = DATA_DIR / "docs"
# 로그 파일을 저장할 폴더 경로를 정의합니다.
LOG_DIR = PROJECT_ROOT / "logs"
# FAISS 인덱스와 기타 캐시를 저장할 폴더 경로를 정의합니다.
CACHE_DIR = PROJECT_ROOT / "cache"
# Settings 클래스는 애플리케이션에서 사용할 모든 환경변수를 타입 안전하게 관리합니다.
class Settings(BaseSettings):
"""FastAPI, LLM, RAG, 재시도 설정을 보관하는 환경설정 모델입니다."""
# model_config는 .env 파일을 읽고 알 수 없는 키는 무시하도록 설정합니다.
model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8", extra="ignore")
# app_name은 Swagger와 웹 화면에 표시할 애플리케이션 이름입니다.
app_name: str = Field(default="승승장구몰 통합 CS Agent")
# app_version은 API 버전을 표시합니다.
app_version: str = Field(default="1.0.0")
# debug는 개발 중 상세 오류 출력 여부를 제어합니다.
debug: bool = Field(default=True)
# default_provider는 별도 지정이 없을 때 사용할 LLM 공급자입니다.
default_provider: Literal["openai", "gemini"] = Field(default="openai")
# openai_api_key는 OpenAI API 인증 키를 저장합니다.
openai_api_key: str = Field(default="", alias="OPENAI_API_KEY")
# openai_model은 OpenAI 채팅 모델 이름입니다.
openai_model: str = Field(default="gpt-4o-mini", alias="OPENAI_MODEL")
# openai_embed_model은 OpenAI 임베딩 모델 이름입니다.
openai_embed_model: str = Field(default="text-embedding-3-small", alias="OPENAI_EMBED_MODEL")
# google_api_key는 Gemini API 인증 키를 저장합니다.
google_api_key: str = Field(default="", alias="GOOGLE_API_KEY")
# gemini_model은 Gemini 채팅 모델 이름입니다.
gemini_model: str = Field(default="gemini-2.5-flash", alias="GEMINI_MODEL")
# gemini_embed_model은 Gemini 임베딩 모델 이름입니다.
gemini_embed_model: str = Field(default="models/gemini-embedding-001", alias="GEMINI_EMBED_MODEL")
# temperature는 답변의 무작위성을 제어합니다.
temperature: float = Field(default=0.1, ge=0.0, le=2.0)
# max_retries는 외부 LLM 호출 실패 시 최대 재시도 횟수입니다.
max_retries: int = Field(default=3, ge=1, le=10)
# retry_base_seconds는 지수 백오프의 최초 대기 시간입니다.
retry_base_seconds: float = Field(default=0.5, ge=0.0)
# rag_chunk_size는 PDF 문서를 자를 청크 크기입니다.
rag_chunk_size: int = Field(default=700, ge=100)
# rag_chunk_overlap은 인접 청크 사이의 중첩 길이입니다.
rag_chunk_overlap: int = Field(default=100, ge=0)
# rag_top_k는 질문과 가장 관련된 상위 문서 개수입니다.
rag_top_k: int = Field(default=4, ge=1, le=20)
# cors_origins는 개발용 브라우저 접근을 허용할 출처 목록입니다.
cors_origins: list[str] = Field(default=["*"])
# lru_cache는 Settings 객체를 프로세스에서 한 번만 생성합니다.
@lru_cache(maxsize=1)
def get_settings() -> Settings:
"""캐시된 환경설정 객체를 반환합니다."""
# Settings 생성 시 .env와 시스템 환경변수를 자동으로 읽습니다.
return Settings()
app.service.data_service.py
import random, string, pandas
threading Lock 여러 작업이 동시에 같은 데이터에 접근할때 한번에 하나의 작업만 접근하도록 막는 장치. lock = Lock() width.lock:
app.core.settings
Lock() 객체 만들기
file_name 매개변수를 받아 이미 로드되었다면 즉시반환, _data_lock 전제조건에서 기존 _tables에 없다면 pd.read.csv()해 return.
order_id를 받아 공백을 strip()제거하고 upper() 대문자로 통일한다. _load_table() orders.csv를 가져와 astype(str).str.upper() 주문번호와 같은 행을 추출, 없다면 안내. 첫번째 행을 기준으로 iloc[0] 가져오기, name을 가져오거나 없으면 '상품명 미상'반영.
수량을 읽어 int형으로 바꾸고 orderdate와 status 주문상태를 읽어 return.
get_stock()도 마찬가지.
search_faq()는 question_column에서와, answer에서의 후보를 찾아 질문과 답변을 합친다. 질문뿐 아니라 답변내용에도 키워드가 있을수 있기 때문. faq에서 combined에서 contains() 문자포함 여부를 검색하되 case 대소문자 무시, regex 정규식 사용안함(C++와 같은 것들이 영향을 받을 수 있음), na 빈값처리 . 없으면 안내, 있으면 일치하는 첫번째 행 hit.iloc[0] 첫번째 행 을 가져와 return.
request_exchange() 주문번호를 정규화해 실제존재하는지 확인. reason 공백을 제거해놓는다.
random uppder 대문자와 숫자 중 6개를 무작위로 생성해 접수번호를 완성하고 안내문구return.
rglob(*)전체 파이리을 for순회하며 있는 파일만 result에 append해 path와stat() 파일정보에서 st_size 파일 크기를 저장해 목록 result return.
# -*- coding: utf-8 -*-
"""data.zip에서 추출한 CSV 파일을 조회하고 교환 접수를 처리하는 서비스입니다."""
# random은 교환 접수번호에 사용할 임의 문자를 선택합니다.
import random
# string은 영문 대문자와 숫자 문자 집합을 제공합니다.
import string
# threading.Lock은 동시에 여러 요청이 들어올 때 CSV 로딩을 안전하게 보호합니다.
from threading import Lock
# pandas는 CSV 데이터를 DataFrame으로 읽고 검색합니다.
import pandas as pd
# DATA_DIR은 data.zip의 실제 데이터가 위치하는 경로입니다.
from app.core.settings import DATA_DIR
# _data_lock은 최초 데이터 로딩 시 경쟁 상태를 방지합니다.
_data_lock = Lock()
# _tables는 한 번 읽은 CSV를 메모리에 저장하는 모듈 캐시입니다.
_tables: dict[str, pd.DataFrame] = {}
# _load_table 함수는 지정한 CSV를 한 번만 읽고 재사용합니다.
def _load_table(file_name: str) -> pd.DataFrame:
"""CSV 파일을 지연 로딩하여 캐시된 DataFrame을 반환합니다."""
# 이미 로드된 데이터라면 파일을 다시 읽지 않고 즉시 반환합니다.
if file_name in _tables:
# 캐시된 DataFrame을 반환합니다.
return _tables[file_name]
# 여러 요청이 동시에 최초 로딩을 시도하지 않도록 잠금을 획득합니다.
with _data_lock:
# 잠금 대기 중 다른 요청이 로딩했을 수 있으므로 다시 확인합니다.
if file_name not in _tables:
# 프로젝트 data 폴더에서 요청한 CSV 파일을 읽습니다.
_tables[file_name] = pd.read_csv(DATA_DIR / file_name)
# 최종적으로 캐시된 DataFrame을 반환합니다.
return _tables[file_name]
# get_order_status는 주문번호로 실제 주문 데이터를 검색합니다.
def get_order_status(order_id: str) -> str:
"""orders.csv에서 주문 상태, 상품, 수량, 금액을 조회합니다."""
# 사용자가 입력한 주문번호의 공백을 제거하고 대문자로 통일합니다.
normalized_id = order_id.strip().upper()
# 주문 CSV를 캐시에서 가져옵니다.
orders = _load_table("orders.csv")
# order_id 열이 정규화된 주문번호와 같은 행만 추출합니다.
hit = orders[orders["order_id"].astype(str).str.upper() == normalized_id]
# 검색 결과가 없으면 환각 없이 명확한 실패 메시지를 반환합니다.
if hit.empty:
# 찾지 못한 주문번호를 함께 표시합니다.
return f"주문번호 {normalized_id}를 찾을 수 없습니다."
# 첫 번째 검색 결과를 Series로 가져옵니다.
row = hit.iloc[0]
# 필요한 열이 데이터에 있는지 확인하며 안전하게 값을 가져옵니다.
product_name = row.get("product_name", "상품명 미상")
# 수량을 읽고 정수 문자열로 변환합니다.
quantity = int(row.get("quantity", 0))
# 금액을 읽고 천 단위 쉼표 형식으로 변환합니다.
amount = int(float(row.get("amount", 0)))
# 주문일을 문자열로 읽습니다.
order_date = row.get("order_date", "주문일 미상")
# 주문 상태를 문자열로 읽습니다.
status = row.get("status", "상태 미상")
# 사용자에게 읽기 쉬운 한 문장으로 조합하여 반환합니다.
return (
f"주문 {normalized_id}: {product_name} {quantity}개, "
f"{amount:,}원, 주문일 {order_date}, 상태={status}"
)
# get_stock은 상품명 일부를 이용해 재고를 검색합니다.
def get_stock(product_name: str) -> str:
"""inventory.csv에서 상품명과 재고 수량을 조회합니다."""
# 검색 키워드의 앞뒤 공백을 제거합니다.
keyword = product_name.strip()
# 재고 CSV를 캐시에서 가져옵니다.
inventory = _load_table("inventory.csv")
# 데이터셋마다 상품명 열 이름이 다를 수 있으므로 후보를 순서대로 찾습니다.
name_column = next(
(column for column in ["product_name", "name", "product"] if column in inventory.columns),
None,
)
# 상품명 열을 찾지 못한 경우 데이터 구조 오류를 반환합니다.
if name_column is None:
# 개발자가 CSV 열을 확인할 수 있는 메시지를 반환합니다.
return "inventory.csv에서 상품명 열을 찾을 수 없습니다."
# 대소문자와 정규식 영향을 제거한 부분 문자열 검색을 수행합니다.
hit = inventory[inventory[name_column].astype(str).str.contains(keyword, case=False, regex=False, na=False)]
# 검색 결과가 없으면 임의 재고 수량을 생성하지 않습니다.
if hit.empty:
# 오타 또는 미등록 상품일 수 있음을 명확히 알립니다.
return f"'{keyword}' 상품의 재고 정보를 찾을 수 없습니다."
# 가장 먼저 일치한 상품 행을 선택합니다.
row = hit.iloc[0]
# 재고 수량 열 이름 후보를 순서대로 찾습니다.
stock_column = next(
(column for column in ["stock", "stock_quantity", "quantity", "inventory"] if column in inventory.columns),
None,
)
# 재고 열이 없으면 열 구조 오류를 반환합니다.
if stock_column is None:
# 개발자가 CSV 구조를 수정할 수 있도록 안내합니다.
return "inventory.csv에서 재고 수량 열을 찾을 수 없습니다."
# 실제 상품명을 읽습니다.
matched_name = row[name_column]
# 재고 수량을 정수로 변환합니다.
stock_value = int(float(row[stock_column]))
# 조회 결과를 사용자 친화적인 문장으로 반환합니다.
return f"상품 '{matched_name}'의 현재 재고는 {stock_value:,}개입니다."
# search_faq는 FAQ 질문과 답변에서 키워드를 검색합니다.
def search_faq(keyword: str) -> str:
"""faq.csv에서 키워드와 관련된 FAQ 답변을 검색합니다."""
# 검색어의 불필요한 공백을 제거합니다.
normalized_keyword = keyword.strip()
# FAQ CSV를 캐시에서 가져옵니다.
faq = _load_table("faq.csv")
# 질문 열 이름 후보를 찾습니다.
question_column = next(
(column for column in ["question", "faq_question", "title"] if column in faq.columns),
None,
)
# 답변 열 이름 후보를 찾습니다.
answer_column = next(
(column for column in ["answer", "faq_answer", "content"] if column in faq.columns),
None,
)
# 필수 열이 없으면 데이터 구조 오류를 반환합니다.
if question_column is None or answer_column is None:
# FAQ 스키마를 확인하라는 메시지를 반환합니다.
return "faq.csv에서 질문 또는 답변 열을 찾을 수 없습니다."
# 질문과 답변을 합친 문자열에서 키워드를 검색합니다.
combined = faq[question_column].astype(str) + " " + faq[answer_column].astype(str)
# 대소문자와 정규식 영향을 제거하여 관련 행을 필터링합니다.
hit = faq[combined.str.contains(normalized_keyword, case=False, regex=False, na=False)]
# 관련 FAQ가 없으면 솔직한 실패 메시지를 반환합니다.
if hit.empty:
# 검색어를 포함해 사용자가 다른 표현을 시도할 수 있게 합니다.
return f"'{normalized_keyword}'와 관련된 FAQ를 찾지 못했습니다."
# 가장 먼저 일치한 FAQ 행을 선택합니다.
row = hit.iloc[0]
# 질문과 답변을 함께 반환해 근거를 확인할 수 있게 합니다.
return f"FAQ 질문: {row[question_column]}\nFAQ 답변: {row[answer_column]}"
# request_exchange는 쓰기 행동 도구의 교육용 예시입니다.
def request_exchange(order_id: str, reason: str) -> str:
"""실제 주문 존재 여부를 확인한 뒤 교환 접수번호를 생성합니다."""
# 주문번호를 표준 형식으로 정규화합니다.
normalized_id = order_id.strip().upper()
# 주문이 실제 존재하는지 먼저 조회합니다.
order_result = get_order_status(normalized_id)
# 찾을 수 없다는 문구가 포함되면 잘못된 주문으로 판단합니다.
if "찾을 수 없습니다" in order_result:
# 존재하지 않는 주문에는 접수번호를 만들지 않습니다.
return order_result + " 따라서 교환을 접수하지 않았습니다."
# 교환 사유의 공백을 제거합니다.
normalized_reason = reason.strip()
# 사유가 비어 있으면 추가 입력을 요청합니다.
if not normalized_reason:
# 쓰기 작업 전에 필수 정보를 확인합니다.
return "교환 사유를 입력해 주세요."
# 영문 대문자와 숫자 중에서 6개 문자를 무작위로 선택합니다.
suffix = "".join(random.choices(string.ascii_uppercase + string.digits, k=6))
# EX 접두사를 붙여 교환 접수번호를 완성합니다.
ticket = f"EX-{suffix}"
# 교육용 접수 결과를 상세히 반환합니다.
return (
f"교환 접수가 완료되었습니다. 접수번호: {ticket}\n"
f"주문번호: {normalized_id}\n교환 사유: {normalized_reason}\n"
"영업일 기준 1~2일 내 담당자가 연락드립니다."
)
# list_data_files는 웹 화면에서 사용 데이터 목록을 보여 줍니다.
def list_data_files() -> list[dict[str, object]]:
"""data 폴더의 CSV, JSON, PDF 파일 정보를 반환합니다."""
# 결과 목록을 빈 리스트로 초기화합니다.
result: list[dict[str, object]] = []
# data 폴더 아래의 모든 파일을 재귀적으로 순회합니다.
for path in sorted(DATA_DIR.rglob("*")):
# 디렉터리는 제외하고 실제 파일만 처리합니다.
if path.is_file():
# 프로젝트 기준 상대 경로와 파일 크기를 저장합니다.
result.append({"name": str(path.relative_to(DATA_DIR)), "size": path.stat().st_size})
# 완성된 파일 정보 목록을 반환합니다.
return result
app.core.common.py
import os, path, dotenv
env 에서 가져올 변수세팅, 읽어 strip() 해 변수에 저장하며 비어있을경우 안내
genai 클라이언트만들기
langchain_google_genai 를 통해 ChatGoogleGenerativeAI Gemini 모델 생성하기
langchain_openai를 통해 ChatOPenAI모델 생성하기
langchain_google_genai를 통해 GoogleGenerativeAIEmbedding 객체 생성하기 차원은 768로 설정.
langchain_openai를 통해 OpenAIEmbeddings 객체 생성하기
파일을 직접 실행할 경우 기본 정보 표시하기
# -*- coding: utf-8 -*-
"""제공된 common.py를 FastAPI 앱 구조에 맞게 경로만 조정한 공통 모듈입니다."""
# os는 운영체제 환경변수에서 API 키와 모델명을 읽습니다.
import os
# pathlib.Path는 Windows와 macOS/Linux에서 동일한 방식으로 경로를 다룹니다.
from pathlib import Path
# load_dotenv는 프로젝트 루트의 .env 파일을 환경변수로 로드합니다.
from dotenv import load_dotenv
# 현재 파일은 app/core/common.py이므로 parents[2]가 프로젝트 루트입니다.
ROOT = Path(__file__).resolve().parents[2]
# DATA는 data.zip에서 추출한 실데이터 폴더 경로입니다.
DATA = ROOT / "data"
# DOCS는 RAG에서 사용할 정책 PDF 폴더 경로입니다.
DOCS = DATA / "docs"
# 프로젝트 루트의 .env 파일을 UTF-8 환경변수로 로드합니다.
load_dotenv(ROOT / ".env")
# GEMINI_MODEL은 환경변수 값이 없을 때 기본 Gemini 채팅 모델을 사용합니다.
GEMINI_MODEL = os.getenv("GEMINI_MODEL", "gemini-2.5-flash")
# GEMINI_EMBED_MODEL은 정책 PDF 벡터화에 사용할 Gemini 임베딩 모델입니다.
GEMINI_EMBED_MODEL = os.getenv("GEMINI_EMBED_MODEL", "models/gemini-embedding-001")
# OPENAI_MODEL은 환경변수 값이 없을 때 기본 OpenAI 채팅 모델을 사용합니다.
OPENAI_MODEL = os.getenv("OPENAI_MODEL", "gpt-4o-mini")
# OPENAI_EMBED_MODEL은 정책 PDF 벡터화에 사용할 OpenAI 임베딩 모델입니다.
OPENAI_EMBED_MODEL = os.getenv("OPENAI_EMBED_MODEL", "text-embedding-3-small")
# require_key 함수는 필수 API 키가 실제로 설정되어 있는지 검사합니다.
def require_key(name: str) -> str:
"""환경변수 키가 없거나 예제 문자열이면 명확한 설정 오류를 발생시킵니다."""
# 요청한 이름의 환경변수 값을 읽습니다.
value = os.getenv(name, "").strip()
# 값이 비어 있거나 예제 문구로 시작하면 유효하지 않은 키로 판단합니다.
if not value or value.startswith("여기에"):
# FastAPI에서 처리 가능한 ValueError로 필요한 조치를 안내합니다.
raise ValueError(
f"{name}가 .env에 설정되지 않았습니다. "
".env.example을 .env로 복사한 뒤 실제 API 키를 입력하세요."
)
# 검증된 키 문자열을 반환합니다.
return value
# get_genai_client 함수는 Google의 원시 SDK 클라이언트를 생성합니다.
def get_genai_client():
"""Google Gemini 원시 SDK 클라이언트를 반환합니다."""
# google.genai는 실제 사용 시에만 지연 import합니다.
from google import genai
# 검증된 Google API 키로 클라이언트를 생성합니다.
return genai.Client(api_key=require_key("GOOGLE_API_KEY"))
# get_chat 함수는 OpenAI 또는 Gemini LangChain 채팅 모델을 통일된 방식으로 생성합니다.
def get_chat(provider: str = "gemini", temperature: float = 0.0):
"""provider 값에 맞는 LangChain ChatModel을 반환합니다."""
# Gemini 공급자를 선택했는지 확인합니다.
if provider == "gemini":
# Google API 키를 먼저 검증합니다.
api_key = require_key("GOOGLE_API_KEY")
# Gemini 전용 LangChain 클래스를 지연 import합니다.
from langchain_google_genai import ChatGoogleGenerativeAI
# 환경변수 모델명과 temperature로 Gemini 모델을 생성합니다.
return ChatGoogleGenerativeAI(
google_api_key=api_key,
model=GEMINI_MODEL,
temperature=temperature,
max_retries=0,
)
# OpenAI 공급자를 선택했는지 확인합니다.
if provider == "openai":
# OpenAI API 키를 먼저 검증합니다.
api_key = require_key("OPENAI_API_KEY")
# OpenAI 전용 LangChain 클래스를 지연 import합니다.
from langchain_openai import ChatOpenAI
# 환경변수 모델명과 temperature로 OpenAI 모델을 생성합니다.
return ChatOpenAI(
api_key=api_key,
model=OPENAI_MODEL,
temperature=temperature,
max_retries=0,
)
# 지원하지 않는 공급자는 잘못된 입력으로 처리합니다.
raise ValueError(f"알 수 없는 provider입니다: {provider}")
# get_embeddings 함수는 선택한 공급자의 임베딩 모델을 통일된 방식으로 생성합니다.
def get_embeddings(provider: str = "gemini"):
"""provider 값에 맞는 LangChain Embeddings 객체를 반환합니다."""
# Gemini 임베딩을 선택했는지 확인합니다.
if provider == "gemini":
# Google API 키를 검증합니다.
api_key = require_key("GOOGLE_API_KEY")
# Gemini 임베딩 클래스를 지연 import합니다.
from langchain_google_genai import GoogleGenerativeAIEmbeddings
# 검색 속도와 저장 크기를 고려해 768차원 임베딩을 사용합니다.
return GoogleGenerativeAIEmbeddings(
google_api_key=api_key,
model=GEMINI_EMBED_MODEL,
output_dimensionality=768,
)
# OpenAI 임베딩을 선택했는지 확인합니다.
if provider == "openai":
# OpenAI API 키를 검증합니다.
api_key = require_key("OPENAI_API_KEY")
# OpenAI 임베딩 클래스를 지연 import합니다.
from langchain_openai import OpenAIEmbeddings
# 환경변수로 지정한 OpenAI 임베딩 모델을 생성합니다.
return OpenAIEmbeddings(api_key=api_key, model=OPENAI_EMBED_MODEL)
# 지원하지 않는 공급자는 오류로 처리합니다.
raise ValueError(f"알 수 없는 provider입니다: {provider}")
# 파일을 직접 실행하면 공통 경로와 API 키 로드 상태를 확인합니다.
if __name__ == "__main__":
# 프로젝트 루트 경로를 출력합니다.
print("ROOT:", ROOT)
# 실데이터 경로와 존재 여부를 출력합니다.
print("DATA:", DATA, "존재:", DATA.exists())
# 정책 문서 경로와 존재 여부를 출력합니다.
print("DOCS:", DOCS, "존재:", DOCS.exists())
# API 키 원문 대신 설정 여부만 출력합니다.
print(
"API 키 설정 상태:",
"GOOGLE_API_KEY=", bool(os.getenv("GOOGLE_API_KEY")),
"OPENAI_API_KEY=", bool(os.getenv("OPENAI_API_KEY")),
)
app.service.llm_factory.py
import typing Literal
app.core.common
app.core.settings
settings에서 정의한 get_settings을 가져와 get_chat() 객체 만들어 반환하기
get_embedding도 get_embeddings를 가져와 객체만드리어 반환하기
# -*- coding: utf-8 -*-
"""제공된 common.py를 재사용하여 채팅 및 임베딩 모델을 생성하는 팩토리입니다."""
# Literal은 공급자 값을 openai 또는 gemini로 제한합니다.
from typing import Literal
# 제공된 공통 모듈의 모델 생성 함수를 가져옵니다.
from app.core.common import get_chat, get_embeddings
# 환경설정에서 temperature 값을 가져옵니다.
from app.core.settings import get_settings
# create_chat_model 함수는 common.py의 get_chat을 FastAPI 서비스 계층에 연결합니다.
def create_chat_model(provider: Literal["openai", "gemini"]):
"""선택 공급자의 LangChain 채팅 모델을 반환합니다."""
# 공통 환경설정 객체를 읽습니다.
settings = get_settings()
# common.py에 공급자와 temperature를 전달하여 모델을 생성합니다.
return get_chat(provider=provider, temperature=settings.temperature)
# create_embeddings 함수는 common.py의 get_embeddings를 RAG 서비스에 연결합니다.
def create_embeddings(provider: Literal["openai", "gemini"]):
"""선택 공급자의 LangChain 임베딩 모델을 반환합니다."""
# common.py가 API 키 검증과 모델명 선택을 공통 처리합니다.
return get_embeddings(provider=provider)
app.service.rag_service .py
threading Lock, 공급자의 여러 요청이 동시에 만들지안히도록 보호한다. 한번에 하나의 thread만 이영역에 접근이 가능함.
typing Literal 공급자값 제한
langchain_core.documnets Document RAG에서 매우 중요한 클래스로 LangChain의 Document는 텍스트 데이터와 텍스트 부가정보를 함께 저장하는 표준 데이터 구조 문자열은 내용만 존재하나 RAG에서는 내용, 출처, 페이지, 파일명, 날짜, 카테고리 정보가 필요함으로 Document를 사용한다 .
Document(
page_content='환불은 마이페이지에서 신청할 수 있습니다.',
metadata={
source:'faq.csv',
category:'payment'
}
)
app.core.settings
app.service.llm_factory
Lock() 객체만들기
_build_vectorstore() 실행시 import langchain_community.document_loaders 해 PyPDFLoader pdf 로더가져오기
langchain_community.vectorstores FAISS vector저장소 FAISS 클래스 가져오기
langchain_text_splitters RecursiveCharacterTestAplitter LLM이나 Embedding 모델은 입력 가능한 Token 제한이 있다. 그대로 Embedding하면 Token이 초과되고 검색정확도가 저하되면 관련없는 애용까지 같이 전달될 수 있음으로 긴문서를 Chunk분할해 작은 Document여러개를 변환한다. RecursiveCharacterTextSplitter는 문자를 기준으로 재귀적 문서를 분할하는 도구로 문장을 최대한 유지하며 문단 구조를 고려해 작은 단위로 분할한다.
*.pdf 자료를 찾아 for문을 돌려 PyPDFLoader() .load() PyPDFLoader가 PDF 페이지를 LangChain Document 목록으로 변환해줌으로 load_pages변수에 할당. file.name을 metadata source_name에 사용한다.
load_pages를 documents.extend 전체 문서 목록에 추가.
RecursiveCharacterTextSplitter 객체를 만들어 분할기 생성.
splitter.split_documents() 한 뒤 chunk 변수에 할당하고,
create_embeddgins() 임베딩 객체를 만들어 FAISS.from_documents(lchunks, embeddings) 객체 반환
공급자를 받아 _vectores[] 인덱스를 반환, width _rag_lock을 걸고, 없을 경우 앞서 작성한 _build_vectorstore() 를 통해 생성해 반환. query와 공급자를매개변수로 받아 provider값으로 get_vectorstore() 실행해 맞는 FAISS객체를 가져온다. vectorstore.similarity_search()로 관련문서를 찾아 document에 할당하고 for문으로 각 결과를 순회해 source_name을 읽고, page, 읽고, document.page_content를 split() 해 join해 content에 할당, sections에 append 한다. 이 sections를 join해 반환.
reset_rag_cache() 실행시 _vectores.clear() 공급자별 딕셔너리를 비우도록 한다.
# -*- coding: utf-8 -*-
"""정책 PDF를 임베딩하고 FAISS 검색기로 제공하는 RAG 서비스입니다."""
# threading.Lock은 동일 공급자의 인덱스를 여러 요청이 동시에 만들지 않도록 보호합니다.
from threading import Lock
# typing.Literal은 지원 공급자 값을 제한합니다.
from typing import Literal
# Document는 검색 결과 문서 타입을 명시할 때 사용합니다.
from langchain_core.documents import Document
# 공통 설정과 정책 문서 경로를 가져옵니다.
from app.core.settings import DOCS_DIR, get_settings
# 공급자별 임베딩 모델 생성 함수를 가져옵니다.
from app.services.llm_factory import create_embeddings
# _rag_lock은 인덱스 초기화 구간을 직렬화합니다.
_rag_lock = Lock()
# _vectorstores는 공급자별 FAISS 인덱스를 메모리에 캐시합니다.
_vectorstores: dict[str, object] = {}
# _build_vectorstore 함수는 정책 PDF 전체로 FAISS 인덱스를 만듭니다.
def _build_vectorstore(provider: Literal["openai", "gemini"]):
"""정책 PDF 로드, 청크 분할, 임베딩, FAISS 색인을 수행합니다."""
# PDF 로더를 지연 import하여 데이터 도구만 사용할 때 초기화 비용을 줄입니다.
from langchain_community.document_loaders import PyPDFLoader
# FAISS 벡터 저장소 클래스를 가져옵니다.
from langchain_community.vectorstores import FAISS
# 긴 문서를 겹치는 청크로 나누는 분할기를 가져옵니다.
from langchain_text_splitters import RecursiveCharacterTextSplitter
# 현재 RAG 설정값을 읽습니다.
settings = get_settings()
# docs 폴더의 모든 PDF를 파일명에 관계없이 찾습니다.
pdf_files = sorted(DOCS_DIR.glob("*.pdf"))
# 정책 PDF가 없으면 검색기를 만들 수 없으므로 오류를 발생시킵니다.
if not pdf_files:
# 정확한 확인 경로를 포함한 메시지를 제공합니다.
raise FileNotFoundError(f"정책 PDF를 찾을 수 없습니다: {DOCS_DIR}")
# 여러 PDF에서 읽은 페이지 문서를 저장할 리스트입니다.
documents: list[Document] = []
# 각 PDF를 순서대로 로드합니다.
for pdf_file in pdf_files:
# PyPDFLoader가 PDF 페이지를 LangChain Document 목록으로 변환합니다.
loaded_pages = PyPDFLoader(str(pdf_file)).load()
# 각 페이지 메타데이터에 원본 파일명을 추가합니다.
for page in loaded_pages:
# source_name은 답변 근거 표시에 사용됩니다.
page.metadata["source_name"] = pdf_file.name
# 로드한 페이지들을 전체 문서 목록에 추가합니다.
documents.extend(loaded_pages)
# 설정값으로 재귀 문자 분할기를 생성합니다.
splitter = RecursiveCharacterTextSplitter(
chunk_size=settings.rag_chunk_size,
chunk_overlap=settings.rag_chunk_overlap,
)
# 페이지 문서를 검색 가능한 크기의 청크로 나눕니다.
chunks = splitter.split_documents(documents)
# 선택한 공급자의 임베딩 모델을 생성합니다.
embeddings = create_embeddings(provider)
# 모든 청크를 벡터로 변환하고 메모리 FAISS 인덱스를 생성합니다.
return FAISS.from_documents(chunks, embeddings)
# get_vectorstore 함수는 공급자별 인덱스를 한 번만 생성해 반환합니다.
def get_vectorstore(provider: Literal["openai", "gemini"]):
"""캐시된 FAISS 벡터 저장소를 반환합니다."""
# 해당 공급자의 인덱스가 있으면 즉시 재사용합니다.
if provider in _vectorstores:
# 캐시된 인덱스를 반환합니다.
return _vectorstores[provider]
# 동시 최초 생성을 막기 위해 잠금을 획득합니다.
with _rag_lock:
# 대기 중 다른 요청이 생성했는지 다시 확인합니다.
if provider not in _vectorstores:
# 공급자별 인덱스를 생성해 캐시에 저장합니다.
_vectorstores[provider] = _build_vectorstore(provider)
# 최종 인덱스를 반환합니다.
return _vectorstores[provider]
# search_policy 함수는 질문과 관련된 정책 문서를 검색합니다.
def search_policy(query: str, provider: Literal["openai", "gemini"]) -> str:
"""정책 PDF에서 관련 청크를 검색하고 출처와 함께 반환합니다."""
# 공통 설정에서 top_k 값을 읽습니다.
settings = get_settings()
# 선택 공급자의 FAISS 인덱스를 가져옵니다.
vectorstore = get_vectorstore(provider)
# 유사도 검색으로 관련 문서 청크를 찾습니다.
documents = vectorstore.similarity_search(query, k=settings.rag_top_k)
# 결과가 없으면 정책 근거를 찾지 못했다고 반환합니다.
if not documents:
# 정책에 없는 내용을 생성하지 않도록 명시합니다.
return "정책 문서에서 관련 근거를 찾지 못했습니다."
# 검색 결과를 번호와 출처가 포함된 문자열로 변환합니다.
sections: list[str] = []
# 각 검색 결과를 순회합니다.
for index, document in enumerate(documents, start=1):
# 문서 메타데이터에서 원본 파일명을 읽습니다.
source_name = document.metadata.get("source_name", document.metadata.get("source", "출처 미상"))
# 페이지 번호가 있으면 사람이 읽는 1부터 시작하는 번호로 바꿉니다.
page_number = int(document.metadata.get("page", 0)) + 1
# 불필요한 줄바꿈을 정리한 본문을 만듭니다.
content = " ".join(document.page_content.split())
# 검색 순위, 출처, 페이지, 본문을 하나의 섹션으로 저장합니다.
sections.append(f"[{index}] 출처={source_name}, 페이지={page_number}\n{content}")
# 모든 근거 청크를 빈 줄로 구분하여 반환합니다.
return "\n\n".join(sections)
# reset_rag_cache 함수는 모델 변경 또는 문서 수정 후 인덱스를 다시 만들 때 사용합니다.
def reset_rag_cache() -> None:
"""메모리에 캐시된 모든 FAISS 인덱스를 제거합니다."""
# 공급자별 인덱스 딕셔너리를 비웁니다.
_vectorstores.clear()
agents.specialists.py
import dataclasses asdict, dataclass @dataclass 생성자, 출력, 비교 등을 자동으로 만들어주는 데코레이터, asdict dataclass 객체를 python dict 형태로 변환한다.
typing Literal 공급자 이름 제한
app.service.data_service
app.service.rag_service
@dataclass(frozen =true) 생성자를 작성할 필요없다. 자동으로 init, repr, eq등을 만들어주며 frozen을 true로 줘 객체 생성후에 내부 값을 변경할 수 없도록 잠근다.
AgentCard 만들기 에이전트카드리스트 목록 설정하기. 리스트로 조회하기
message를 매개변수로 받는 _extract_order_id는 정규표현식 re를 활용해 주문번호 패턴을 검색해 match 될경우 할당, upper()해 return한다.
agent, message, provider를 받아 메세지에서 주문번호 추출, order_id가 없으면 요청문구 return.
있으면 get_order_status() 주문 결과 반환.
target_agent라면 get_stock() , 나머지 faq, rag, exchange도 마찬가지로 수행.
이 외의 agent를 호출했다면 안내문구 return.
# -*- coding: utf-8 -*-
"""A2A 방식으로 역할을 나눈 전문 에이전트와 에이전트 카드를 정의합니다."""
# dataclass는 에이전트 카드 정보를 간결한 객체로 표현합니다.
from dataclasses import asdict, dataclass
# typing.Literal은 공급자 이름을 제한합니다.
from typing import Literal
# 비즈니스 서비스 함수를 가져옵니다.
from app.services.data_service import get_order_status, get_stock, request_exchange, search_faq
# 정책 RAG 검색 함수를 가져옵니다.
from app.services.rag_service import search_policy
# AgentCard는 A2A 에이전트 발견에 필요한 공개 메타데이터를 표현합니다.
@dataclass(frozen=True)
class AgentCard:
"""전문 에이전트의 이름, 설명, 기술, 엔드포인트 정보를 보관합니다."""
# name은 에이전트의 고유 식별자입니다.
name: str
# description은 에이전트가 수행할 수 있는 역할 설명입니다.
description: str
# skills는 에이전트가 공개하는 작업 능력 목록입니다.
skills: list[str]
# endpoint는 A2A 메시지를 받을 HTTP 경로입니다.
endpoint: str = "/api/v1/a2a/message"
# version은 에이전트 카드 스키마 버전입니다.
version: str = "1.0.0"
# AGENT_CARDS는 외부 에이전트가 발견할 수 있는 전문 에이전트 카드 목록입니다.
AGENT_CARDS = {
"order-agent": AgentCard("order-agent", "주문 상태 전문 에이전트", ["주문번호 조회", "배송 상태 확인"]),
"inventory-agent": AgentCard("inventory-agent", "상품 재고 전문 에이전트", ["상품명 검색", "재고 수량 확인"]),
"faq-agent": AgentCard("faq-agent", "FAQ 검색 전문 에이전트", ["자주 묻는 질문 검색"]),
"policy-agent": AgentCard("policy-agent", "정책 RAG 전문 에이전트", ["환불 정책", "교환 정책", "멤버십 정책"]),
"exchange-agent": AgentCard("exchange-agent", "교환 접수 전문 에이전트", ["주문 검증", "교환 접수"]),
}
# list_agent_cards 함수는 모든 카드를 JSON 직렬화 가능한 딕셔너리로 반환합니다.
def list_agent_cards() -> list[dict[str, object]]:
"""A2A 발견용 에이전트 카드 목록을 반환합니다."""
# dataclass 객체를 일반 딕셔너리로 변환합니다.
return [asdict(card) for card in AGENT_CARDS.values()]
# _extract_order_id 함수는 교육용 메시지에서 O로 시작하는 주문번호를 찾습니다.
def _extract_order_id(message: str) -> str:
"""자연어 메시지에서 O000000 형식의 주문번호를 추출합니다."""
# 정규표현식 모듈을 함수 내부에서 가져옵니다.
import re
# 대소문자와 무관하게 주문번호 패턴을 검색합니다.
match = re.search(r"\bO\d{6}\b", message, flags=re.IGNORECASE)
# 주문번호가 있으면 대문자로 변환해 반환합니다.
if match:
# 일관된 비교를 위해 대문자로 변환합니다.
return match.group(0).upper()
# 주문번호가 없으면 빈 문자열을 반환합니다.
return ""
# delegate_to_agent 함수는 대상 전문 에이전트에 작업을 위임합니다.
def delegate_to_agent(
target_agent: str,
message: str,
provider: Literal["openai", "gemini"],
) -> str:
"""A2A 게이트웨이처럼 대상 에이전트의 능력에 맞춰 요청을 전달합니다."""
# 주문 전문 에이전트 요청인지 확인합니다.
if target_agent == "order-agent":
# 메시지에서 주문번호를 추출합니다.
order_id = _extract_order_id(message)
# 주문번호가 없으면 필요한 형식을 안내합니다.
if not order_id:
# 전문 에이전트가 명확화 요청을 반환합니다.
return "주문번호를 O000000 형식으로 입력해 주세요."
# 실제 주문 서비스 결과를 반환합니다.
return get_order_status(order_id)
# 재고 전문 에이전트 요청인지 확인합니다.
if target_agent == "inventory-agent":
# 교육용 구현에서는 전체 메시지를 상품 검색어로 사용합니다.
return get_stock(message)
# FAQ 전문 에이전트 요청인지 확인합니다.
if target_agent == "faq-agent":
# 전체 메시지를 FAQ 검색 키워드로 전달합니다.
return search_faq(message)
# 정책 RAG 전문 에이전트 요청인지 확인합니다.
if target_agent == "policy-agent":
# 공급자별 임베딩을 사용하여 정책 문서를 검색합니다.
return search_policy(message, provider)
# 교환 접수 전문 에이전트 요청인지 확인합니다.
if target_agent == "exchange-agent":
# 메시지에서 주문번호를 추출합니다.
order_id = _extract_order_id(message)
# 주문번호가 없으면 교환 접수를 중단하고 입력을 요청합니다.
if not order_id:
# 주문 검증 없는 쓰기 작업을 방지합니다.
return "교환할 주문번호를 O000000 형식으로 입력해 주세요."
# 메시지 전체를 교환 사유로 기록하는 교육용 처리입니다.
return request_exchange(order_id, message)
# 등록되지 않은 에이전트 이름이면 오류를 반환합니다.
return f"등록되지 않은 A2A 에이전트입니다: {target_agent}"
app.models.schemas .py
typing Any, Literal 어떤 타입이든 허용하는 Any와 제한하는Literal 모두 가져오기
pydantic
각 질문 api의 field 설정하기
# -*- coding: utf-8 -*-
"""FastAPI 요청 및 응답에 사용하는 Pydantic 스키마 모듈입니다."""
# typing.Any는 실행 추적 정보처럼 다양한 타입의 값을 허용합니다.
from typing import Any, Literal
# BaseModel은 JSON 요청과 응답을 검증하는 기본 클래스입니다.
from pydantic import BaseModel, Field
# ChatRequest는 통합 에이전트 질문 API의 입력 구조입니다.
class ChatRequest(BaseModel):
"""사용자 질문과 세션 정보를 전달하는 요청 모델입니다."""
# message는 에이전트가 처리할 자연어 질문입니다.
message: str = Field(min_length=1, description="사용자 질문")
# thread_id는 LangGraph 메모리에서 대화 세션을 구분합니다.
thread_id: str = Field(default="web-user", min_length=1, description="대화 세션 ID")
# provider는 이번 요청에서 사용할 LLM 공급자입니다.
provider: Literal["openai", "gemini"] = Field(default="openai")
# ChatResponse는 최종 답변과 워크플로우 실행 정보를 반환합니다.
class ChatResponse(BaseModel):
"""통합 워크플로우의 최종 결과 모델입니다."""
# answer는 사용자에게 보여 줄 최종 한국어 답변입니다.
answer: str
# provider는 실제 사용된 모델 공급자입니다.
provider: str
# thread_id는 응답이 속한 대화 세션입니다.
thread_id: str
# route는 분류기가 선택한 처리 경로입니다.
route: str
# trace는 ReAct, RAG, A2A, LangGraph, MCP 단계 실행 기록입니다.
trace: list[dict[str, Any]] = Field(default_factory=list)
# ToolRequest는 개별 데이터 도구 테스트에 사용하는 공통 입력 모델입니다.
class ToolRequest(BaseModel):
"""도구 이름과 도구 인자를 전달하는 요청 모델입니다."""
# tool_name은 실행할 도구의 고유 이름입니다.
tool_name: Literal["get_order_status", "get_stock", "search_faq", "request_exchange"]
# arguments는 각 도구에 필요한 키-값 인자입니다.
arguments: dict[str, Any] = Field(default_factory=dict)
# ToolResponse는 MCP 또는 로컬 도구 실행 결과를 반환합니다.
class ToolResponse(BaseModel):
"""도구 실행 결과와 오류 상태를 반환하는 모델입니다."""
# success는 도구 실행 성공 여부입니다.
success: bool
# tool_name은 실행한 도구 이름입니다.
tool_name: str
# result는 도구가 생성한 문자열 결과입니다.
result: str
# A2AMessageRequest는 에이전트 간 위임 메시지를 표현합니다.
class A2AMessageRequest(BaseModel):
"""A2A 게이트웨이에 전달하는 에이전트 메시지 모델입니다."""
# target_agent는 작업을 받을 전문 에이전트 이름입니다.
target_agent: Literal["order-agent", "inventory-agent", "faq-agent", "policy-agent", "exchange-agent"]
# message는 전문 에이전트가 처리할 사용자 요청입니다.
message: str = Field(min_length=1)
# provider는 전문 에이전트가 사용할 LLM 공급자입니다.
provider: Literal["openai", "gemini"] = Field(default="openai")
# thread_id는 에이전트 간에도 동일하게 유지할 세션 ID입니다.
thread_id: str = Field(default="a2a-user")
app.core.logging_config.py
import json, loggins, datetime
app.core.settings
record 매개변수에서 hasattr request_id가 있다면 payload.requestid에 할당해 jsom.dumps로 반환하기 이때 ensure_ancii False 설정시 유니코드 이스케이프 없이 기록하며 json.dumps는 python객체를 json문자열로 변환하는 것. dump 저장하다 python dict가 json 형태의 str문자열로 변환되어 return된다.
exist_ok True 로그 디렉터리가 없으면 자동으로 생성한다. logger는 프로그램에서 발생하는 정보를 기록하는 객체이다.
final_system 로거를 가져와 setLevel을 info로 설정하고 propagate False 로 로그를 상위 부모에게 전달을 막는다. 기본값을 true로 에러발생시 로그가 부모로 올라간다. False 설정시 로그 중복 출력을 막는다.
logger.handler가 있으면 즉시 반환한다. handler 로그 처리기. logger가 로그를 발생시키는 역할이라면 handler는 로그를 어디로 보낼지를 결정하는 역할이다 .
jsonFormatter객체를 생성해 logging.StremHandler console에 로그를 표시하며 setFormatter로 json 포맷을 적용한다. logging.FileHandler를 활용헤 app..log에 이 내용을 기록하며마찬가지로 setformatter
logger에이 handler를 등록한뒤 logger를 반환한다. logger set 완료.
# -*- coding: utf-8 -*-
"""콘솔과 파일에 동시에 기록하는 구조적 로깅 설정 모듈입니다."""
# json은 로그 레코드를 JSON 문자열로 직렬화합니다.
import json
# logging은 파이썬 표준 로깅 기능을 제공합니다.
import logging
# datetime은 로그 발생 시각을 ISO 형식으로 만듭니다.
from datetime import datetime, timezone
# 프로젝트 로그 경로를 가져옵니다.
from app.core.settings import LOG_DIR
# JsonFormatter는 각 로그를 한 줄 JSON으로 변환합니다.
class JsonFormatter(logging.Formatter):
"""운영 환경에서 검색하기 쉬운 JSON 로그 형식을 생성합니다."""
# format 메서드는 logging이 각 레코드를 출력할 때 호출합니다.
def format(self, record: logging.LogRecord) -> str:
"""로그 레코드를 JSON 문자열로 변환합니다."""
# payload는 로그에서 공통으로 사용할 핵심 필드를 담습니다.
payload = {
"timestamp": datetime.now(timezone.utc).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
# record에 request_id가 있으면 요청 추적을 위해 함께 기록합니다.
if hasattr(record, "request_id"):
# getattr은 선택적 속성을 안전하게 읽습니다.
payload["request_id"] = getattr(record, "request_id")
# ensure_ascii=False는 한국어를 유니코드 이스케이프 없이 기록합니다.
return json.dumps(payload, ensure_ascii=False)
# setup_logging 함수는 중복 핸들러 없이 로깅을 초기화합니다.
def setup_logging() -> logging.Logger:
"""애플리케이션 공용 로거를 생성하고 반환합니다."""
# 로그 디렉터리가 없으면 자동으로 생성합니다.
LOG_DIR.mkdir(parents=True, exist_ok=True)
# 동일 이름의 로거를 가져옵니다.
logger = logging.getLogger("final_system")
# INFO 이상 수준을 기록하도록 지정합니다.
logger.setLevel(logging.INFO)
# 상위 root 로거로 중복 전파되지 않도록 막습니다.
logger.propagate = False
# 이미 핸들러가 있으면 기존 설정을 그대로 재사용합니다.
if logger.handlers:
# 초기화된 로거를 즉시 반환합니다.
return logger
# JSON 포매터 객체를 생성합니다.
formatter = JsonFormatter()
# 콘솔 핸들러는 PyCharm 실행창에 로그를 표시합니다.
console_handler = logging.StreamHandler()
# 콘솔에도 JSON 포맷을 적용합니다.
console_handler.setFormatter(formatter)
# 파일 핸들러는 logs/app.log 파일에 로그를 누적합니다.
file_handler = logging.FileHandler(LOG_DIR / "app.log", encoding="utf-8")
# 파일 로그에도 동일한 JSON 포맷을 적용합니다.
file_handler.setFormatter(formatter)
# 두 핸들러를 공용 로거에 등록합니다.
logger.addHandler(console_handler)
# 파일 핸들러도 등록합니다.
logger.addHandler(file_handler)
# 완성된 로거를 반환합니다.
return logger
app.mcp_server.client.py
import typing Any
app.service.data_service
Local_tool_registry라는 key값과 value설정된 비즈니스 dict.
if 여기 존재하지않으면 안내, tool_name을 가져다가 str형태로 return.
# -*- coding: utf-8 -*-
"""FastAPI 내부에서 MCP 도구와 동일한 비즈니스 함수를 호출하는 클라이언트 추상화입니다."""
# Any는 도구마다 다른 인자 구조를 허용합니다.
from typing import Any
# 실제 데이터 서비스 함수를 가져옵니다.
from app.services.data_service import get_order_status, get_stock, request_exchange, search_faq
# LOCAL_TOOL_REGISTRY는 MCP 서버와 FastAPI가 공유하는 단일 비즈니스 함수 레지스트리입니다.
LOCAL_TOOL_REGISTRY = {
"get_order_status": get_order_status,
"get_stock": get_stock,
"search_faq": search_faq,
"request_exchange": request_exchange,
}
# call_local_mcp_tool 함수는 MCP 도구 호출과 동일한 이름 및 인자 규칙을 적용합니다.
def call_local_mcp_tool(tool_name: str, arguments: dict[str, Any]) -> str:
"""등록된 MCP 도구를 이름으로 찾아 키워드 인자로 실행합니다."""
# 요청한 도구가 레지스트리에 있는지 확인합니다.
if tool_name not in LOCAL_TOOL_REGISTRY:
# 지원하지 않는 도구는 임의 실행하지 않고 오류를 반환합니다.
return f"등록되지 않은 MCP 도구입니다: {tool_name}"
# 도구 이름에 대응하는 파이썬 함수를 가져옵니다.
tool_function = LOCAL_TOOL_REGISTRY[tool_name]
# JSON 객체로 받은 인자를 키워드 인자로 펼쳐 함수를 호출합니다.
return str(tool_function(**arguments))
tools.py
import contextvars 요청마다 공급자 값을 안전하게 보관한다. python에서 실행흐름별로 값을 따로 저장한다. 일반적인 전역 변수라면 provier가 하나만 존재하나 fastapi의 경우 여러 요청이 동시에 들어옴으로 provider값이 섞일 수 있음으로 contextvar를 사용해서 실행 context마다 provider값을 따로 저장할 공간을 만드는 것.
typing literal
langchain_core.tools tool
app.services.data_service
app.service.rag_service
current provider를 contestvar에서 가져다 할당한다.
각 함수이름을 적용한 것들을 @tool로 등록
모든 도구 목록도 정의.
# -*- coding: utf-8 -*-
"""ReAct 에이전트와 MCP 서버가 공통으로 사용하는 LangChain 도구 정의입니다."""
# ContextVar는 요청마다 선택한 공급자 값을 안전하게 보관합니다.
from contextvars import ContextVar
# Literal은 공급자 값을 제한합니다.
from typing import Literal
# tool 데코레이터는 일반 함수를 LLM 호출 가능 도구로 바꿉니다.
from langchain_core.tools import tool
# CSV 기반 비즈니스 로직을 가져옵니다.
from app.services.data_service import get_order_status, get_stock, request_exchange, search_faq
# 정책 RAG 검색 서비스를 가져옵니다.
from app.services.rag_service import search_policy
# current_provider는 현재 요청에서 정책 검색에 사용할 임베딩 공급자를 보관합니다.
current_provider: ContextVar[Literal["openai", "gemini"]] = ContextVar(
"current_provider",
default="openai",
)
# order_status_tool은 주문 질문을 처리하는 읽기 도구입니다.
@tool("get_order_status")
def order_status_tool(order_id: str) -> str:
"""주문번호로 실제 orders.csv의 상품, 금액, 주문일, 배송 상태를 조회한다."""
# 서비스 함수에 주문번호를 전달하고 결과를 그대로 반환합니다.
return get_order_status(order_id)
# stock_tool은 재고 질문을 처리하는 읽기 도구입니다.
@tool("get_stock")
def stock_tool(product_name: str) -> str:
"""상품명으로 실제 inventory.csv의 현재 재고 수량을 조회한다."""
# 서비스 함수에 상품명을 전달하고 실제 재고 결과를 반환합니다.
return get_stock(product_name)
# faq_tool은 자주 묻는 질문을 검색하는 읽기 도구입니다.
@tool("search_faq")
def faq_tool(keyword: str) -> str:
"""배송, 결제, 회원, 주문 등 자주 묻는 질문을 faq.csv에서 검색한다."""
# FAQ 서비스 함수에 검색어를 전달합니다.
return search_faq(keyword)
# policy_tool은 RAG 검색기를 ReAct 도구로 노출합니다.
@tool("policy_search")
def policy_tool(query: str) -> str:
"""환불, 교환, 멤버십 등 사내 정책 PDF에서 근거 문서를 검색한다."""
# 현재 요청의 공급자 값을 ContextVar에서 읽습니다.
provider = current_provider.get()
# 정책 검색 서비스에 질문과 공급자를 전달합니다.
return search_policy(query, provider)
# exchange_tool은 실제 주문을 확인한 뒤 교환을 접수하는 쓰기 행동 도구입니다.
@tool("request_exchange")
def exchange_tool(order_id: str, reason: str) -> str:
"""고객의 명시적 교환 요청을 접수하고 EX 접수번호를 반환한다."""
# 주문번호와 교환 사유를 서비스 함수에 전달합니다.
return request_exchange(order_id, reason)
# ALL_TOOLS는 ReAct 에이전트가 선택할 수 있는 전체 도구 목록입니다.
ALL_TOOLS = [order_status_tool, stock_tool, faq_tool, policy_tool, exchange_tool]
react_agent.py
import threading Lock
Typing Literal
langchain.agents creat_agent
langraph.checkpoint.memory InMemoriySaver
app.agents.tools
app.services.llm_factory
프롬프트 설정
provider를 받아 agent를 가 있을경우 반환, 없으면 _agent_Lock 인 상태로 create_agent() 해 반환
current_provider를 받아다가 provider를 사용해 에이전트를 가져와 invoke(), result 반환, current_provicer,reset() 으로 다음 요청에 공급자 값이 적용되지않도록 함.
# -*- coding: utf-8 -*-
"""도구를 생각하고 실행하고 관찰하는 ReAct 패턴의 전문 에이전트입니다."""
# threading.Lock은 공급자별 에이전트 중복 생성을 방지합니다.
from threading import Lock
# typing.Literal은 허용 공급자를 제한합니다.
from typing import Literal
# create_agent는 최신 LangChain의 production-ready ReAct 도구 루프를 만듭니다.
from langchain.agents import create_agent
# InMemorySaver는 thread_id별 대화 상태를 메모리에 저장합니다.
from langgraph.checkpoint.memory import InMemorySaver
# 전체 비즈니스 도구와 공급자 ContextVar를 가져옵니다.
from app.agents.tools import ALL_TOOLS, current_provider
# 채팅 모델 팩토리를 가져옵니다.
from app.services.llm_factory import create_chat_model
# SYSTEM_PROMPT는 ReAct 에이전트의 역할과 안전 규칙을 정의합니다.
SYSTEM_PROMPT = """
너는 승승장구몰 통합 CS 상담원이다. 반드시 친절한 한국어로 답한다.
질문을 분석하고 필요한 도구를 선택하여 Thought → Action → Observation 방식으로 문제를 해결한다.
주문 상태는 get_order_status, 재고는 get_stock, FAQ는 search_faq를 사용한다.
환불·교환·멤버십 정책은 반드시 policy_search가 반환한 PDF 근거로만 답한다.
고객이 실제 교환 접수를 명시적으로 요청할 때만 request_exchange를 호출한다.
도구가 찾을 수 없다고 반환한 값은 절대로 추측하거나 만들어내지 않는다.
최종 답변에는 내부 추론을 노출하지 말고 확인된 결과와 필요한 안내만 제공한다.
""".strip()
# _agent_lock은 공급자별 에이전트 생성 구간을 보호합니다.
_agent_lock = Lock()
# _agents는 OpenAI와 Gemini 에이전트를 각각 캐시합니다.
_agents: dict[str, object] = {}
# _memory는 모든 에이전트가 thread_id별 대화를 기억하는 체크포인터입니다.
_memory = InMemorySaver()
# get_react_agent 함수는 공급자별 ReAct 에이전트를 한 번만 생성합니다.
def get_react_agent(provider: Literal["openai", "gemini"]):
"""도구 5개와 단기 메모리를 결합한 ReAct 에이전트를 반환합니다."""
# 이미 생성된 공급자의 에이전트가 있으면 즉시 반환합니다.
if provider in _agents:
# 캐시된 에이전트를 재사용합니다.
return _agents[provider]
# 동시 생성을 막기 위해 잠금을 획득합니다.
with _agent_lock:
# 잠금 대기 중 생성 여부를 다시 확인합니다.
if provider not in _agents:
# 선택한 공급자의 LLM 객체를 생성합니다.
model = create_chat_model(provider)
# 최신 create_agent로 도구 호출 루프와 메모리를 결합합니다.
_agents[provider] = create_agent(
model=model,
tools=ALL_TOOLS,
system_prompt=SYSTEM_PROMPT,
checkpointer=_memory,
)
# 생성 또는 캐시된 에이전트를 반환합니다.
return _agents[provider]
# invoke_react_agent 함수는 현재 요청의 공급자를 설정하고 에이전트를 호출합니다.
def invoke_react_agent(message: str, thread_id: str, provider: Literal["openai", "gemini"]) -> str:
"""ReAct 에이전트를 호출하여 최종 텍스트 답변을 반환합니다."""
# 정책 도구가 올바른 임베딩 공급자를 사용하도록 ContextVar 값을 설정합니다.
token = current_provider.set(provider)
# 예외가 나도 ContextVar가 반드시 복구되도록 try-finally를 사용합니다.
try:
# 공급자별 ReAct 에이전트를 가져옵니다.
agent = get_react_agent(provider)
# 동일 thread_id가 같은 대화 메모리를 사용하도록 configurable을 지정합니다.
result = agent.invoke(
{"messages": [{"role": "user", "content": message}]},
config={"configurable": {"thread_id": thread_id}},
)
# 에이전트 결과의 마지막 메시지를 가져옵니다.
final_message = result["messages"][-1]
# 메시지 content를 문자열로 변환해 반환합니다.
return str(final_message.content)
finally:
# 다음 요청에 공급자 값이 섞이지 않도록 이전 ContextVar 상태를 복원합니다.
current_provider.reset(token)
app.graph.workflow .py
import operator
re, time, langgraph.graph end, start, stateGrap
typing Annotated, Literal, TypedDict
app.agent.specialists typedDict Dict의 구조를 정의하는 타입 일반적인 dict는 python이 무엇이 key인지 value인지 판단할 수 없음으로 내부 구조를 알 수 없지만 typeddict를 사용하면 구조가 명확해진다. Annotated 기존 타입에 추가정보를 붙이는 기능.
name: Annotated[str, "사용자 이름"]
Literal 허용가능한 값을 제한하는 타입
app.core.logging_config
app.core.settings
app.mcp_server.client
app.service.rag_service
logger 초기화.
워크플로우 state 클래스 생성
_trace() stage와 detail 를 받아 dict형 list 으로 return.
수령한 worrkflowstate를 state로, strip()후 lower() 적용해 message에 할당.
message에 keyward별로 route, agent 등의 state를 설정해 return.
rag_service에서 pdf에서 관련 엋크를 검색해 출처와 함께 반환했던 함수를 활욯해 context에 할당해 구조를 갖춰 return.
specialists.py에서 target_agent값등에 따라 전문 에이전트에게 위임했던 함수 delegate_to_agent를 가져다가 적용후 answer에 결과값을 할당해 return.
주문번호를 추출해 client.py에서 요청한 도구가 local_tool_registry에 있는지 확인하고 대응하는 함수가 실행되 return했던 call_local_mcp_tool 함수를 통해 return.
react_agent.py에서 적용했던 invoke_react_agent를 활용해 reAct 에이전트를 실행해 anser return.
프롬프트를 추가해서 검색근거를 전달해 invoke_react_agnet 수행. return
error 발생히 return.
route 만 추출해 반환
build_graph.
질문을 분류하는 classify_node를 거쳐 rag 노드르르 등록하고 a2a 위임 노드르르 등록.
mcp 도구 등록.
reAct 자율 처리 등록
rag 근거 답변 생성 노드 등록
error 응답 노드 등록
분류결과에 따라 4경로중 하나로 조건 분기
rag 겁색후 근거기반 답변으로 연결,
최종처리 노드를 그래프 종료점과 연결
a2a, mcp, react 종료
workflow는 모듈 로드시 요청마다 재사용할 수 있도록 변수할당
settings 값을 가져다 for문으로 settings에 최대설정 횟수만큼 실행. _workflow.invoke() result 반환.
예외발생시 logger에 기록하고 if 시도횟수보다 적으면 다시 실행
대기시간 늘리기
스레드 ㄷ기시키기
모든 시도가 실패시 error_node 실행해 응답 안내.
# -*- coding: utf-8 -*-
"""ReAct, RAG, A2A, MCP를 LangGraph 상태 그래프로 연결하는 핵심 워크플로우입니다."""
# operator.add는 노드별 trace 목록을 누적하는 리듀서로 사용합니다.
import operator
# re는 사용자 질문에서 주문번호를 추출할 때 사용합니다.
import re
# time은 재시도 사이의 지수 백오프 대기에 사용합니다.
import time
# typing의 Annotated는 LangGraph 상태 필드에 리듀서를 연결합니다.
from typing import Annotated, Literal, TypedDict
# END와 START는 LangGraph 그래프의 시작과 종료 지점을 나타냅니다.
from langgraph.graph import END, START, StateGraph
# ReAct 전문 에이전트 호출 함수를 가져옵니다.
from app.agents.react_agent import invoke_react_agent
# A2A 위임 함수를 가져옵니다.
from app.agents.specialists import delegate_to_agent
# 환경설정과 로거를 가져옵니다.
from app.core.logging_config import setup_logging
from app.core.settings import get_settings
# 로컬 MCP 클라이언트 추상화를 가져옵니다.
from app.mcp_server.client import call_local_mcp_tool
# 정책 RAG 직접 검색 함수를 가져옵니다.
from app.services.rag_service import search_policy
# 공용 로거를 모듈 로드 시 한 번 초기화합니다.
logger = setup_logging()
# WorkflowState는 그래프 노드 사이에 전달되는 상태 구조입니다.
class WorkflowState(TypedDict, total=False):
"""통합 에이전트 워크플로우에서 공유하는 상태입니다."""
# message는 사용자가 입력한 원문 질문입니다.
message: str
# thread_id는 메모리 세션 식별자입니다.
thread_id: str
# provider는 OpenAI 또는 Gemini 공급자입니다.
provider: Literal["openai", "gemini"]
# route는 분류 노드가 선택한 처리 경로입니다.
route: str
# target_agent는 A2A 위임 대상 전문 에이전트입니다.
target_agent: str
# context는 RAG 또는 MCP에서 수집한 근거입니다.
context: str
# answer는 최종 사용자 답변입니다.
answer: str
# error는 실행 중 발생한 오류 메시지입니다.
error: str
# trace는 노드 실행 정보를 누적합니다.
trace: Annotated[list[dict[str, object]], operator.add]
# _trace 함수는 일관된 단계 추적 딕셔너리를 만듭니다.
def _trace(stage: str, detail: str) -> list[dict[str, object]]:
"""LangGraph 응답에 포함할 단계 추적 레코드를 생성합니다."""
# 단계 이름과 세부 설명을 하나의 원소 리스트로 반환합니다.
return [{"stage": stage, "detail": detail}]
# classify_node는 질문 유형을 결정합니다.
def classify_node(state: WorkflowState) -> WorkflowState:
"""키워드 기반의 결정적 라우팅으로 첫 처리 경로를 선택합니다."""
# 질문을 소문자로 변환하고 공백을 제거합니다.
message = state["message"].strip().lower()
# 교환을 실제로 요청하는 질문은 A2A 교환 에이전트로 보냅니다.
if any(keyword in message for keyword in ["교환 신청", "교환하고", "교환 접수"]):
# 쓰기 작업은 전문 에이전트에 위임합니다.
return {"route": "a2a", "target_agent": "exchange-agent", "trace": _trace("LangGraph", "교환 쓰기 작업을 A2A 경로로 분류")}
# 환불, 정책, 멤버십 질문은 RAG 경로로 보냅니다.
if any(keyword in message for keyword in ["정책", "환불", "멤버십", "반품"]):
# 정책 근거를 먼저 검색하도록 지정합니다.
return {"route": "rag", "target_agent": "policy-agent", "trace": _trace("LangGraph", "정책 질문을 RAG 경로로 분류")}
# 주문번호가 포함되면 MCP 주문 도구를 직접 사용합니다.
if re.search(r"\b[oO]\d{6}\b", message):
# 표준 도구 호출 계층인 MCP 경로를 선택합니다.
return {"route": "mcp", "target_agent": "order-agent", "trace": _trace("LangGraph", "주문번호 질문을 MCP 경로로 분류")}
# 재고 관련 키워드는 A2A 재고 에이전트로 보냅니다.
if any(keyword in message for keyword in ["재고", "남아", "있어"]):
# 전문 에이전트 위임 경로를 선택합니다.
return {"route": "a2a", "target_agent": "inventory-agent", "trace": _trace("LangGraph", "재고 질문을 A2A 경로로 분류")}
# 그 밖의 복합 질문은 ReAct가 도구를 자율 선택하도록 합니다.
return {"route": "react", "trace": _trace("LangGraph", "복합 질문을 ReAct 경로로 분류")}
# rag_node는 정책 문서 근거를 직접 검색합니다.
def rag_node(state: WorkflowState) -> WorkflowState:
"""RAG 단계에서 정책 PDF 근거 청크를 수집합니다."""
# 선택된 공급자 임베딩으로 정책 검색을 실행합니다.
context = search_policy(state["message"], state["provider"])
# 검색 근거와 단계 추적을 상태에 저장합니다.
return {"context": context, "trace": _trace("RAG", "정책 PDF에서 관련 근거 청크 검색 완료")}
# a2a_node는 역할별 전문 에이전트에 작업을 위임합니다.
def a2a_node(state: WorkflowState) -> WorkflowState:
"""A2A 게이트웨이를 통해 전문 에이전트 결과를 받습니다."""
# 분류된 대상 에이전트에 질문을 위임합니다.
answer = delegate_to_agent(state["target_agent"], state["message"], state["provider"])
# 전문 에이전트의 결과를 최종 답변 후보로 저장합니다.
return {"answer": answer, "trace": _trace("A2A", f"{state['target_agent']}에 작업 위임 완료")}
# mcp_node는 MCP 도구 호출 인터페이스를 통해 주문 도구를 실행합니다.
def mcp_node(state: WorkflowState) -> WorkflowState:
"""MCP 클라이언트 계층으로 표준화된 도구 호출을 실행합니다."""
# 질문에서 주문번호를 추출합니다.
match = re.search(r"\b[oO]\d{6}\b", state["message"])
# 주문번호가 없으면 도구 인자를 받을 수 없으므로 명확화 응답을 만듭니다.
if match is None:
# 잘못된 입력을 임의 추정하지 않습니다.
return {"answer": "주문번호를 O000000 형식으로 입력해 주세요.", "trace": _trace("MCP", "주문번호 누락으로 도구 호출 중단")}
# 로컬 MCP 클라이언트 추상화로 주문 도구를 실행합니다.
result = call_local_mcp_tool("get_order_status",
{"order_id": match.group(0).upper()})
# MCP 도구 결과를 최종 답변 후보로 저장합니다.
return {"answer": result, "trace": _trace("MCP", "get_order_status 도구 호출 완료")}
# react_node는 자율 도구 선택이 필요한 질문을 ReAct 에이전트로 처리합니다.
def react_node(state: WorkflowState) -> WorkflowState:
"""최신 LangChain create_agent 기반 ReAct 루프를 실행합니다."""
# ReAct 에이전트에 질문과 메모리 세션을 전달합니다.
answer = invoke_react_agent(state["message"], state["thread_id"], state["provider"])
# 에이전트의 최종 답변을 상태에 저장합니다.
return {"answer": answer, "trace": _trace("ReAct", "도구 선택·실행·관찰 루프 완료")}
# grounded_answer_node는 RAG 근거를 LLM이 읽기 쉬운 질문으로 재구성해 답변합니다.
def grounded_answer_node(state: WorkflowState) -> WorkflowState:
"""RAG 검색 결과만 근거로 최종 정책 답변을 생성합니다."""
# 원 질문과 검색 근거를 ReAct 에이전트에 전달할 메시지로 조합합니다.
grounded_message = (
"다음 정책 문서 근거만 사용하여 질문에 답하세요. 근거에 없으면 모른다고 답하세요.\n\n"
f"[사용자 질문]\n{state['message']}\n\n[검색 근거]\n{state.get('context', '')}"
)
# 기존 thread_id와 공급자로 답변 생성 단계를 실행합니다.
answer = invoke_react_agent(grounded_message, state["thread_id"], state["provider"])
# 최종 답변과 근거 고정 단계를 추적합니다.
return {"answer": answer, "trace": _trace("Grounded Answer", "RAG 근거로만 정책 답변 생성 완료")}
# error_node는 모든 처리 실패를 사용자 친화적인 응답으로 바꿉니다.
def error_node(state: WorkflowState) -> WorkflowState:
"""그래프 오류 상태를 안전한 사용자 응답으로 변환합니다."""
# 내부 상세 오류를 그대로 노출하지 않고 일반 안내를 제공합니다.
answer = "요청 처리 중 오류가 발생했습니다. API 키, 네트워크, 데이터 파일을 확인해 주세요."
# 오류 내용은 추적 정보에 남겨 개발자가 확인할 수 있게 합니다.
return {"answer": answer, "trace": _trace("Error", state.get("error", "알 수 없는 오류"))}
# _route_after_classify 함수는 분류 결과에 맞는 다음 노드를 반환합니다.
def _route_after_classify(state: WorkflowState) -> str:
"""route 상태값을 LangGraph 노드 이름으로 변환합니다."""
# 저장된 route 값을 그대로 반환합니다.
return state["route"]
# _build_graph 함수는 노드와 간선을 연결해 컴파일된 그래프를 만듭니다.
def _build_graph():
"""통합 처리 흐름을 StateGraph로 선언하고 컴파일합니다."""
# WorkflowState 타입을 사용하는 그래프 빌더를 생성합니다.
graph = StateGraph(WorkflowState)
# 질문 분류 노드를 등록합니다.
graph.add_node("classify", classify_node)
# RAG 검색 노드를 등록합니다.
graph.add_node("rag", rag_node)
# A2A 위임 노드를 등록합니다.
graph.add_node("a2a", a2a_node)
# MCP 도구 호출 노드를 등록합니다.
graph.add_node("mcp", mcp_node)
# ReAct 자율 처리 노드를 등록합니다.
graph.add_node("react", react_node)
# RAG 근거 답변 생성 노드를 등록합니다.
graph.add_node("grounded_answer", grounded_answer_node)
# 오류 응답 노드를 등록합니다.
graph.add_node("error", error_node)
# 시작점에서 분류 노드로 연결합니다.
graph.add_edge(START, "classify")
# 분류 결과에 따라 네 경로 중 하나로 조건 분기합니다.
graph.add_conditional_edges(
"classify",
_route_after_classify,
{"rag": "rag", "a2a": "a2a", "mcp": "mcp", "react": "react"},
)
# RAG 검색 후에는 근거 기반 답변 생성으로 연결합니다.
graph.add_edge("rag", "grounded_answer")
# 각 최종 처리 노드를 그래프 종료점과 연결합니다.
graph.add_edge("grounded_answer", END)
# A2A 결과는 바로 종료합니다.
graph.add_edge("a2a", END)
# MCP 결과는 바로 종료합니다.
graph.add_edge("mcp", END)
# ReAct 결과는 바로 종료합니다.
graph.add_edge("react", END)
# 완성된 그래프를 실행 가능한 객체로 컴파일합니다.
return graph.compile()
# _workflow는 모듈 로드 시 한 번 컴파일하여 요청마다 재사용합니다.
_workflow = _build_graph()
# run_workflow 함수는 재시도와 지수 백오프를 포함해 그래프를 실행합니다.
def run_workflow(message: str, thread_id: str, provider: Literal["openai", "gemini"]) -> WorkflowState:
"""통합 LangGraph를 실행하고 최종 상태를 반환합니다."""
# 환경설정에서 재시도 횟수와 대기 기준값을 읽습니다.
settings = get_settings()
# 마지막 오류를 저장할 변수를 초기화합니다.
last_error = ""
# 설정한 최대 횟수만큼 그래프 실행을 시도합니다.
for attempt in range(1, settings.max_retries + 1):
# 네트워크 또는 모델 오류를 처리하기 위해 try-except를 사용합니다.
try:
# 그래프 초기 상태를 구성하여 실행합니다.
result = _workflow.invoke(
{
"message": message,
"thread_id": thread_id,
"provider": provider,
"trace": _trace("Start", f"요청 시작: attempt={attempt}"),
}
)
# 성공한 최종 상태를 반환합니다.
return result
except Exception as exc:
# 예외 메시지를 개발용 변수에 저장합니다.
last_error = f"{type(exc).__name__}: {exc}"
# 실패 정보를 구조적 로그에 남깁니다.
logger.exception("워크플로우 실행 실패: attempt=%s", attempt)
# 마지막 시도가 아니면 지수 백오프 후 다시 시도합니다.
if attempt < settings.max_retries:
# 대기 시간은 base * 2^(시도-1) 공식으로 증가합니다.
delay = settings.retry_base_seconds * (2 ** (attempt - 1))
# 계산된 시간만큼 현재 요청 스레드를 대기시킵니다.
time.sleep(delay)
# 모든 시도가 실패하면 오류 노드를 직접 호출해 안전한 응답을 만듭니다.
return error_node({"error": last_error, "trace": _trace("Retry", "모든 재시도 소진")})
routes.py
import time, fastapi,
app.agents.specialists
app.core.settings
app.models.schemas
app.graph.workflow
app.mcp_server.client
app.services.data_service
app.services.rag_service
router 만들기
@router get health
get_settings()를 읽어 상태 return.
@router post chat
time.perf_counter 시간 기록.
run_workflow() workflow.py에서 정의했던 run_workflow 수행. 답변이 없으면 안내, result에 trace 목록을 복사해 처리시간을 기록.
응답 모델 실행정보를 반환해주는 chatResponse() 반환
@router.post tools/call
call_local_mcp_tool를 통해 도구이름에 맞게 수행한 뒤 toolTesponse형으로 reutn.
에러시 httpexceptions 형으로 return.
@router.get a2a/agents
카드 목록 반환.
@router.post a2a/message
specialists에서 구현한 delegate_to_agent 함수를 활용해 전문 에이전트함수를 실행하여 return.
@router.get data/files 파일목록 return
@router.post rag/reset
캐시삭제후 message와 함께 return.
@router,post exercies/exchange
교환도구이름이 아니면 httpesception 을 raise,
call_local_mcp_tool() 를 실행해 result를 toolresponse() 형으로 reutrn.
@router.get exercises/memory
문자열 return.
# -*- coding: utf-8 -*-
"""FastAPI REST 엔드포인트를 정의하는 라우터 모듈입니다."""
# time은 API 응답 시간을 측정합니다.
import time
# APIRouter는 큰 FastAPI 앱을 기능별 파일로 분리합니다.
from fastapi import APIRouter, HTTPException
# A2A 에이전트 카드와 위임 함수를 가져옵니다.
from app.agents.specialists import delegate_to_agent, list_agent_cards
# 환경설정과 데이터 목록 서비스를 가져옵니다.
from app.core.settings import get_settings
# 요청 및 응답 스키마를 가져옵니다.
from app.models.schemas import A2AMessageRequest, ChatRequest, ChatResponse, ToolRequest, ToolResponse
# 통합 LangGraph 실행 함수를 가져옵니다.
from app.graph.workflow import run_workflow
# MCP 로컬 호출 함수를 가져옵니다.
from app.mcp_server.client import call_local_mcp_tool
# 데이터 파일 목록과 RAG 캐시 초기화 함수를 가져옵니다.
from app.services.data_service import list_data_files
from app.services.rag_service import reset_rag_cache
# router는 /api/v1 아래에 등록될 API 라우터 객체입니다.
router = APIRouter(prefix="/api/v1", tags=["Final System Agent"])
# health 엔드포인트는 서버와 API 키 설정 상태를 확인합니다.
@router.get("/health")
def health() -> dict[str, object]:
"""애플리케이션 상태와 공급자별 API 키 설정 여부를 반환합니다."""
# 현재 환경설정을 읽습니다.
settings = get_settings()
# 실제 키 문자열은 노출하지 않고 존재 여부만 반환합니다.
return {
"status": "ok",
"app": settings.app_name,
"version": settings.app_version,
"openai_key_configured": bool(settings.openai_api_key),
"gemini_key_configured": bool(settings.google_api_key),
}
# chat 엔드포인트는 전체 ReAct-RAG-A2A-LangGraph-MCP 흐름을 실행합니다.
@router.post("/chat", response_model=ChatResponse)
def chat(request: ChatRequest) -> ChatResponse:
"""사용자 질문을 통합 LangGraph 워크플로우로 처리합니다."""
# 워크플로우 실행 시작 시간을 기록합니다.
started_at = time.perf_counter()
# 요청 스키마 값을 그래프 실행 함수에 전달합니다.
result = run_workflow(request.message, request.thread_id, request.provider)
# 최종 답변이 없으면 안전한 기본 메시지를 사용합니다.
answer = result.get("answer", "답변을 생성하지 못했습니다.")
# 기존 trace 목록을 복사합니다.
trace = list(result.get("trace", []))
# 전체 처리 시간을 마지막 추적 단계로 추가합니다.
trace.append({"stage": "Complete", "detail": f"총 처리시간={time.perf_counter() - started_at:.3f}초"})
# Pydantic 응답 모델로 검증된 JSON 결과를 반환합니다.
return ChatResponse(
answer=str(answer),
provider=request.provider,
thread_id=request.thread_id,
route=str(result.get("route", "error")),
trace=trace,
)
# tools 엔드포인트는 MCP와 같은 규칙으로 개별 도구를 테스트합니다.
@router.post("/tools/call", response_model=ToolResponse)
def call_tool(request: ToolRequest) -> ToolResponse:
"""주문, 재고, FAQ, 교환 도구를 LLM 없이 직접 실행합니다."""
# 잘못된 인자 구조를 HTTP 400으로 변환하기 위해 예외를 처리합니다.
try:
# 도구 이름과 인자를 로컬 MCP 클라이언트에 전달합니다.
result = call_local_mcp_tool(request.tool_name, request.arguments)
# 성공 응답 모델을 반환합니다.
return ToolResponse(success=True, tool_name=request.tool_name, result=result)
except TypeError as exc:
# 필수 인자 누락 또는 잘못된 인자 이름을 상세히 안내합니다.
raise HTTPException(status_code=400, detail=f"도구 인자가 올바르지 않습니다: {exc}") from exc
# agent_cards 엔드포인트는 A2A 에이전트 발견 정보를 제공합니다.
@router.get("/a2a/agents")
def agent_cards() -> dict[str, object]:
"""현재 서비스가 제공하는 A2A 전문 에이전트 카드를 반환합니다."""
# 카드 목록을 agents 키로 감싸 반환합니다.
return {"agents": list_agent_cards()}
# a2a_message 엔드포인트는 특정 전문 에이전트에 직접 작업을 위임합니다.
@router.post("/a2a/message")
def a2a_message(request: A2AMessageRequest) -> dict[str, object]:
"""대상 에이전트 이름을 지정하여 A2A 위임 결과를 확인합니다."""
# 전문 에이전트 게이트웨이에 메시지를 전달합니다.
result = delegate_to_agent(request.target_agent, request.message, request.provider)
# A2A 메시지 응답에 대상과 세션 정보를 포함합니다.
return {
"target_agent": request.target_agent,
"thread_id": request.thread_id,
"provider": request.provider,
"result": result,
}
# data_files 엔드포인트는 프로젝트에 포함된 실데이터를 확인합니다.
@router.get("/data/files")
def data_files() -> dict[str, object]:
"""data.zip에서 추출된 전체 파일명과 크기를 반환합니다."""
# 파일 목록을 files 키로 감싸 반환합니다.
return {"files": list_data_files()}
# reset_rag 엔드포인트는 정책 문서 변경 후 인덱스를 초기화합니다.
@router.post("/rag/reset")
def reset_rag() -> dict[str, str]:
"""메모리의 공급자별 FAISS 인덱스를 삭제합니다."""
# 캐시된 벡터 인덱스를 제거합니다.
reset_rag_cache()
# 다음 정책 요청에서 다시 생성된다는 안내를 반환합니다.
return {"message": "RAG 캐시를 초기화했습니다. 다음 정책 요청에서 다시 인덱싱합니다."}
# exercise_exchange는 실습문제 1 해답을 API로 실행합니다.
@router.post("/exercises/exchange", response_model=ToolResponse)
def exercise_exchange(request: ToolRequest) -> ToolResponse:
"""교환신청 쓰기 도구 실습 해답을 실행합니다."""
# 교환 도구가 아닌 요청은 잘못된 실습 호출로 처리합니다.
if request.tool_name != "request_exchange":
# 올바른 tool_name을 안내하는 HTTP 400 오류를 발생시킵니다.
raise HTTPException(status_code=400, detail="tool_name은 request_exchange여야 합니다.")
# 공통 MCP 호출 계층으로 교환 도구를 실행합니다.
result = call_local_mcp_tool(request.tool_name, request.arguments)
# 실습 실행 결과를 반환합니다.
return ToolResponse(success=True, tool_name=request.tool_name, result=result)
# exercise_memory는 서로 다른 thread_id의 메모리 격리 테스트 방법을 반환합니다.
@router.get("/exercises/memory")
def exercise_memory() -> dict[str, object]:
"""실습문제 2의 세션 메모리 격리 테스트 시나리오를 제공합니다."""
# 두 사용자에 대해 서로 다른 thread_id를 사용하도록 시나리오를 반환합니다.
return {
"description": "동일 provider로 /chat을 호출하되 user-a와 user-b에 서로 다른 thread_id를 사용합니다.",
"scenario": [
{"thread_id": "user-a", "message": "환불 절차 알려줘"},
{"thread_id": "user-b", "message": "내가 아까 물어본 정책이 뭐였지?"},
{"thread_id": "user-a", "message": "아까 그 정책 다시 알려줘"},
],
"expected": "user-a만 환불 맥락을 기억하고 user-b는 맥락이 없다고 답해야 합니다.",
}
app.js
각 요소를 변수에 할당
appendMessage() tole에 맞춰 적용, html 객체를 만들어 append messagebox에
for문을 돌려 div와 b span 요소를 만들어 row에 append.
addeventlistener api/v1/chat post형식으로 요청.
const chatForm = document.querySelector("#chatForm");
const messageInput = document.querySelector("#message");
const providerInput = document.querySelector("#provider");
const threadInput = document.querySelector("#threadId");
const messagesBox = document.querySelector("#messages");
const traceBox = document.querySelector("#trace");
const submitButton = chatForm.querySelector("button");
function appendMessage(role, text) {
const article = document.createElement("article");
article.className = `message ${role}`;
const title = document.createElement("b");
title.textContent = role === "user" ? "고객" : "상담원";
const body = document.createElement("p");
body.textContent = text;
article.append(title, body);
messagesBox.appendChild(article);
messagesBox.scrollTop = messagesBox.scrollHeight;
}
function renderTrace(items) {
traceBox.innerHTML = "";
for (const item of items) {
const row = document.createElement("div");
row.className = "trace-item";
const stage = document.createElement("b");
stage.textContent = item.stage;
const detail = document.createElement("span");
detail.textContent = item.detail;
row.append(stage, detail);
traceBox.appendChild(row);
}
}
chatForm.addEventListener("submit", async (event) => {
event.preventDefault();
const message = messageInput.value.trim();
if (!message) return;
appendMessage("user", message);
messageInput.value = "";
submitButton.disabled = true;
submitButton.textContent = "처리 중...";
try {
const response = await fetch("/api/v1/chat", {
method: "POST",
headers: { "Content-Type": "application/json" },
body: JSON.stringify({
message,
thread_id: threadInput.value.trim() || "web-user",
provider: providerInput.value,
}),
});
const payload = await response.json();
if (!response.ok) throw new Error(payload.detail || "API 요청에 실패했습니다.");
appendMessage("assistant", payload.answer);
renderTrace(payload.trace || []);
} catch (error) {
appendMessage("assistant", `오류: ${error.message}`);
} finally {
submitButton.disabled = false;
submitButton.textContent = "통합 워크플로우 실행";
messageInput.focus();
}
});
style.css
꾸며주기
* { box-sizing: border-box; }
body { margin: 0; font-family: Arial, "Noto Sans KR", sans-serif; background: #f4f6fb; color: #172033; }
.layout { width: min(1180px, calc(100% - 32px)); margin: 32px auto; }
.hero { display: flex; justify-content: space-between; align-items: center; gap: 24px; padding: 30px; border-radius: 22px; color: white; background: linear-gradient(135deg, #101827, #263b6a); box-shadow: 0 18px 40px rgba(20, 34, 63, .2); }
.hero h1 { margin: 8px 0; font-size: clamp(28px, 5vw, 46px); }
.hero p { margin: 0; opacity: .85; }
.eyebrow { font-weight: 700; letter-spacing: .12em; font-size: 12px; }
.swagger { padding: 12px 18px; border-radius: 12px; text-decoration: none; color: #10203c; background: white; font-weight: 700; white-space: nowrap; }
.architecture { margin: 22px 0; display: grid; grid-template-columns: repeat(9, auto); align-items: center; gap: 9px; overflow-x: auto; }
.step { min-width: 145px; padding: 16px; border-radius: 15px; background: white; box-shadow: 0 8px 20px rgba(35, 54, 90, .08); }
.step b, .step span { display: block; }
.step span { margin-top: 6px; color: #657087; font-size: 13px; }
.arrow { font-size: 22px; color: #74809a; }
.panel { margin-top: 20px; padding: 24px; border-radius: 20px; background: white; box-shadow: 0 10px 30px rgba(35, 54, 90, .08); }
.controls { display: grid; grid-template-columns: 1fr 2fr; gap: 16px; }
label { display: grid; gap: 7px; font-weight: 700; }
input, select, textarea { width: 100%; padding: 12px; border: 1px solid #d7ddea; border-radius: 10px; font: inherit; }
.messages { min-height: 290px; max-height: 520px; overflow-y: auto; margin: 20px 0; padding: 14px; border-radius: 14px; background: #f7f9fd; }
.message { width: min(82%, 760px); margin: 10px 0; padding: 14px 16px; border-radius: 15px; }
.message p { white-space: pre-wrap; margin-bottom: 0; line-height: 1.65; }
.message.user { margin-left: auto; color: white; background: #2f5fc7; }
.message.assistant { background: white; border: 1px solid #e3e7f0; }
.chat-form { display: grid; grid-template-columns: 1fr 190px; gap: 12px; }
button { border: 0; border-radius: 12px; color: white; background: #1f4eaf; font-weight: 700; cursor: pointer; }
button:disabled { opacity: .55; cursor: wait; }
.trace { display: grid; gap: 10px; }
.trace-item { padding: 12px 14px; border-left: 4px solid #345fc0; border-radius: 8px; background: #f6f8fc; }
.trace-item b { margin-right: 8px; }
@media (max-width: 760px) {
.hero { align-items: flex-start; flex-direction: column; }
.controls, .chat-form { grid-template-columns: 1fr; }
.chat-form button { min-height: 50px; }
}
main.py
import
contextlib asynccontextmanager
기존의 FastAPI 의 애플리케이션 시작, 종료 실행 작업 관리 방식을
@app.on_event("startup")
def startup():
pass
현재 FastAPI가 권장하는 방식을 사용한다.
from contextlib import asynccontextmanager
from fastapi import FastAPI
@asynccontextmanager
async def lifespan(app: FastAPI):
# 서버 시작 시 실행
print("서버 시작")
yield
# 서버 종료 시 실행
print("서버 종료")
app = FastAPI(
lifespan=lifespan
)
fastapi FastAPI, Request API 라우터 등록, Middleware 설정, Template연결 등의 전체 애플리케이션 설정 담당, HTTP 요청 정보를 담는 객체
fastapi.middlewear,cors CORSMiddleware Cross Origin Resource Sharing 미들웨어, 프론트와 백엔드 서버가 다른 도메인일때 발생하는 접근 제한을 해결한다 .
fastapi.responses HTMLResponse FastAPI 에서 HTML 형태의 응답을 반환할때 사용한다.
fastapi.staticfiles StaticFiles 정적파일을 제공하는 기능
fastapi.templating Jinja2Templates HTML Template Engine 사용, 동적인 HTML 생성이 가능하다. <h1> {{ username }} </h1>
app.api.routes router
app.core logging_config setup_lodding
app.core.settings RPOJECT_ROOT, get_settings
@asynccontextmanager 권장코드에 맞도록 lifespan 생성. logger.info()를 수행하고 yield, 종료될때도 logger.info()
app. fastapo() 만들기.
app.add)middelware(coresmiddleware())만들기 cors를 쓰는 이유는 브라우저 보안 정책 때문이며 front와 back이 서로 다른 출처 origin일때 브라우저가 api 요청을 차단하지않도록 허용하는 설정이다.
app.mount static 경로에 정적 파일 폴더 연결하기
app.include_router app에 router 등록하기
@app.get /
templates.TemplateRsponse() index.html 반환하기.
# -*- coding: utf-8 -*-
"""FastAPI 애플리케이션 진입점입니다."""
# asynccontextmanager는 FastAPI lifespan 시작 및 종료 처리를 정의합니다.
from contextlib import asynccontextmanager
# FastAPI는 API 서버 객체를 생성합니다.
from fastapi import FastAPI, Request
# CORSMiddleware는 브라우저의 교차 출처 요청을 제어합니다.
from fastapi.middleware.cors import CORSMiddleware
# HTMLResponse는 루트 웹 UI를 HTML로 반환합니다.
from fastapi.responses import HTMLResponse
# StaticFiles는 CSS와 JavaScript 같은 정적 파일을 제공합니다.
from fastapi.staticfiles import StaticFiles
# Jinja2Templates는 HTML 템플릿을 렌더링합니다.
from fastapi.templating import Jinja2Templates
# API 라우터를 가져옵니다.
from app.api.routes import router
# 로깅과 설정을 가져옵니다.
from app.core.logging_config import setup_logging
from app.core.settings import PROJECT_ROOT, get_settings
# 애플리케이션 환경설정 객체를 한 번 읽습니다.
settings = get_settings()
# 공용 구조적 로거를 초기화합니다.
logger = setup_logging()
# templates 폴더를 사용하는 Jinja2 렌더러를 생성합니다.
templates = Jinja2Templates(directory=str(PROJECT_ROOT / "app" / "templates"))
# lifespan 함수는 서버 시작과 종료 시 실행할 작업을 정의합니다.
@asynccontextmanager
async def lifespan(app: FastAPI):
"""FastAPI 프로세스 생명주기 동안 초기화와 종료 로그를 기록합니다."""
# 서버 시작 정보를 로그에 남깁니다.
logger.info("FastAPI 시작: app=%s version=%s", settings.app_name, settings.app_version)
# yield 전까지가 시작 단계이고 이후가 종료 단계입니다.
yield
# 서버가 정상 종료될 때 로그를 남깁니다.
logger.info("FastAPI 종료")
# FastAPI 앱 객체를 생성하고 Swagger 메타데이터를 설정합니다.
app = FastAPI(
title=settings.app_name,
version=settings.app_version,
description="ReAct → RAG → A2A → LangGraph → MCP 구조를 적용한 통합 CS 에이전트",
lifespan=lifespan,
)
# 설정된 출처에서 브라우저 API 호출을 허용하도록 CORS 미들웨어를 등록합니다.
app.add_middleware(
CORSMiddleware,
allow_origins=settings.cors_origins,
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# /static 경로에 정적 파일 폴더를 연결합니다.
app.mount("/static", StaticFiles(directory=str(PROJECT_ROOT / "app" / "static")), name="static")
# 모든 /api/v1 엔드포인트를 앱에 등록합니다.
app.include_router(router)
# root 함수는 Swagger 없이 사용할 수 있는 웹 채팅 화면을 제공합니다.
@app.get("/", response_class=HTMLResponse)
def root(request: Request):
"""통합 에이전트 테스트 UI를 렌더링합니다."""
# index.html 템플릿에 요청 객체와 앱 이름을 전달합니다.
return templates.TemplateResponse(
request=request,
name="index.html",
context={"app_name": settings.app_name, "version": settings.app_version},
)
index.html
css 불러오기
각 영역 그리기
form chatform을 사용해 submit 버튼 통합 워크플로우를 실행 만들기
<!DOCTYPE html>
<html lang="ko">
<head>
<meta charset="UTF-8">
<meta name="viewport" content="width=device-width, initial-scale=1.0">
<title>{{ app_name }}</title>
<link rel="stylesheet" href="/static/style.css">
</head>
<body>
<main class="layout">
<section class="hero">
<div>
<p class="eyebrow">FINAL SYSTEM · VERSION {{ version }}</p>
<h1>{{ app_name }}</h1>
<p>ReAct → RAG → A2A → LangGraph → MCP 통합 FastAPI 애플리케이션</p>
</div>
<a class="swagger" href="/docs" target="_blank">Swagger 열기</a>
</section>
<section class="architecture">
<div class="step"><b>1. ReAct</b><span>질문 분석과 도구 선택</span></div>
<div class="arrow">→</div>
<div class="step"><b>2. RAG</b><span>정책 PDF 근거 검색</span></div>
<div class="arrow">→</div>
<div class="step"><b>3. A2A</b><span>전문 에이전트 위임</span></div>
<div class="arrow">→</div>
<div class="step"><b>4. LangGraph</b><span>상태·분기·흐름 제어</span></div>
<div class="arrow">→</div>
<div class="step"><b>5. MCP</b><span>표준 도구 호출</span></div>
</section>
<section class="panel">
<div class="controls">
<label>LLM 공급자
<select id="provider">
<option value="openai">OpenAI</option>
<option value="gemini">Gemini</option>
</select>
</label>
<label>Thread ID
<input id="threadId" value="web-user-1001">
</label>
</div>
<div id="messages" class="messages">
<article class="message assistant">
<b>상담원</b>
<p>주문 조회, 상품 재고, FAQ, 환불·교환·멤버십 정책, 교환 신청을 요청해 보세요.</p>
</article>
</div>
<form id="chatForm" class="chat-form">
<textarea id="message" rows="3" placeholder="예: 주문 O000050 상태 알려줘" required></textarea>
<button type="submit">통합 워크플로우 실행</button>
</form>
</section>
<section class="panel">
<h2>실행 추적</h2>
<div id="trace" class="trace"><p>질문을 실행하면 LangGraph 단계가 표시됩니다.</p></div>
</section>
</main>
<script src="/static/app.js"></script>
</body>
</html>
run.py
import uvicorn
직접 실행된 경우에만 univorn.run() 수행해 개발서버 시작하기
# -*- coding: utf-8 -*-
"""PyCharm에서 우클릭 실행할 수 있는 개발 서버 시작 파일입니다."""
# uvicorn은 FastAPI ASGI 애플리케이션을 실행합니다.
import uvicorn
# 직접 실행된 경우에만 개발 서버를 시작합니다.
if __name__ == "__main__":
# reload=True는 코드 수정 시 서버를 자동 재시작합니다.
uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=True)
server.py
독립실행형 MCP server.
일반 python 코드 흐름이면 fom server import mcp 하는데 직접실행하는 파일.
Langchain/reAct용과 MCP 외부 연결용인 server,py가있다.
같은 비즈니스 로직을 여러 ai시스템에서 쓰기 위한것. cursor, claude desktop등에 쓰인다.
data_service.py
|
--------------------------------
| |
↓ ↓
tools.py server.py
LangChain Tool MCP Tool
ReAct Agent Claude/Cursorimport mcp.server.fastmcp FastMCP
app.service.data_service
mcp 생성,
@mcp.tool()
함수연결,
stork, taq, exchange도
직접실행할때 stdio형 MCP 서버를 시작한다.
# -*- coding: utf-8 -*-
"""독립 실행 가능한 Model Context Protocol 도구 서버입니다."""
# FastMCP는 공식 MCP Python SDK가 제공하는 간결한 서버 API입니다.
from mcp.server.fastmcp import FastMCP
# CSV 기반 비즈니스 서비스 함수를 가져옵니다.
from app.services.data_service import get_order_status, get_stock, request_exchange, search_faq
# FastMCP 서버 객체를 이름과 함께 생성합니다.
mcp = FastMCP("승승장구몰 CS Tools")
# mcp_order_status는 주문 조회 기능을 MCP 도구로 공개합니다.
@mcp.tool()
def mcp_order_status(order_id: str) -> str:
"""주문번호로 실제 주문 상태를 조회합니다."""
# 공통 서비스 함수를 호출해 동일한 비즈니스 규칙을 재사용합니다.
return get_order_status(order_id)
# mcp_stock은 재고 조회 기능을 MCP 도구로 공개합니다.
@mcp.tool()
def mcp_stock(product_name: str) -> str:
"""상품명으로 실제 재고 수량을 조회합니다."""
# 공통 재고 서비스 결과를 반환합니다.
return get_stock(product_name)
# mcp_faq는 FAQ 검색 기능을 MCP 도구로 공개합니다.
@mcp.tool()
def mcp_faq(keyword: str) -> str:
"""키워드와 관련된 FAQ를 검색합니다."""
# 공통 FAQ 검색 서비스 결과를 반환합니다.
return search_faq(keyword)
# mcp_request_exchange는 교환 접수 기능을 MCP 쓰기 도구로 공개합니다.
@mcp.tool()
def mcp_request_exchange(order_id: str, reason: str) -> str:
"""실제 주문을 검증하고 교환을 접수합니다."""
# 공통 교환 서비스 함수로 주문 검증과 접수번호 생성을 수행합니다.
return request_exchange(order_id, reason)
# 직접 실행할 때 stdio 전송 방식의 MCP 서버를 시작합니다.
if __name__ == "__main__":
# stdio는 Claude Desktop, Cursor 같은 MCP 클라이언트가 표준 입출력으로 연결할 때 사용합니다.
mcp.run(transport="stdio")
config.yaml
프로젝트에 있는 yaml은 무엇일까?
env와 상당히 많이 역할이 겹친다. 중복으로 남아있는 경우.
pip install pyyaml 후
import yaml
from pathlib import Path
CONFIG_PATH = Path("config.yaml")
def load_config():
with open(CONFIG_PATH, "r", encoding="utf-8") as file:
return yaml.safe_load(file)
코드를 추가하면되지만 env에서 yaml로 바꾸면 코드가 길어진다. 현 코드 pydantic-setting가 이미 좋은 방식이라 바꿀 필요가없다.
# 승승장구몰 CS Agent 설정 파일 (29강)
app:
name: "승승장구 CS Agent"
version: "1.0.0"
language: "ko"
llm:
provider: "gemini" # gemini | openai
model: "gemini-2.5-flash"
temperature: 0.2
max_retries: 3
rag:
embed_model: "models/gemini-embedding-001"
chunk_size: 500
chunk_overlap: 50
top_k: 4
logging:
level: "INFO"
file: "logs/agent.log"
buggy_script,.py
직접 실행할경우 load_sales로 csv파일을 가져다가 각 계산함수를 실행해볼 수있다.
프로젝트나 애플리케이션에서 호출되는 서비스 코드가 아닌 실습용 독립 스크립트.
# -*- coding: utf-8 -*-
"""sales_report.py — 일자별 매출 CSV에서 평균/최대 매출일을 구한다. (버그 포함)"""
import csv
def load_sales(path):
rows = []
with open(path) as f: # 버그1: 한글 csv 인코딩 미지정
reader = csv.DictReader(f)
for r in reader:
rows.append(r)
return rows
def average_sales(rows):
total = 0
for r in rows:
total += r["sales"] # 버그2: 문자열 + 정수
return total / len(rows)
def best_day(rows):
best = rows[0]
for r in rows:
if r["sales"] > best["sales"]: # 버그3: 문자열 비교
best = r
return best
if __name__ == "__main__":
data = load_sales("data/sales_daily.csv")
print("평균 매출:", average_sales(data))
print("최대 매출일:", best_day(data)["date"])
'Personal > SK 네트웍스 AI 캠프' 카테고리의 다른 글
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day52_PEFT 기반 경량 LLM 파인 튜닝 기법 (0) | 2026.07.24 |
|---|---|
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day51_LLM 파인 튜닝 기법과 sLLM_모델 이해 (0) | 2026.07.23 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day49_AI Agent 구현 실습 2 (0) | 2026.07.22 |
| SK 네트웍스 AI 캠프 - 3_초거대언어모델(LLM) - Day48_AI Agent 구현 실습 1 (0) | 2026.07.20 |
| [SK네트웍스 Family AI 캠프] 32기 12주차 회고: Day44 ~ Day47 (0) | 2026.07.16 |
