Agentic AI: 모듈화 아키텍처 및 실시간 뉴스 가이드북

 

 

 

AI PIPELINE & CONTENT STRATEGY

Data & AI Intelligence

엔터프라이즈급 Agentic AI: 모듈화 아키텍처 및 실시간 뉴스 가이드북

개요

실무에서 확장 가능한 Agentic AI를 구현하기 위해서는 인프라(LLM/RAG 엔진)와 로직(Retriever)의 분리가 필수입니다. 본 가이드는 Ollama 3.2, PostgreSQL(pgvector), 실시간 뉴스 API를 결합한 엔터프라이즈급 모듈형 아키텍처와 상세 코드, 그리고 챗봇 콘텐츠 기획부터 카피라이팅까지의 전 과정을 상세히 안내합니다.

Ollama 3.2

pgvector (PostgreSQL)

LangGraph

FastMCP

LangSmith

목차

1. 핵심 기술 용어 및 알고리즘 해설 2. 프로젝트 모듈 구조 및 파일 역할 3. 환경 구축 및 프로젝트 초기화 4. 핵심 엔진 및 모듈 구현 5. 챗봇 콘텐츠 기획 및 대화 시나리오 설계 6. 아키텍처 워크플로우 7. 단계별 기술적 원리 분석 8. 자가 교정(Self-Correction) 핵심 로직 9. 고급 튜닝 및 최적화 전략 10. 시스템 안정화 이후 고도화 로드맵

핵심 기술 용어 및 알고리즘 해설

Q: pgvector는 벡터 DB인가요? 아니면 디비 안의 알고리즘인가요?
A: pgvector는 PostgreSQL이라는 기존 관계형 데이터베이스(RDB) 내에서 벡터 데이터를 저장하고 검색할 수 있게 해주는 확장 프로그램(Extension)입니다. 즉, 별도의 독립된 벡터 전용 DB라기보다 PostgreSQL을 벡터 DB로 변신시켜주는 핵심 컴포넌트입니다. 그 내부에서 HNSW나 IVFFlat 같은 고도의 검색 알고리즘을 선택하여 구동할 수 있습니다.

LLM (Ollama 3.2)

설명

로컬 인프라 기반 추론용 모델

쉬운설명

내 컴퓨터에서 돌아가는 똑똑한 AI 두뇌

예시

인터넷 없이 동작하는 문서 요약 AI

RAG

설명

검색 기반의 답변 생성 프레임워크

쉬운설명

AI에게 교과서를 주고 시험 문제를 풀게 하는 것

예시

사내 매뉴얼 PDF를 기반으로 답하는 챗봇

pgvector (PostgreSQL)

설명

PostgreSQL용 오픈소스 벡터 유사도 검색 확장

쉬운설명

관계형 DB에 AI의 기억 장치를 이식한 형태

Chunking (청킹)

설명

방대한 텍스트를 LLM이 소화 가능한 최적의 크기로 분할하는 전처리 과정

쉬운설명

두꺼운 책을 AI가 이해하기 쉽게 주제별로 요약 쪽지를 만드는 작업

Semantic Chunking

설명

단순 글자수가 아닌 의미적 유사도에 기반한 지능형 분할 방식

쉬운설명

문장이 중간에 잘리지 않도록 문맥이 끝나는 지점을 찾아 쪼개는 기술

LangChain

설명

AI 컴포넌트 오케스트레이션 프레임워크

쉬운설명

AI를 데이터와 도구에 잇는 접착제

LangGraph

설명

상태 기반 순환 에이전트 워크플로우

쉬운설명

AI가 막히면 스스로 다른 방법을 시도하는 설계도

MCP (Context Protocol)

설명

Model Context Protocol, 외부 도구와 컨텍스트를 연결하는 표준

쉬운설명

AI에게 도구를 끼워 쓰는 만능 플러그

LangSmith

설명

AI 파이프라인 관측 및 디버깅 플랫폼

쉬운설명

AI가 왜 틀렸는지 확인하는 블랙박스

HNSW Index

설명

계층적 근사 근접 이웃 검색 알고리즘

쉬운설명

