랭체인(LangChain)과 RAG, AI 에이전트 구축 완벽 가이드 (2편)

import openai
import os

# 1. API 키 설정 (환경변수 방식 권장)
os.environ['OPENAI_API_KEY'] = "YOUR_API_KEY"
client = openai.OpenAI()

# 2. 질문과 역할 정의
messages = [
{"role": "system", "content": "너는 어려운 IT 개념을 초등학생도 알기 쉽게 설명해주는 친절한 선생님이야."},
{"role": "user", "content": "RAG(검색 증강 생성)가 뭐야?"}
]

# 3. 모델 호출 및 결과 출력
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.3, # 0에 가까울수록 차분하고 정확한 답변
max_tokens=1024, # 출력 최대 토큰 수
seed=3379 # 결과의 일관성을 유지하기 위한 난수 고정
)
print(response.choices[0].message.content)

import openai
import os

# 1. API 키 설정 (환경변수 방식 권장)
os.environ['OPENAI_API_KEY'] = "YOUR_API_KEY"
client = openai.OpenAI()

# 2. 질문과 역할 정의
messages = [
{"role": "system", "content": "너는 어려운 IT 개념을 초등학생도 알기 쉽게 설명해주는 친절한 선생님이야."},
{"role": "user", "content": "RAG(검색 증강 생성)가 뭐야?"}
]

# 3. 모델 호출 및 결과 출력
response = client.chat.completions.create(
model="gpt-4o",
messages=messages,
temperature=0.3, # 0에 가까울수록 차분하고 정확한 답변
max_tokens=1024, # 출력 최대 토큰 수
seed=3379 # 결과의 일관성을 유지하기 위한 난수 고정
)
print(response.choices[0].message.content)

import tiktoken

tokenizer_4o = tiktoken.encoding_for_model("gpt-4o")
prompt = "GPT 모델별 토크나이저를 확인하고, 프롬프트에 사용된 토큰의 개수를 구할 수 있다."
tokens = tokenizer_4o.encode(prompt)

print("총 글자 수:", len(prompt))
print("총 토큰 수:", len(tokens))  # gpt-4o (o200k_base) 기준 24개 토큰

import tiktoken

tokenizer_4o = tiktoken.encoding_for_model("gpt-4o")
prompt = "GPT 모델별 토크나이저를 확인하고, 프롬프트에 사용된 토큰의 개수를 구할 수 있다."
tokens = tokenizer_4o.encode(prompt)

print("총 글자 수:", len(prompt))
print("총 토큰 수:", len(tokens))  # gpt-4o (o200k_base) 기준 24개 토큰

import os
from typing import List
import openai
from pydantic import BaseModel, Field

client = openai.OpenAI()

# 1. 데이터 규격 정의
class OrderItem(BaseModel):
    메뉴명: str = Field(description="주문 메뉴 이름")
    수량: int = Field(description="주문 수량")
    요청사항: str = Field(description="세부 요청사항 (없으면 '없음')")

class Order(BaseModel):
    주문목록: List[OrderItem]

# 2. 자연어 주문 메시지
user_message = "치즈버거 2개 갖다주시고요, 피클은 빼주세요. 감자튀김은 1개인데 케첩 많이 부탁드려요!"

# 3. 구조화 파싱 호출 (.parse)
completion = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "손님의 주문 텍스트에서 주문 내역을 품목별로 정확하게 추출하세요."},
        {"role": "user", "content": user_message}
    ],
    response_format=Order
)

order_result = completion.choices[0].message.parsed

# 4. 구조화된 데이터 확인
for item in order_result.주문목록:
    print(f"메뉴: {item.메뉴명} | 수량: {item.수량} | 요청: {item.요청사항}")
    
    
#  실행결과
#  메뉴: 치즈버거 | 수량: 2 | 요청: 피클은 빼주세요
#  메뉴: 감자튀김 | 수량: 1 | 요청: 케첩 많이 부탁드려요  
    

import os
from typing import List
import openai
from pydantic import BaseModel, Field

client = openai.OpenAI()

