종합 프로젝트로 완성하는 업무용 AI 어시스턴트
프로젝트 목표 정하기
지금까지 우리는 프롬프트, LCEL, 출력 파싱, 문서 처리, RAG, 대화 상태, 도구 호출, LangGraph, 평가와 운영을 따로 배웠다. 마지막 장에서는 이 기능을 하나의 프로젝트로 묶는다.
만들 앱은 업무용 AI 어시스턴트다. 사용자는 사내 문서에 대해 질문하고, 회의록을 요약하고, 일정 계산을 요청하고, 보고서 초안을 만들 수 있다. 단순 챗봇이 아니라 업무 흐름을 조금씩 대신 처리하는 도구다.
먼저 요구사항을 작게 정의하자.
| 기능 | 설명 |
|---|---|
| 문서 검색 답변 | 사내 문서에서 근거를 찾아 답변하고 출처를 표시한다. |
| 회의록 요약 | 긴 회의록에서 결정 사항, 할 일, 담당자, 기한을 추출한다. |
| 일정 계산 | 영업일 기준 마감일을 계산한다. |
| 보고서 초안 | 주제와 핵심 bullet을 받아 보고서 초안을 작성한다. |
| 대화 상태 | 사용자별 세션을 분리해 이전 대화 맥락을 반영한다. |
| 권한 제어 | 사용자 역할과 부서에 따라 검색 문서와 사용 가능한 도구를 제한한다. |
| 평가 | 출시 전 테스트 질문 세트로 검색, 답변, 도구 호출 품질을 점검한다. |
중요한 기준은 하나다. 모델이 모든 것을 알아서 하게 만들지 않는다. 검색, 권한, 도구 실행, 평가 기준은 애플리케이션 코드가 책임진다. 모델은 언어 이해와 생성에 강하지만, 권한 판단이나 업무 규칙을 완전히 맡길 대상은 아니다.
AI 생성 이미지
전체 구조 설계
업무용 AI 어시스턴트는 다음 흐름으로 구성할 수 있다.
사용자 요청
↓
사용자 인증 / 권한 확인
↓
세션 기록 불러오기
↓
요청 유형 판단
├─ 문서 질문 → 권한 기반 RAG 체인
├─ 회의록 요약 → 요약 도구
├─ 일정 계산 → 계산 도구
└─ 보고서 초안 → 작성 도구
↓
응답 생성
↓
로그 / 평가 데이터 저장고정된 단일 흐름이면 LCEL 체인만으로 충분하다. 하지만 이 프로젝트는 요청 유형에 따라 검색을 하거나 도구를 호출하고, 경우에 따라 추가 질문을 해야 한다. 그래서 실제 서비스에서는 LangGraph로 흐름을 명시하는 편이 좋다. 다만 각 노드 내부의 작업은 여전히 LCEL 체인으로 구성할 수 있다.
프로젝트 폴더는 다음처럼 나눌 수 있다.
company_assistant/
app.py # FastAPI 진입점
chains/
rag.py # RAG 체인
summary.py # 회의록 요약 체인
report.py # 보고서 작성 체인
tools/
business_day.py # 일정 계산 도구
meeting.py # 회의록 도구
report.py # 보고서 도구
graph.py # LangGraph 워크플로
auth.py # 사용자 권한 처리
evals/
test_cases.jsonl # 평가 질문 세트
run_eval.py # 평가 실행 스크립트처음부터 완벽한 구조를 만들 필요는 없다. 그러나 검색, 도구, 권한, 평가 코드를 한 파일에 모두 넣으면 금방 유지보수가 어려워진다. 기능 단위로 나누면 실패 원인도 찾기 쉽다.
권한이 있는 RAG 챗봇 만들기
RAG의 기본 흐름은 이미 배웠다. 질문을 리트리버에 보내 관련 문서를 찾고, 그 문서를 프롬프트의 context로 넣어 답변을 생성한다. 이번 프로젝트에서 달라지는 점은 사용자 권한이다.
사내 문서에는 공개 문서, 인사 문서, 재무 문서, 개발 문서처럼 접근 범위가 다를 수 있다. 따라서 문서를 저장할 때부터 metadata를 넣어야 한다.
from langchain_core.documents import Document
Document(
page_content="2026년 복리후생 제도 개편안 ...",
metadata={
"source": "hr/benefits_2026.pdf",
"department": "hr",
"allowed_roles": ["employee", "hr_manager"],
"allowed_departments": ["hr"],
"version": "2026-01",
},
)여기서 source는 나중에 출처를 표시할 때 쓰고, allowed_roles와 allowed_departments는 검색 권한을 판단할 때 쓴다. 모든 직원이 볼 수 있는 공개 문서라면 부서 제한을 비워 두거나, 별도의 visibility 값을 두는 식으로 규칙을 정하면 된다. 중요한 것은 권한 규칙이 프롬프트 문장에만 들어 있지 않고 metadata와 애플리케이션 코드로 표현되어야 한다는 점이다.
검색 함수는 사용자 정보를 받아 접근 가능한 문서만 반환해야 한다.
def has_access(doc, user):
allowed_roles = doc.metadata.get("allowed_roles", [])
allowed_departments = doc.metadata.get("allowed_departments", [])
role_ok = not allowed_roles or user["role"] in allowed_roles
department_ok = (
not allowed_departments
or user.get("department") in allowed_departments
)
return role_ok and department_ok
def retrieve_docs(question: str, user: dict):
# 예제에서는 검색 후 필터링한다.
# 운영 환경에서는 벡터스토어의 metadata filter나 인덱스 분리로
# 검색 단계부터 권한을 제한하는 것이 안전하다.
docs = retriever.invoke(question)
return [doc for doc in docs if has_access(doc, user)]보안 목적이라면 검색 후 필터링만으로는 부족할 수 있다. 권한 없는 문서가 애플리케이션 메모리로 올라오는 것 자체가 위험하기 때문이다. 운영 환경에서는 부서별 인덱스를 분리하거나, 벡터스토어의 metadata filter를 사용해 검색 단계에서 차단하는 편이 안전하다. 벡터스토어마다 filter 문법은 다르므로, "검색한 뒤 숨기기"가 아니라 "처음부터 검색 대상에서 제외하기"를 기준으로 구현해야 한다.
이제 RAG 체인을 만든다.
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda, RunnablePassthrough
model = ChatOpenAI(model="gpt-4.1-mini", temperature=0)
def format_docs(docs):
if not docs:
return "검색된 문서가 없습니다."
blocks = []
for i, doc in enumerate(docs, start=1):
source = doc.metadata.get("source", "unknown")
blocks.append(f"[문서 {i}] 출처: {source}n{doc.page_content}")
return "nn".join(blocks)
rag_prompt = ChatPromptTemplate.from_messages([
("system", """
당신은 사내 업무 지원 AI입니다.
반드시 제공된 문맥 안에서만 답변하세요.
문맥에 근거가 없으면 모른다고 말하세요.
답변에는 관련 문서 번호를 함께 표시하세요.
"""),
MessagesPlaceholder("history"),
("human", "질문: {question}nn문맥:n{context}"),
])
def add_docs(inputs):
docs = retrieve_docs(inputs["question"], inputs["user"])
return {
**inputs,
"docs": docs,
"context": format_docs(docs),
}
rag_chain = (
RunnableLambda(add_docs)
| RunnablePassthrough.assign(
answer=rag_prompt | model | StrOutputParser()
)
)이 체인의 결과에는 answer뿐 아니라 docs도 남는다. UI에서는 answer를 보여 주고, 출처 목록은 docs의 metadata에서 직접 만든다. 출처를 모델이 생성하게 두면 실제로 존재하지 않는 파일명을 지어낼 수 있다. 모델에게 "문서 번호를 함께 표시하라"고 지시하는 것은 답변과 근거를 연결하기 위한 보조 장치이고, 최종 출처 표시는 코드가 metadata를 기준으로 처리하는 것이 안전하다.
업무 자동화 도구 연결하기
문서 질문만 처리하면 RAG 챗봇이다. 업무용 어시스턴트가 되려면 행동을 수행하는 도구가 필요하다. 여기서는 세 가지 도구를 만든다.
첫째, 회의록 요약 도구다.
from langchain_core.tools import tool
from pydantic import BaseModel, Field
summary_prompt = ChatPromptTemplate.from_messages([
("system", "회의록을 결정 사항, 할 일, 담당자, 기한 중심으로 요약하세요."),
("human", "회의록:n{transcript}"),
])
summary_chain = summary_prompt | model | StrOutputParser()
class MeetingSummaryInput(BaseModel):
transcript: str = Field(description="요약할 회의록 원문")
@tool(args_schema=MeetingSummaryInput)
def summarize_meeting(transcript: str) -> str:
"""회의록을 업무 실행 항목 중심으로 요약한다."""
return summary_chain.invoke({"transcript": transcript})둘째, 영업일 기준 일정 계산 도구다. 이런 계산은 LLM보다 코드가 더 정확하다.
from datetime import date, timedelta
class BusinessDayInput(BaseModel):
start_date: str = Field(description="시작일. YYYY-MM-DD 형식")
days: int = Field(description="더할 영업일 수", ge=0, le=120)
@tool(args_schema=BusinessDayInput)
def add_business_days(start_date: str, days: int) -> str:
"""주말을 제외하고 영업일 기준 마감일을 계산한다."""
current = date.fromisoformat(start_date)
added = 0
while added < days:
current += timedelta(days=1)
if current.weekday() < 5:
added += 1
return current.isoformat()이 예제는 주말만 제외한다. 실제 회사 업무에서는 공휴일, 대체공휴일, 회사 휴무일이 필요할 수 있으므로 휴일 캘린더를 따로 연결해야 한다. 중요한 원칙은 날짜 계산처럼 규칙이 분명한 일은 모델의 추측이 아니라 코드로 처리한다는 것이다.
셋째, 보고서 초안 작성 도구다.
report_prompt = ChatPromptTemplate.from_messages([
("system", "간결한 업무 보고서 초안을 작성하세요. 과장하지 말고 근거와 가정을 구분하세요."),
("human", "주제: {topic}n대상 독자: {audience}n핵심 내용:n{bullets}"),
])
report_chain = report_prompt | model | StrOutputParser()
class ReportDraftInput(BaseModel):
topic: str = Field(description="보고서 주제")
audience: str = Field(description="보고서를 읽을 대상")
bullets: str = Field(description="포함해야 할 핵심 bullet 목록")
@tool(args_schema=ReportDraftInput)
def draft_report(topic: str, audience: str, bullets: str) -> str:
"""주제와 핵심 내용을 바탕으로 업무 보고서 초안을 작성한다."""
return report_chain.invoke({
"topic": topic,
"audience": audience,
"bullets": bullets,
})도구 설명은 짧지만 구체적이어야 한다. 모델은 함수 내부 코드를 읽고 판단하지 않는다. 도구 이름, 설명, 입력 스키마를 보고 어떤 도구를 쓸지 고른다. 그래서 도구 이름은 행동이 드러나게 짓고, 입력 필드에는 형식과 제약을 분명히 적는 편이 좋다.
사용자별 세션과 도구 권한 제한
대화 기록은 사용자와 세션별로 분리해야 한다. 같은 사용자가 여러 업무를 동시에 진행할 수 있고, 다른 사용자의 대화가 섞이면 보안 사고가 된다.
def make_session_id(user_id: str, client_session_id: str) -> str:
return f"{user_id}:{client_session_id}"클라이언트가 보낸 user_id를 그대로 믿으면 안 된다. 실제 서비스에서는 인증 토큰이나 사내 SSO에서 사용자 정보를 확인한 뒤 서버가 user_id, role, department를 결정해야 한다. client_session_id 역시 너무 단순한 값보다는 충분히 예측하기 어려운 값으로 만들고, 서버 쪽 저장소에서 소유자를 확인하는 편이 안전하다.
도구도 사용자 역할에 따라 제한한다.
ALL_TOOLS = [summarize_meeting, add_business_days, draft_report]
def get_tools_for_user(user: dict):
tools = [add_business_days]
if user["role"] in ["employee", "manager", "executive"]:
tools.append(summarize_meeting)
if user["role"] in ["manager", "executive"]:
tools.append(draft_report)
return tools중요한 점은 프롬프트에 "권한이 없으면 쓰지 마세요"라고 말하는 것만으로는 부족하다는 것이다. 권한 없는 도구는 아예 모델에 전달하지 않아야 한다. 모델이 선택할 수 없는 도구는 실행될 수 없다.
문서 검색도 같은 원칙을 따른다. 사용자가 볼 수 없는 문서는 프롬프트에 들어가면 안 된다. "이 문서는 비밀이니 말하지 마"라고 넣는 순간 이미 모델 입력에 비밀을 제공한 것이다.
LangGraph로 요청 흐름 묶기
이제 요청을 어디로 보낼지 정해야 한다. 간단한 방식은 분류 체인으로 요청 유형을 판단한 뒤, RAG 또는 도구 실행으로 분기하는 것이다. 여기서는 분류 결과를 자유 텍스트로 받지 않고 구조화된 출력으로 받는다. 그래야 모델이 "이 요청은 아마 rag입니다"처럼 애매하게 답해도 파싱 문제를 줄일 수 있다.
from typing import Literal
from typing_extensions import TypedDict, NotRequired
from pydantic import BaseModel, Field
class AssistantState(TypedDict):
user: dict
question: str
history: list
route: NotRequired[Literal["rag", "tool"]]
answer: NotRequired[str]
class RouteDecision(BaseModel):
route: Literal["rag", "tool"] = Field(
description="문서 내용 질문이면 rag, 회의록 요약/일정 계산/보고서 작성이면 tool"
)
route_prompt = ChatPromptTemplate.from_messages([
("system", "사용자 요청을 rag 또는 tool 중 하나로 분류하세요."),
("human", "요청: {question}"),
])
route_chain = route_prompt | model.with_structured_output(RouteDecision)
def route_node(state: AssistantState):
decision = route_chain.invoke({"question": state["question"]})
return {"route": decision.route}
def rag_node(state: AssistantState):
result = rag_chain.invoke({
"question": state["question"],
"history": state["history"],
"user": state["user"],
})
return {"answer": result["answer"]}도구 실행 노드는 사용자에게 허용된 도구만 모델에 묶어 실행한다. 단, bind_tools는 모델이 어떤 도구를 호출할지 결정하게 해 줄 뿐, 애플리케이션 대신 도구를 끝까지 실행해 주는 것은 아니다. 모델의 도구 호출 요청을 확인하고, 코드가 실제 도구를 실행한 뒤, 그 결과를 다시 모델에 전달해야 한다.
from langchain_core.messages import HumanMessage, ToolMessage
def tool_node(state: AssistantState):
tools = get_tools_for_user(state["user"])
tools_by_name = {tool.name: tool for tool in tools}
tool_model = model.bind_tools(tools)
messages = [HumanMessage(content=state["question"])]
ai_message = tool_model.invoke(messages)
if not ai_message.tool_calls:
return {"answer": ai_message.content}
tool_messages = []
for tool_call in ai_message.tool_calls[:3]:
tool = tools_by_name.get(tool_call["name"])
if tool is None:
continue
try:
result = tool.invoke(tool_call["args"])
except Exception:
result = "도구 실행 중 오류가 발생했습니다. 입력값을 확인해 주세요."
tool_messages.append(ToolMessage(
content=str(result),
tool_call_id=tool_call["id"],
))
final_message = model.invoke([ai_message, *tool_messages])
return {"answer": final_message.content}위 코드는 흐름을 보여 주기 위한 최소 예시다. 운영용 에이전트라면 사용자 질문까지 포함한 전체 메시지 목록을 유지하고, 도구 호출 반복 한도, 오류 처리, 실행 로그 저장, 민감 정보 마스킹을 함께 구현해야 한다. 도구가 외부 시스템을 변경하는 작업이라면 실행 전 확인 절차를 넣는 것도 좋다.
LangGraph를 쓰면 route 값에 따라 rag_node 또는 tool_node로 이동하도록 조건부 엣지를 만들 수 있다.
from langgraph.graph import StateGraph, START, END
workflow = StateGraph(AssistantState)
workflow.add_node("route", route_node)
workflow.add_node("rag", rag_node)
workflow.add_node("tool", tool_node)
workflow.add_edge(START, "route")
workflow.add_conditional_edges(
"route",
lambda state: state["route"],
{"rag": "rag", "tool": "tool"},
)
workflow.add_edge("rag", END)
workflow.add_edge("tool", END)
assistant_graph = workflow.compile()이 구조의 장점은 업무 규칙이 프롬프트 안에만 숨어 있지 않고 그래프 구조로 드러난다는 점이다. 나중에 "권한이 애매하면 승인 요청으로 보내기", "도구 실행 전 사용자에게 확인받기", "검색 결과가 없으면 되묻기" 같은 절차를 추가할 때도 노드와 엣지를 늘리는 방식으로 확장할 수 있다.
출시 전 평가 세트 만들기
종합 프로젝트는 데모가 잘 된다고 끝이 아니다. 출시 전에 테스트 질문 세트를 만들어야 한다. 평가는 거창하게 시작하지 않아도 된다. 자주 들어올 질문 30개만 모아도 많은 문제가 드러난다.
예를 들어 다음과 같은 평가 데이터를 만들 수 있다.
{"id":"rag_001","type":"rag","question":"2026년 연차 이월 규정은 어떻게 되나요?","expected_source":"hr/leave_policy_2026.pdf","must_include":["이월","기한"]}
{"id":"rag_002","type":"rag","question":"재무팀 전용 예산 지침을 일반 직원이 볼 수 있나요?","expected_behavior":"권한 없음"}
{"id":"tool_001","type":"tool","question":"2026-07-20부터 영업일 기준 10일 뒤는 언제야?","expected_answer":"2026-08-03"}
{"id":"tool_002","type":"tool","question":"다음 회의록에서 할 일을 정리해 줘...","must_include":["담당자","기한","결정 사항"]}평가 지표는 기능별로 나눠야 한다.
| 영역 | 지표 | 확인할 것 |
|---|---|---|
| 검색 | 필요한 문서가 top_k 안에 있는가 | 청크, 임베딩, 필터 설정 |
| 답변 | 검색 문서에 근거해 답했는가 | 환각, 과도한 추론 |
| 출처 | 실제 metadata 기반 출처가 표시되는가 | 가짜 출처 방지 |
| 도구 | 올바른 도구와 인자를 선택했는가 | 스키마, 도구 설명 |
| 권한 | 권한 없는 문서와 도구가 차단되는가 | 보안 규칙 |
| 운영 | 지연 시간과 비용이 허용 범위인가 | 모델 선택, 캐싱, 스트리밍 |
특히 권한 테스트는 반드시 포함해야 한다. 품질이 조금 낮은 답변은 개선할 수 있지만, 권한 없는 정보를 노출하는 답변은 서비스 신뢰를 크게 훼손한다.
평가 결과는 단순 점수보다 실패 유형을 남기는 것이 중요하다.
실패 유형 예시
- 검색 실패: 관련 문서를 찾지 못함
- 근거 부족: 문서에 없는 내용을 답함
- 출처 오류: 답변과 출처가 맞지 않음
- 도구 선택 오류: 일정 계산 대신 일반 답변을 생성함
- 권한 오류: 접근 불가 문서를 사용함
- 형식 오류: 요약 결과에 담당자나 기한이 빠짐실패 사례는 다음 버전의 평가 세트에 추가한다. 이렇게 해야 앱이 좋아지는지 나빠지는지 비교할 수 있다. 처음에는 사람이 직접 판정해도 충분하고, 규모가 커지면 LangSmith 같은 관찰·평가 도구를 연결해 실행 기록과 평가 결과를 함께 관리할 수 있다.
완성 기준
이 프로젝트의 완성 기준은 "모델이 똑똑해 보이는가"가 아니다. 실무 앱이라면 다음 질문에 답할 수 있어야 한다.
사용자의 질문에 필요한 문서를 검색했는가?
답변이 검색 문서에 근거하는가?
출처를 코드가 안정적으로 표시하는가?
권한 없는 문서와 도구가 차단되는가?
대화 기록이 사용자별로 분리되는가?
도구 호출 실패 시 안전하게 응답하는가?
출시 전 평가 세트에서 기준 점수를 넘는가?
운영 중 비용, 지연 시간, 실패율을 추적할 수 있는가?
랭체인을 실무에 적용한다는 것은 여러 추상화를 무작정 많이 쓰는 일이 아니다. 프롬프트는 입력 계약을 만들고, LCEL은 실행 흐름을 연결하며, RAG는 외부 지식을 가져오고, 도구는 실제 행동을 수행하게 한다. LangGraph는 복잡한 절차를 명시하고, LangSmith와 평가 세트는 품질을 확인하게 해 준다.
마지막으로 기억할 점은 단순하다. 좋은 AI 어시스턴트는 모델 하나로 완성되지 않는다. 문서, 권한, 상태, 도구, 평가가 함께 설계될 때 비로소 업무에 사용할 수 있는 애플리케이션이 된다.
이 노트는 보도·비평·교육·연구 목적으로 작성되었습니다. 저작권 문의가 있으시면 에서 알려주세요.