데이터가 많아도 길을 잃지 않게 지도를 그리는 기술

Cosine Similarity

설명

두 벡터 간의 코사인 각도를 측정하여 유사도 계산

쉬운설명

단어의 의미가 얼마나 비슷한 방향을 보고 있는지 계산하는 수학 공식

L2 Distance (Euclidean)

설명

두 점 사이의 직선 거리를 측정하는 벡터 검색 방식

쉬운설명

벡터 공간에서 두 정보 사이의 실제 거리를 재는 방법

IVFFlat

설명

Inverted File Index, 데이터를 클러스터로 나누어 검색 범위 축소

쉬운설명

비슷한 것끼리 상자에 미리 담아두어 찾는 속도를 높이는 기술

PRO-TIP 왜 청킹(Chunking)이 필수인가요?

📍 컨텍스트 제한 해결
LLM의 한정된 토큰 처리량을 효율적으로 배분합니다.

🎯 검색 정확도 향상
최소 단위 텍스트 추출로 답변의 노이즈를 제거합니다.

⚡ 비용 및 속도 최적화
입력값을 최소화하여 추론 속도와 API 비용을 개선합니다.

프로젝트 모듈 구조 및 파일 역할

전문적인 에이전트 시스템은 관심사의 분리(Separation of Concerns)가 핵심입니다.

project_root/
├── main.py                # 통합 실행: 전체 파이프라인 제어 및 에이전트 구동 레이어
├── .env
├── ingest.py              # 데이터 적재: 지식 베이스 구축 전용 스크립트
├── src/					# [보안]: API 키, DB 접속 비밀번호 등 민감 정보 저장 (Git 제외 필수)
│   ├── config.py          # 환경 설정: API 키, 데이터베이스 연결 URL 및 전역 설정 관리
│   ├── llm_engine.py      # 추론 엔진: Ollama 3.2 모델 인스턴스 생성 및 파라미터 최적화
│   ├── pg_engine.py       # 데이터 관리: pgvector 연동, 의미론적 청킹 및 데이터 추출 로직
│   ├── retrieval.py       # 검색 전략: Vector DB + BM25를 결합한 Hybrid Retriever 구현
│   ├── content_plan.py    # 기획 정의: 챗봇 페르소나 설계 및 대화 시나리오/카피라이팅 정의
│   ├── graph.py           # 워크플로우: LangGraph 기반 상태 관리 및 자가 교정(Self-Correction) 순환 로직
│   └── news_tool.py       # 외부 도구: MCP 규격 기반 실시간 뉴스 검색 API 모듈화
└── utils/                 # 공통 유틸: 커스텀 로깅, 시간 측정 및 시스템 예외 처리 유틸리티

환경 구축 및 프로젝트 초기화

실시간 뉴스 API 데이터 사이트 추천:

• NewsAPI.org: 범용적인 뉴스 검색 API (가장 많이 사용됨)
• GNews API: Google News 기반의 검색 엔진 API
• SerpApi: 실제 검색 엔진 결과값을 JSON으로 파싱하여 제공

Infra Phase 01

Core Infrastructure (Ollama & pgvector) 에이전트의 두뇌(LLM)와 기억 장치(Vector DB) 로컬 서버를 구축합니다.

 

 

 

# 1. Ollama 설치 및 LLM 모델 다운로드 ollama pull llama3.2 # 2. Docker를 이용한 pgvector(PostgreSQL) 서버 가동 docker run --name vectordb -e POSTGRES_PASSWORD=your_password -p 5432:5432 -d ankane/pgvector # [선택] DB 접속 후 벡터 확장 기능 활성화 SQL CREATE EXTENSION IF NOT EXISTS vector;

Infra Phase 02

Python Library Installation 프레임워크 오케스트레이션 및 DB 연동을 위한 필수 라이브러리를 설치합니다.

 

 

 

# 프로젝트 루트 폴더 터미널에서 아래 명령어 통합 실행 pip install langchain langchain-community langchain-ollama langchain-postgres langgraph pip install pydantic-settings requests "psycopg[binary]" mcp fastmcp langsmith

Infra Phase 03