# 1. 데이터 규격 정의
class OrderItem(BaseModel):
    메뉴명: str = Field(description="주문 메뉴 이름")
    수량: int = Field(description="주문 수량")
    요청사항: str = Field(description="세부 요청사항 (없으면 '없음')")

class Order(BaseModel):
    주문목록: List[OrderItem]

# 2. 자연어 주문 메시지
user_message = "치즈버거 2개 갖다주시고요, 피클은 빼주세요. 감자튀김은 1개인데 케첩 많이 부탁드려요!"

# 3. 구조화 파싱 호출 (.parse)
completion = client.beta.chat.completions.parse(
    model="gpt-4o-2024-08-06",
    messages=[
        {"role": "system", "content": "손님의 주문 텍스트에서 주문 내역을 품목별로 정확하게 추출하세요."},
        {"role": "user", "content": user_message}
    ],
    response_format=Order
)

order_result = completion.choices[0].message.parsed

# 4. 구조화된 데이터 확인
for item in order_result.주문목록:
    print(f"메뉴: {item.메뉴명} | 수량: {item.수량} | 요청: {item.요청사항}")
    
    
#  실행결과
#  메뉴: 치즈버거 | 수량: 2 | 요청: 피클은 빼주세요
#  메뉴: 감자튀김 | 수량: 1 | 요청: 케첩 많이 부탁드려요  
    

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 0. API 키 설정 (환경 변수 또는 getpass 활용)
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR-API-KEY")

# 1. 시스템 프롬프트 (CoT 구조: 판단 이유를 먼저 적고 최종 분류를 내리도록 유도)
rule_prompt = """
당신은 스마트폰의 스팸 문자 판별기입니다.
다음 판별 기준에 따라 문맥을 면밀히 분석하세요:
- 단순히 '돈', '선물' 등의 단어가 포함되었다고 무조건 스팸으로 간주하지 않습니다.
- 지인이나 가족 간의 일상적 안부/대화라면 '일반'입니다.
- 출처가 불분명한 URL 링크 클릭을 유도하거나, 무료 혜택·대출을 미끼로 개인정보를 요구하면 '스팸'입니다.

반드시 아래 형식에 맞춰 답변하세요:
이유: (왜 스팸인지 혹은 일반인지 문맥을 분석한 핵심 근거 1줄)
분류: (스팸 또는 일반 중 택1)
"""

# 2. 질문 템플릿 정의
prompt = ChatPromptTemplate.from_messages([
    ("system", rule_prompt),
    ("user", "{message}")
])

# 3. LLM 정의 (분류/판단 작업이므로 일관성을 위해 temperature=0 설정)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 4. LCEL(쇠 파이프 |) 체인 조립
spam_filter_chain = prompt | llm | StrOutputParser()

# 5. 테스트 실행
test_msg = "[Web발신] 고객님! 신청하신 해외결제 85만원이 승인되었습니다. 본인이 아니면 즉시 클릭: http://bit.ly/fake"
result = spam_filter_chain.invoke({"message": test_msg})

print(result)


#  실행 결과
#  이유: 승인 확인을 빌미로 출처가 불분명한 단축 URL(http://bit.ly/fake) 클릭을 유도하고 있습니다.
#  분류: 스팸

import os
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 0. API 키 설정 (환경 변수 또는 getpass 활용)
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR-API-KEY")

# 1. 시스템 프롬프트 (CoT 구조: 판단 이유를 먼저 적고 최종 분류를 내리도록 유도)
rule_prompt = """
당신은 스마트폰의 스팸 문자 판별기입니다.
다음 판별 기준에 따라 문맥을 면밀히 분석하세요:
- 단순히 '돈', '선물' 등의 단어가 포함되었다고 무조건 스팸으로 간주하지 않습니다.
- 지인이나 가족 간의 일상적 안부/대화라면 '일반'입니다.
- 출처가 불분명한 URL 링크 클릭을 유도하거나, 무료 혜택·대출을 미끼로 개인정보를 요구하면 '스팸'입니다.

반드시 아래 형식에 맞춰 답변하세요:
이유: (왜 스팸인지 혹은 일반인지 문맥을 분석한 핵심 근거 1줄)
분류: (스팸 또는 일반 중 택1)
"""

# 2. 질문 템플릿 정의
prompt = ChatPromptTemplate.from_messages([
    ("system", rule_prompt),
    ("user", "{message}")
])

# 3. LLM 정의 (분류/판단 작업이므로 일관성을 위해 temperature=0 설정)
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 4. LCEL(쇠 파이프 |) 체인 조립
spam_filter_chain = prompt | llm | StrOutputParser()

# 5. 테스트 실행
test_msg = "[Web발신] 고객님! 신청하신 해외결제 85만원이 승인되었습니다. 본인이 아니면 즉시 클릭: http://bit.ly/fake"
result = spam_filter_chain.invoke({"message": test_msg})

print(result)


#  실행 결과
#  이유: 승인 확인을 빌미로 출처가 불분명한 단축 URL(http://bit.ly/fake) 클릭을 유도하고 있습니다.
#  분류: 스팸

pip install langchain langchain-openai langchain-community pymupdf faiss-cpu

pip install langchain langchain-openai langchain-community pymupdf faiss-cpu

import os
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 0. API 키 설정
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR-API-KEY")

# 1단계: 사내 규정 PDF 문서 로드 (PyMuPDFLoader)
# 실행 폴더에 위치한 규정 문서를 불러옵니다.
loader = PyMuPDFLoader("company_welfare_rules.pdf")
docs = loader.load()

# 2단계: 문서 분할 (RecursiveCharacterTextSplitter)
# 500자 단위로 자르고 맥락 단절을 막기 위해 50자씩 겹치도록 설정합니다.
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", " ", ""]
)
splits = text_splitter.split_documents(docs)

# 3단계: 임베딩 생성 및 FAISS 벡터 데이터베이스 적재
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_db = FAISS.from_documents(splits, embeddings)

# 질문과 가장 관련된 규정 문서 2개를 찾아오는 검색기(Retriever) 생성
retriever = vector_db.as_retriever(search_kwargs={"k": 2})

# 4단계: AI에게 건넬 프롬프트 템플릿 정의
prompt = ChatPromptTemplate.from_template("""
당신은 사내 복지 규정을 안내하는 친절한 인사팀 비서입니다.
반드시 아래 [사내 규정]에 적힌 내용만을 근거로 정확하게 답변하세요.
문서에 명시되지 않은 내용은 임의로 추측하지 말고 솔직하게 "해당 규정은 문서에 명시되어 있지 않습니다."라고 답하세요.

[사내 규정]:
{context}

질문: {question}
""")

# 검색된 다중 문서 객체(Document)의 텍스트를 하나로 묶어주는 포맷팅 함수
def format_docs(retrieved_docs):
    return "\n\n".join(doc.page_content for doc in retrieved_docs)

# 5단계: LCEL(쇠 파이프 |)로 공정 조립
# [문서 검색 + 원본 질문 유지] -> [프롬프트 완성] -> [LLM 추론] -> [텍스트 추출]
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

# 6단계: 질문 던져보기
question = "신입사원 첫해에 연차 며칠 쓸 수 있나요?"
answer = rag_chain.invoke(question)

print("질문:", question)
print("답변:", answer)


#  실행결과
#  질문: 신입사원 첫해에 연차 며칠 쓸 수 있나요?
#  답변: 사내 규정에 따르면, 신입사원의 연차 휴가는 입사 첫해 1년 미만 기간 동안 1개월 개근 시 1일씩 부여되어 최대 총 11일까지 사용할 수 있습니다.

import os
from langchain_community.document_loaders import PyMuPDFLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
from langchain_core.output_parsers import StrOutputParser

# 0. API 키 설정
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR-API-KEY")

# 1단계: 사내 규정 PDF 문서 로드 (PyMuPDFLoader)
# 실행 폴더에 위치한 규정 문서를 불러옵니다.
loader = PyMuPDFLoader("company_welfare_rules.pdf")
docs = loader.load()