Project Directory Structure 시스템이 문서를 읽고 코드를 인식할 수 있도록 기초 폴더 구조를 생성합니다.

 

 

 

# 1. 지식 베이스(원본 문서) 적재 폴더 생성 mkdir -p data/documents # 2. 모듈화 소스 코드 관리 폴더 생성 mkdir src # [Tip] data/documents/ 폴더 안에 분석용 .txt 파일들을 미리 준비해 주세요.

🚀 구축 완료 안내

이제 엔터프라이즈급 Agentic AI를 구동하기 위한 모든 물리적 준비가 끝났습니다. 이어지는 섹션부터는 앞서 정의한 모듈별 아키텍처에 따라 실제 파이썬 코드를 구현해 보겠습니다.

핵심 엔진 및 모듈 구현

Environment Setup

.env 프로젝트 루트에 생성하며, 민감한 API 키와 데이터베이스 접속 정보를 격리하여 관리합니다.

 

 

 

# [Security] 이 파일은 절대로 버전 관리 시스템(Git)에 올리지 마세요. # 1. LLM 인프라 설정 OLLAMA_MODEL="llama3.2" # 2. PostgreSQL / pgvector 접속 정보 # 형식: postgresql+psycopg://사용자:비밀번호@호스트:포트/DB명 PG_URL="postgresql+psycopg://user:pass@localhost:5432/vectordb" COLLECTION_NAME="enterprise_kb" # 3. 실시간 뉴스 검색 API 키 (NewsAPI.org) NEWS_API_KEY="your_actual_api_key_here" # 4. 모니터링 및 관측성 (LangSmith) LANGCHAIN_TRACING_V2="true" LANGCHAIN_API_KEY="ls__your_key_here"

Step 1

src/config.py 시스템 환경 변수를 중앙 집중형으로 관리합니다.

 

 

 

from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): OLLAMA_MODEL: str = "llama3.2" PG_URL: str = "postgresql+psycopg://user:pass@localhost:5432/vectordb" NEWS_API_KEY: str = "your_api_key_here" COLLECTION_NAME: str = "enterprise_kb" settings = Settings()

Step 2

src/llm_engine.py 로컬 추론 엔진 인스턴스를 생성합니다.

 

 

 

from langchain_ollama import ChatOllama from src.config import settings class LLMEngine: def get_model(self): return ChatOllama( model=settings.OLLAMA_MODEL, temperature=0, num_ctx=8192 )

Step 3

src/pg_engine.py 지능형 청킹 및 pgvector 적재 로직이 통합된 엔진입니다.

 

 

 

from langchain_postgres import PGVector from langchain_ollama import OllamaEmbeddings from langchain.text_splitter import RecursiveCharacterTextSplitter from src.config import settings class PGEngine: def __init__(self): # 1. 임베딩 모델 인스턴스 생성 필수 self.embeddings = OllamaEmbeddings(model=settings.OLLAMA_MODEL) # 2. 지능형 청킹 설정 self.text_splitter = RecursiveCharacterTextSplitter( chunk_size=1000, chunk_overlap=150, separators=["\n\n", "\n", ".", " ", ""] ) # 3. pgvector 초기화 (최신 connection_string 규격) self.vector_store = PGVector( embeddings=self.embeddings, collection_name=settings.COLLECTION_NAME, connection_string=settings.PG_URL, use_jsonb=True ) def process_and_save(self, documents): chunks = self.text_splitter.split_documents(documents) self.vector_store.add_documents(chunks) # DB 적재 수행 return len(chunks) def get_all_texts(self): try: docs = self.vector_store.similarity_search("", k=100) return [d.page_content for d in docs] if docs else [] except Exception: return []

Step 4

src/news_tool.py (고도화된 Tools 세트) FastMCP 규격을 활용하여 LLM이 직접 사용 가능한 실시간 뉴스 분석 툴 모음입니다.

 

 

 

import asyncio from mcp.server.fastmcp import FastMCP import requests from src.config import settings # MCP 서버 인스턴스 생성 mcp = FastMCP("NewsExplorer") @mcp.tool() async def get_latest_news(query: str) -> str: """지정된 키워드에 대한 실시간 종합 뉴스를 검색합니다.""" url = f"https://newsapi.org/v2/everything?q={query}&apiKey={settings.NEWS_API_KEY}" response = requests.get(url).json() articles = response.get("articles", [])[:3] return "\n".join([f"[{a['title']}] {a['description']}" for a in articles]) @mcp.tool() async def get_economic_briefing(topic: str) -> str: """특정 경제 지표나 시장 동향에 대한 전문 브리핑 뉴스를 필터링하여 가져옵니다.""" url = f"https://newsapi.org/v2/top-headlines?category=business&q={topic}&apiKey={settings.NEWS_API_KEY}" response = requests.get(url).json() articles = response.get("articles", [])[:2] return "경제 동향 리포트:\n" + "\n".join([f"- {a['title']}" for a in articles]) @mcp.tool() async def search_tech_trends(tech_keyword: str) -> str: """최신 IT 기술 트렌드 및 기술 블로그 뉴스를 심층 검색합니다.""" url = f"https://newsapi.org/v2/everything?q={tech_keyword}+tech+trend&sortBy=relevancy&apiKey={settings.NEWS_API_KEY}" response = requests.get(url).json() articles = response.get("articles", [])[:3] return "기술 트렌드 분석:\n" + "\n".join([f"▶ {a['title']}" for a in articles]) if __name__ == "__main__": mcp.run()

Step 5

src/retrieval.py Vector + BM25 하이브리드 검색 전략 (안정성 강화)

 

 

 

from langchain.retrievers import EnsembleRetriever from langchain_community.retrievers import BM25Retriever class AdvancedRetriever: def __init__(self, vector_store, docs_list): # [교정] 빈 리스트로 인한 초기화 에러 방지 가드 로직 safe_list = docs_list if docs_list else ["지식 베이스 초기화 중"] vector_r = vector_store.as_retriever(search_kwargs={"k": 3}) bm25_r = BM25Retriever.from_texts(safe_list) self.hybrid_retriever = EnsembleRetriever( retrievers=[vector_r, bm25_r], weights=[0.7, 0.3] ) def invoke(self, query: str): return self.hybrid_retriever.invoke(query)

Step 6

src/graph.py LangGraph 기반 자가 교정 워크플로우 (루프 제어 최적화)

 

 

 

from typing import TypedDict, List from langgraph.graph import StateGraph, END from src.llm_engine import LLMEngine class AgentState(TypedDict):     query: str     context: List[str]     answer: str     is_hallucinated: bool     retry_count: int def call_model(state: AgentState):     llm = LLMEngine().get_model()     context_str = "\n\n".join(state["context"])     prompt = f"Context: {context_str}\n\nQuestion: {state['query']}"          response = llm.invoke(prompt)     answer = response.content          # [교정] 답변 품질 판별 및 retry_count 기반 탈출 조건 준비     is_bad = ("모르겠습니다" in answer or "정보가 없습니다" in answer)          return {         "answer": answer,         "is_hallucinated": is_bad,         "retry_count": state["retry_count"] + 1     } def grade_answer(state: AgentState):     # [교정] 재시도 횟수 제한(1회)을 두어 무한 루프 방지     if state["is_hallucinated"] and state["retry_count"] < 2:         return "agent"     return END workflow = StateGraph(AgentState) workflow.add_node("agent", call_model) workflow.set_entry_point("agent") workflow.add_conditional_edges("agent", grade_answer) agent_app = workflow.compile()

Step 7

ingest.py 기존 데이터를 삭제(Drop)한 후 최신 지식을 새로 적재하는 스크립트입니다.

 

 

 

from src.pg_engine import PGEngine from langchain_community.document_loaders import DirectoryLoader, TextLoader def run_ingest(): pg = PGEngine() # 1. 기존 데이터 초기화 (Drop Collection) print("기존 데이터 삭제 및 테이블 초기화 중...") pg.vector_store.drop_tables() # 2. 로컬 데이터 로드 print("최신 지식 문서를 불러오는 중...") loader = DirectoryLoader("./data/documents", glob="*.txt", loader_cls=TextLoader) raw_docs = loader.load() # 3. 새로운 데이터 적재 pg.process_and_save(raw_docs) print(f"적재 완료! 총 {len(raw_docs)}개의 문서가 새로 저장되었습니다.") if __name__ == "__main__": run_ingest()