# 2단계: 문서 분할 (RecursiveCharacterTextSplitter)
# 500자 단위로 자르고 맥락 단절을 막기 위해 50자씩 겹치도록 설정합니다.
text_splitter = RecursiveCharacterTextSplitter(
    chunk_size=500,
    chunk_overlap=50,
    separators=["\n\n", "\n", " ", ""]
)
splits = text_splitter.split_documents(docs)

# 3단계: 임베딩 생성 및 FAISS 벡터 데이터베이스 적재
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vector_db = FAISS.from_documents(splits, embeddings)

# 질문과 가장 관련된 규정 문서 2개를 찾아오는 검색기(Retriever) 생성
retriever = vector_db.as_retriever(search_kwargs={"k": 2})

# 4단계: AI에게 건넬 프롬프트 템플릿 정의
prompt = ChatPromptTemplate.from_template("""
당신은 사내 복지 규정을 안내하는 친절한 인사팀 비서입니다.
반드시 아래 [사내 규정]에 적힌 내용만을 근거로 정확하게 답변하세요.
문서에 명시되지 않은 내용은 임의로 추측하지 말고 솔직하게 "해당 규정은 문서에 명시되어 있지 않습니다."라고 답하세요.

[사내 규정]:
{context}

질문: {question}
""")

# 검색된 다중 문서 객체(Document)의 텍스트를 하나로 묶어주는 포맷팅 함수
def format_docs(retrieved_docs):
    return "\n\n".join(doc.page_content for doc in retrieved_docs)

# 5단계: LCEL(쇠 파이프 |)로 공정 조립
# [문서 검색 + 원본 질문 유지] -> [프롬프트 완성] -> [LLM 추론] -> [텍스트 추출]
rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

# 6단계: 질문 던져보기
question = "신입사원 첫해에 연차 며칠 쓸 수 있나요?"
answer = rag_chain.invoke(question)

print("질문:", question)
print("답변:", answer)


#  실행결과
#  질문: 신입사원 첫해에 연차 며칠 쓸 수 있나요?
#  답변: 사내 규정에 따르면, 신입사원의 연차 휴가는 입사 첫해 1년 미만 기간 동안 1개월 개근 시 1일씩 부여되어 최대 총 11일까지 사용할 수 있습니다.

import openai

client = openai.OpenAI()

# 신제품 홍보용 배너 이미지 생성 요청
response = client.images.generate(
    model="dall-e-3",
    prompt="모던한 대리석 테이블 위에 놓인 프리미엄 무선 헤드폰, 은은한 스튜디오 간접 조명, 고급 광고 화보 사진 스타일",
    size="1024x1024",
    quality="standard",
    n=1
)

# 1. 생성된 이미지 확인
image_url = response.data[0].url
print("생성된 이미지 URL:", image_url)

# 2. DALL-E 3가 전문가처럼 보강한 세부 프롬프트 확인
print("\n[AI가 보강한 프롬프트]:\n", response.data[0].revised_prompt)

import openai

client = openai.OpenAI()

# 신제품 홍보용 배너 이미지 생성 요청
response = client.images.generate(
    model="dall-e-3",
    prompt="모던한 대리석 테이블 위에 놓인 프리미엄 무선 헤드폰, 은은한 스튜디오 간접 조명, 고급 광고 화보 사진 스타일",
    size="1024x1024",
    quality="standard",
    n=1
)

# 1. 생성된 이미지 확인
image_url = response.data[0].url
print("생성된 이미지 URL:", image_url)

# 2. DALL-E 3가 전문가처럼 보강한 세부 프롬프트 확인
print("\n[AI가 보강한 프롬프트]:\n", response.data[0].revised_prompt)

import os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

# 0. API 키 설정
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR-API-KEY")

# 1. 에이전트가 사용할 도구(Tool) 정의
@tool
def calculate(a: int, b: int, operator: str) -> str:
    """간단한 사칙연산을 수행합니다.
    operator는 '+', '-', '*', '/' 중 하나여야 합니다."""
    if operator == '+': return f"계산 결과: {a + b}"
    elif operator == '-': return f"계산 결과: {a - b}"
    elif operator == '*': return f"계산 결과: {a * b}"
    elif operator == '/': return f"계산 결과: {a / b}" if b != 0 else "0으로 나눌 수 없습니다."
    return "지원하지 않는 연산자입니다."