Step 8

main.py 통합 실행 레이어 (Document 객체 파싱 및 비동기 통합)

 

 

 

import asyncio from src.pg_engine import PGEngine from src.news_tool import get_latest_news from src.retrieval import AdvancedRetriever from src.graph import agent_app async def main(): pg = PGEngine() query = "2026년 AI 시장 전략 보고" # 검색 엔진 초기화 (BM25 통계 구축 포함) retriever = AdvancedRetriever(pg.vector_store, pg.get_all_texts()) db_docs = retriever.invoke(query) news_text = await get_latest_news(query) # 에이전트 실행 (Document 객체 파싱 후 주입) inputs = { "query": query, "context": [d.page_content for d in db_docs] + [news_text], "retry_count": 0 } result = await agent_app.ainvoke(inputs) print(f"최종 결과:\n{result['answer']}") if __name__ == "__main__": asyncio.run(main())

챗봇 콘텐츠 기획 및 대화 시나리오 설계

💡 LLM CONTENT ENGINEERING STRATEGY

페르소나 및 카피라이팅 설계

단순 답변 생성을 넘어 브랜드의 목소리를 입히는 과정입니다.

페르소나: "냉철한 시장 분석 전문가"
카피라이팅 예시: "최신 지식 베이스와 실시간 뉴스를 종합 분석한 결과, 해당 트렌드의 신뢰도는 85%로 산출됩니다."

대화 시나리오 (Happy Path)

1

의도 파악: 사용자의 질문을 분석하여 내부 지식 우선인지 실시간 정보 우선인지 판단.

2

하이브리드 검색: pgvector와 뉴스 API를 병렬로 가동하여 최적의 컨텍스트 확보.

3

자가 교정: 확보된 정보 간 모순이 발견되면 스스로 검색 쿼리를 수정하여 재수행.

예외 처리 (Fallback Scenario)

데이터 부재 시 "현재 지식 베이스에는 없지만, 관련 실시간 뉴스를 탐색해 드릴까요?"라고 반문하여 대화의 맥락을 유지합니다.

아키텍처 워크플로우

전체 시스템 흐름도

Step 1 시스템 초기화 및 하이브리드 인덱싱

PGEngine을 통해 DB 연결을 활성화하고 벡터와 키워드 데이터를 결합하여 리트리버를 구축합니다.

 

Step 2 다중 소스 데이터 수집

사용자 질의 시 내부 Vector DB와 실시간 뉴스 API를 비동기로 호출하여 컨텍스트를 확보합니다.

 

Step 3 에이전트 추론 및 답변 생성

LangGraph 상태값에 컨텍스트를 주입하고 Ollama 3.2 모델이 페르소나에 맞춰 답변을 생성합니다.

 

Step 4 자가 교정 및 검증 루프

답변 품질을 판별하여 할루시네이션이 의심될 경우 최대 1회 재시도 로직을 수행합니다.

 

Step 5 최종 출력 및 관측성 기록

검증된 답변을 사용자에게 제공하고 전체 과정을 LangSmith에 기록하여 최적화합니다.

단계별 기술적 원리 분석

Analysis Part 1: 데이터 전처리 및 인덱싱

[SYSTEM INITIALIZATION & INDEXING] 1. [main.py] PGEngine 인스턴스 생성 및 DB 연결 활성화 2. [pg_engine.py] RecursiveCharacterTextSplitter를 통한 의미 단위 청킹 실행 3. [pg_engine.py] get_all_texts()로 청킹된 모든 조각 추출 (BM25 사전 구축용) 4. [retrieval.py] pgvector(의미)와 BM25(키워드)를 결합한 하이브리드 리트리버 초기화

Analysis Part 2: 다중 소스 검색 및 추론

[SEARCH & INFERENCE PIPELINE] 5. [main.py] 사용자 질의 수신 및 retriever.invoke() 실행 6. [news_tool.py] FastMCP 도구를 통해 실시간 뉴스 API 비동기 수집 (Parallel) 7. [main.py] 내부 DB 컨텍스트와 외부 뉴스 데이터를 하나로 결합 (Context Augmentation) 8. [graph.py] LangGraph 상태값 주입 및 Ollama 3.2 모델 1차 답변 생성

Analysis Part 3: 결과 검증 및 루프 전이 (Bridge)

[VALIDATION & LOOP TRANSITION] 9. [graph.py] call_model의 출력값 내 '모르겠습니다' 등 특정 키워드 패턴 매칭 10. [graph.py] is_hallucinated 상태값을 True로 전환하여 조건부 엣지 트리거 11. [Transition] 검증 노드 판독 결과에 따라 7번 자가 교정 로직으로 제어권 위임

자가 교정(Self-Correction) 핵심 로직 상세

Logic Phase 01

System Reasoning Flow 답변의 품질을 검증하고 실행 경로를 결정하는 제어 로직입니다.

 

 

 

[REASONING LOOP CONTROL] 12. [Decision] grade_answer 노드가 상태값 검사 13. [Route A] True (Bad): retry_count < 2 이면 'agent' 노드로 루프 백 14. [Route B] False (Good): 즉시 END로 진입하여 최종 답변 확정

💡

핵심 로직 분석

Point 01. 상태 기반 분기

6번 단계에서 판별된 is_hallucinated 값에 따라 진행 방향이 결정됩니다. 이는 고정된 파이프라인이 아닌 AI가 스스로 판단하는 Agentic Workflow의 정점입니다.

Point 02. 무한 루프 방지

retry_count를 체크하여 무한 재시도를 방지합니다. 엔터프라이즈 환경에서 응답 지연과 리소스 낭비를 막는 핵심 안전장치입니다.

Point 03. 컨텍스트 재주입

재시도 시 기존 컨텍스트를 유지하거나 쿼리를 수정하여 새로운 정보를 수집할 수 있는 구조적 유연성을 제공합니다.

🚩

실무자 팁 (Pro-Tip) 자가 교정 루프 설계 시 무한 루프에 빠지지 않도록 반드시 최대 재시도 횟수(Max Retry)를 설정해야 합니다.

고급 튜닝 및 최적화 전략

RAG 검색 품질 고도화

설명

검색 정확도 향상을 위한 하이브리드 전략

쉬운설명

단어 의미와 핵심 키워드를 함께 검색하여 정확한 문서 찾기

에이전트 성능 최적화

설명

에이전트 판단 오류 최소화 전략

쉬운설명

AI가 실수를 줄이도록 모범 답안 예시 제공

인프라 최적화

설명

연산 리소스 및 비용 절감

쉬운설명

AI가 똑같은 생각을 반복하지 않게 미리 기억해두기

pgvector HNSW 최적화

설명

대규모 벡터 데이터 검색 가속화 인덱싱

쉬운설명

수백만 개의 데이터 중 가장 비슷한 것을 0.1초 만에 찾기

시스템 안정화 이후 고도화 로드맵

Phase 1: Contextual Compression 도입

단순 검색을 넘어, 검색된 여러 청크 중 질문과 가장 관련 높은 핵심 문장만 압축하여 LLM에 전달합니다. 이는 토큰 비용을 줄이고 답변의 정확도를 극대화하는 다음 단계의 최적화입니다.

Phase 2: Hybrid Retrieval (Vector + Sparse)

pgvector의 의미 검색과 PostgreSQL의 Full-text search(tsvector)를 결합합니다. 고유 명사나 제품 코드 검색 품질을 비약적으로 상승시키는 실무 필수 과정입니다.

Phase 3: Self-Correction 순환 아키텍처

LangGraph의 순환 로직을 도입하여 Answer Grader 노드를 추가합니다. 할루시네이션이 감지될 경우 에이전트가 스스로 검색 쿼리를 수정하여 재시도하게 하는 "생각하는 AI"를 완성합니다.

Category: Data & AI Intelligence

TOP