@tool
def embed_pdf_to_faiss(file_path: str) -> str:
    """PDF 파일의 내용을 파싱하여 벡터 데이터베이스(FAISS)에 저장합니다."""
    # (내부 로직 생략: PyMuPDFLoader 및 FAISS 적재 파이프라인)
    return f"'{file_path}' 문서의 벡터 데이터베이스 저장이 완료되었습니다."

@tool
def crawl_naver_news_and_embed(search_term: str) -> str:
    """지정한 검색어로 네이버 뉴스를 크롤링하여 벡터 DB에 실시간 저장합니다."""
    # (내부 로직 생략: 네이버 뉴스 API 호출 및 FAISS 저장)
    return f"'{search_term}' 관련 최신 뉴스 수집 및 임베딩 저장이 완료되었습니다."

tools = [calculate, embed_pdf_to_faiss, crawl_naver_news_and_embed]

# 2. 두뇌(LLM) 준비 및 LangGraph ReAct 에이전트 초기화
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent_executor = create_react_agent(llm, tools)

# 3. 자연어로 복합 명령 내리기
# query = "사내 규정 PDF 문서 좀 디비에 저장해줘"
# query = "25 곱하기 14가 뭐야?"
query = "GPU라는 키워드로 네이버 뉴스를 검색해서 트렌드를 저장해줘."
inputs = {"messages": [("user", query)]}

# 4. 에이전트 실행 및 단계별 판단 흐름 확인
for step in agent_executor.stream(inputs, stream_mode="values"):
    latest_message = step["messages"][-1]
    latest_message.pretty_print()

import os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langgraph.prebuilt import create_react_agent

# 0. API 키 설정
os.environ["OPENAI_API_KEY"] = os.getenv("OPENAI_API_KEY", "YOUR-API-KEY")

# 1. 에이전트가 사용할 도구(Tool) 정의
@tool
def calculate(a: int, b: int, operator: str) -> str:
    """간단한 사칙연산을 수행합니다.
    operator는 '+', '-', '*', '/' 중 하나여야 합니다."""
    if operator == '+': return f"계산 결과: {a + b}"
    elif operator == '-': return f"계산 결과: {a - b}"
    elif operator == '*': return f"계산 결과: {a * b}"
    elif operator == '/': return f"계산 결과: {a / b}" if b != 0 else "0으로 나눌 수 없습니다."
    return "지원하지 않는 연산자입니다."

@tool
def embed_pdf_to_faiss(file_path: str) -> str:
    """PDF 파일의 내용을 파싱하여 벡터 데이터베이스(FAISS)에 저장합니다."""
    # (내부 로직 생략: PyMuPDFLoader 및 FAISS 적재 파이프라인)
    return f"'{file_path}' 문서의 벡터 데이터베이스 저장이 완료되었습니다."

@tool
def crawl_naver_news_and_embed(search_term: str) -> str:
    """지정한 검색어로 네이버 뉴스를 크롤링하여 벡터 DB에 실시간 저장합니다."""
    # (내부 로직 생략: 네이버 뉴스 API 호출 및 FAISS 저장)
    return f"'{search_term}' 관련 최신 뉴스 수집 및 임베딩 저장이 완료되었습니다."

tools = [calculate, embed_pdf_to_faiss, crawl_naver_news_and_embed]

# 2. 두뇌(LLM) 준비 및 LangGraph ReAct 에이전트 초기화
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
agent_executor = create_react_agent(llm, tools)

# 3. 자연어로 복합 명령 내리기
# query = "사내 규정 PDF 문서 좀 디비에 저장해줘"
# query = "25 곱하기 14가 뭐야?"
query = "GPU라는 키워드로 네이버 뉴스를 검색해서 트렌드를 저장해줘."
inputs = {"messages": [("user", query)]}

# 4. 에이전트 실행 및 단계별 판단 흐름 확인
for step in agent_executor.stream(inputs, stream_mode="values"):
    latest_message = step["messages"][-1]
    latest_message.pretty_print()