LC LangChain 완전 가이드

from operator import itemgetter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableParallel, RunnablePassthrough, RunnableLambda
from langchain_openai import ChatOpenAI

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)
parser = StrOutputParser()

# ── 1. 기본 파이프 조합 ──────────────────────────────────────────
prompt = ChatPromptTemplate.from_template("{topic}를 한 문장으로 설명해줘.")
basic_chain = prompt | model | parser

# ── 2. RunnableParallel: 두 체인 동시 실행 ───────────────────────
summary_prompt   = ChatPromptTemplate.from_template("{text}를 3줄로 요약해줘.")
sentiment_prompt = ChatPromptTemplate.from_template("{text}의 감성을 분석해줘.")

parallel_chain = RunnableParallel(
    summary=summary_prompt | model | parser,
    sentiment=sentiment_prompt | model | parser,
)

# ── 3. RunnablePassthrough + assign 패턴 ────────────────────────
# 원본 입력을 유지하면서 새 키를 추가
augmented_chain = RunnablePassthrough.assign(
    summary=itemgetter("text") | summary_prompt | model | parser
)

# ── 4. RunnableLambda: 커스텀 전처리 ────────────────────────────
def preprocess(inputs: dict) -> dict:
    """입력 텍스트를 정규화"""
    return {"text": inputs["text"].strip().lower()}

full_chain = (
    RunnableLambda(preprocess)
    | RunnableParallel(
        summary=summary_prompt | model | parser,
        sentiment=sentiment_prompt | model | parser,
    )
)

# ── 5. 실행 ──────────────────────────────────────────────────────
if __name__ == "__main__":
    print(basic_chain.invoke({"topic": "LCEL"}))

    result = parallel_chain.invoke({"text": "LangChain은 AI 개발 프레임워크입니다."})
    print("요약:", result["summary"])
    print("감성:", result["sentiment"])

    # 배치 실행 (여러 입력을 한 번에)
    topics = [{"topic": "RAG"}, {"topic": "Agent"}, {"topic": "Embedding"}]
    for r in basic_chain.batch(topics):
        print(r)

import json
import sqlite3
from typing import Optional
from pydantic import BaseModel, Field
from langchain_core.tools import tool, StructuredTool

# ── 1. @tool 데코레이터: 가장 간단한 방식 ───────────────────────
@tool
def web_search(query: str) -> str:
    """인터넷에서 최신 정보를 검색합니다. 실시간 데이터나 최신 뉴스가 필요할 때 사용하세요."""
    # 실제 구현: from tavily import TavilyClient 등 사용
    return f"'{query}'에 대한 검색 결과: [검색 API 응답]"

@tool
def python_repl(code: str) -> str:
    """Python 코드를 실행하고 결과를 반환합니다. 계산, 데이터 처리, 코드 검증에 사용하세요."""
    import io, contextlib
    buf = io.StringIO()
    try:
        with contextlib.redirect_stdout(buf):
            exec(code, {})  # 프로덕션에서는 샌드박스 환경 사용
        return buf.getvalue() or "실행 완료 (출력 없음)"
    except Exception as e:
        return f"오류: {e}"

@tool
def read_file(path: str) -> str:
    """로컬 파일을 읽어 내용을 반환합니다. 문서 분석이나 설정 파일 확인에 사용하세요."""
    try:
        with open(path, "r", encoding="utf-8") as f:
            return f.read()
    except FileNotFoundError:
        return f"파일을 찾을 수 없습니다: {path}"

# ── 2. StructuredTool: Pydantic 스키마로 입력 검증 ───────────────
class DBQueryInput(BaseModel):
    sql: str = Field(description="실행할 SELECT SQL 쿼리")
    db_path: str = Field(default="./data.db", description="SQLite 파일 경로")
    limit: Optional[int] = Field(default=10, description="최대 반환 행 수")

def run_db_query(sql: str, db_path: str = "./data.db", limit: int = 10) -> str:
    """SQLite 데이터베이스에 SELECT 쿼리를 실행합니다."""
    if not sql.strip().upper().startswith("SELECT"):
        return "오류: SELECT 쿼리만 허용됩니다."
    try:
        conn = sqlite3.connect(db_path)
        cursor = conn.execute(sql)
        rows = cursor.fetchmany(limit)
        columns = [d[0] for d in cursor.description]
        conn.close()
        return json.dumps({"columns": columns, "rows": rows}, ensure_ascii=False)
    except Exception as e:
        return f"DB 오류: {e}"

db_query_tool = StructuredTool.from_function(
    func=run_db_query,
    name="db_query",
    description="데이터베이스에서 정보를 조회합니다. 통계, 기록 검색에 사용하세요.",
    args_schema=DBQueryInput,
)

# ── 3. 도구 목록 확인 ────────────────────────────────────────────
tools = [web_search, python_repl, read_file, db_query_tool]
for t in tools:
    print(f"[{t.name}] {t.description[:60]}...")

import os
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.messages import HumanMessage, AIMessage
from langchain import hub
from langchain.agents import create_react_agent, AgentExecutor

os.environ["OPENAI_API_KEY"] = "sk-..."

# ── 도구 정의 ────────────────────────────────────────────────────
@tool
def calculator(expression: str) -> str:
    """수학 계산을 수행합니다. 예: '2 ** 10', 'sum([1,2,3])'"""
    try:
        return str(eval(expression, {"__builtins__": {}}, {}))
    except Exception as e:
        return f"계산 오류: {e}"

@tool
def web_search(query: str) -> str:
    """최신 정보를 검색합니다. 실시간 데이터나 날씨, 뉴스가 필요할 때 사용하세요."""
    return f"검색 결과: '{query}'에 대한 최신 정보입니다."

@tool
def get_weather(city: str) -> str:
    """특정 도시의 현재 날씨를 조회합니다."""
    return f"{city}의 현재 날씨: 맑음, 22°C, 습도 55%"

tools = [calculator, web_search, get_weather]

# ── ReAct 에이전트 생성 ──────────────────────────────────────────
model = ChatOpenAI(model="gpt-4o", temperature=0)
react_prompt = hub.pull("hwchase17/react")  # Thought/Action/Observation 형식 강제

agent = create_react_agent(llm=model, tools=tools, prompt=react_prompt)
agent_executor = AgentExecutor(
    agent=agent, tools=tools,
    verbose=True,          # Thought/Action/Observation 로그 출력
    max_iterations=10,     # 무한 루프 방지
    handle_parsing_errors=True,
)

# ── 실행 예시 ────────────────────────────────────────────────────
if __name__ == "__main__":
    # 멀티 스텝 추론: 날씨 조회 + 단위 변환
    result = agent_executor.invoke({
        "input": "서울의 현재 날씨를 알려주고, 기온을 화씨로 변환해줘."
    })
    print("\n최종 답변:", result["output"])

    # 대화형 에이전트 (히스토리 포함)
    chat_history = []
    queries = ["2의 32제곱은 얼마야?", "그 숫자가 몇 GB인지 계산해줘."]

    for query in queries:
        result = agent_executor.invoke({"input": query, "chat_history": chat_history})
        chat_history.extend([
            HumanMessage(content=query),
            AIMessage(content=result["output"]),
        ])
        print(f"Q: {query}\nA: {result['output']}\n")

from typing import Optional, List
from typing_extensions import TypedDict
from pydantic import BaseModel, Field
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import PydanticOutputParser

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 1. Pydantic 모델로 구조화된 출력 ────────────────────────────
class Product(BaseModel):
    name: str = Field(description="제품명")
    price: float = Field(description="가격 (원)")
    category: str = Field(description="카테고리")
    in_stock: bool = Field(description="재고 여부")
    tags: List[str] = Field(default_factory=list, description="태그 목록")
    description: Optional[str] = Field(default=None, description="제품 설명")

structured_model = model.with_structured_output(Product)
prompt = ChatPromptTemplate.from_template("다음 텍스트에서 제품 정보를 추출하세요:\n\n{text}")
extraction_chain = prompt | structured_model

product = extraction_chain.invoke({
    "text": "신형 무선 이어폰 AX-100, 가격 89,000원, 재고 있음. 노이즈 캔슬링, 블루투스"
})
print(f"제품: {product.name}, 가격: {product.price:,}원")
print(f"태그: {product.tags}")

# ── 2. TypedDict 스키마 ──────────────────────────────────────────
class AnalysisResult(TypedDict):
    sentiment: str       # positive / negative / neutral
    score: float         # 0.0 ~ 1.0
    keywords: List[str]  # 핵심 키워드
    summary: str         # 요약

typed_model = model.with_structured_output(AnalysisResult)
analysis_chain = ChatPromptTemplate.from_template("다음 리뷰를 분석하세요:\n{review}") | typed_model

result = analysis_chain.invoke({"review": "배송이 빠르고 제품 품질이 좋습니다. 가격 대비 만족도가 높아요!"})
print(f"감성: {result['sentiment']}, 점수: {result['score']}")

# ── 3. PydanticOutputParser (function calling 미지원 모델 대응) ──
class ResearchReport(BaseModel):
    title: str
    key_findings: List[str]
    conclusion: str
    confidence: float = Field(ge=0, le=1)

pydantic_parser = PydanticOutputParser(pydantic_object=ResearchReport)
format_instructions = pydantic_parser.get_format_instructions()

report_prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 연구 보고서 작성 전문가입니다.\n{format_instructions}"),
    ("user", "{topic}에 대한 연구 보고서를 작성하세요."),
])
report = (report_prompt | model | pydantic_parser).invoke({
    "topic": "대형언어모델(LLM)의 기업 도입 현황",
    "format_instructions": format_instructions,
})
print(f"보고서: {report.title}")
print(f"핵심 발견: {report.key_findings}")

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory
from langchain.memory import ConversationBufferWindowMemory, VectorStoreRetrieverMemory
from langchain_community.vectorstores import Chroma

model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# ── 1. ConversationBufferWindowMemory: 최근 N턴만 유지 ──────────
window_memory = ConversationBufferWindowMemory(
    k=5,                   # 최근 5턴만 유지
    return_messages=True,  # Message 객체로 반환
    memory_key="chat_history",
)

# ── 2. VectorStoreRetrieverMemory: 관련 기억만 검색 ─────────────
vectorstore = Chroma(
    collection_name="agent_memory",
    embedding_function=OpenAIEmbeddings(),
    persist_directory="./chroma_memory",
)
vector_memory = VectorStoreRetrieverMemory(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
    memory_key="relevant_history",
    return_messages=False,
)
vector_memory.save_context({"input": "내 이름은 김민수입니다."}, {"output": "안녕하세요!"})
vector_memory.save_context({"input": "나는 파이썬 개발자입니다."}, {"output": "파이썬 개발자시군요!"})

# ── 3. RunnableWithMessageHistory (권장 패턴) ────────────────────
session_store: dict = {}

def get_session_history(session_id: str) -> ChatMessageHistory:
    """세션 ID로 대화 기록을 가져옵니다. 없으면 새로 생성합니다."""
    if session_id not in session_store:
        session_store[session_id] = ChatMessageHistory()
    return session_store[session_id]

chat_prompt = ChatPromptTemplate.from_messages([
    ("system", "당신은 친절한 AI 어시스턴트입니다. 이전 대화를 기억하세요."),
    MessagesPlaceholder(variable_name="history"),  # 히스토리 삽입 위치
    ("human", "{input}"),
])

chain_with_history = RunnableWithMessageHistory(
    chat_prompt | model,
    get_session_history,
    input_messages_key="input",
    history_messages_key="history",
)

# ── 실행 예시 ────────────────────────────────────────────────────
if __name__ == "__main__":
    config = {"configurable": {"session_id": "user_001"}}

    r1 = chain_with_history.invoke({"input": "안녕! 나는 서울에 살아."}, config=config)
    print("AI:", r1.content)

    # 이전 맥락을 기억하는지 확인
    r2 = chain_with_history.invoke({"input": "내가 어디 산다고 했지?"}, config=config)
    print("AI:", r2.content)  # "서울에 산다고 하셨습니다."

    history = get_session_history("user_001")
    print(f"\n대화 기록 ({len(history.messages)}개 메시지):")
    for msg in history.messages:
        role = "사용자" if msg.type == "human" else "AI"
        print(f"  [{role}] {msg.content[:50]}...")

import json
from typing import Literal
from langchain_openai import ChatOpenAI
from langchain_core.tools import tool
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.agents import create_react_agent, AgentExecutor
from langchain import hub

model = ChatOpenAI(model="gpt-4o", temperature=0)
react_prompt = hub.pull("hwchase17/react")

# ── Worker 도구 정의 ──────────────────────────────────────────────
@tool
def analyze_data(data: str) -> str:
    """데이터를 분석하고 통계 및 인사이트를 제공합니다."""
    return f"분석 결과: {data}에서 평균 85.3, 표준편차 12.1을 발견했습니다."

@tool
def write_report(topic: str, findings: str) -> str:
    """분석 결과를 바탕으로 보고서를 작성합니다."""
    return f"# {topic} 보고서\n\n## 주요 발견\n{findings}\n\n## 결론\n분석 완료."

@tool
def search_web(query: str) -> str:
    """최신 정보를 웹에서 검색합니다."""
    return f"'{query}' 검색 결과: 관련 최신 정보 수집 완료."

@tool
def execute_code(code: str) -> str:
    """Python 코드를 실행합니다."""
    import io, contextlib
    buf = io.StringIO()
    with contextlib.redirect_stdout(buf):
        exec(code, {})
    return buf.getvalue() or "실행 완료"

# ── 전문화된 Worker 에이전트 ─────────────────────────────────────
def make_executor(tools_list: list) -> AgentExecutor:
    agent = create_react_agent(llm=model, tools=tools_list, prompt=react_prompt)
    return AgentExecutor(
        agent=agent, tools=tools_list,
        verbose=True, max_iterations=5, handle_parsing_errors=True,
    )

data_analyst  = make_executor([analyze_data, execute_code])
report_writer = make_executor([write_report, search_web])

# ── Supervisor: 라우팅 로직 ──────────────────────────────────────
supervisor_prompt = ChatPromptTemplate.from_messages([
    ("system", """당신은 AI 팀 매니저입니다. 사용자 요청을 분석해 적절한 담당자에게 라우팅합니다.

담당자:
- data_analyst: 데이터 분석, 계산, 코드 실행이 필요한 작업
- report_writer: 보고서 작성, 정보 검색이 필요한 작업
- finish: 작업 완료 또는 직접 답변 가능한 경우

JSON만 반환하세요: {{\"next\": \"data_analyst\" | \"report_writer\" | \"finish\", \"reason\": \"이유\"}}"""),
    ("human", "작업: {task}\n이전 결과: {previous_result}"),
])
supervisor_chain = supervisor_prompt | model | StrOutputParser()

def supervisor_route(task: str) -> Literal["data_analyst", "report_writer", "finish"]:
    result = supervisor_chain.invoke({"task": task, "previous_result": ""})
    try:
        return json.loads(result).get("next", "finish")
    except Exception:
        return "finish"

# ── 멀티 에이전트 오케스트레이터 ────────────────────────────────
def run_multi_agent(task: str) -> str:
    print(f"\n{'='*60}\n작업: {task}\n{'='*60}")
    results = []
    for step in range(5):
        next_agent = supervisor_route(task)
        print(f"\n[Step {step + 1}] Supervisor → {next_agent}")
        if next_agent == "finish":
            break
        elif next_agent == "data_analyst":
            r = data_analyst.invoke({"input": task})
            results.append(f"데이터 분석: {r['output']}")
        elif next_agent == "report_writer":
            r = report_writer.invoke({"input": task})
            results.append(f"보고서: {r['output']}")
    return "\n".join(results)

if __name__ == "__main__":
    print(run_multi_agent("지난 분기 매출 데이터를 분석하고 경영진 보고서를 작성해줘."))

import asyncio
from fastapi import FastAPI
from fastapi.responses import StreamingResponse
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

# ── 1. 터미널 스트리밍 ───────────────────────────────────────────
def streaming_to_terminal():
    """토큰이 생성될 때마다 터미널에 즉시 출력"""
    m = ChatOpenAI(model="gpt-4o-mini", streaming=True,
                   callbacks=[StreamingStdOutCallbackHandler()])
    m.invoke("AI Agent의 미래를 200자로 설명해줘.")

# ── 2. astream() — 비동기 스트리밍 ──────────────────────────────
async def async_streaming():
    model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
    chain = ChatPromptTemplate.from_template("{topic}에 대해 설명해줘.") | model | StrOutputParser()
    full = ""
    async for chunk in chain.astream({"topic": "양자 컴퓨팅"}):
        print(chunk, end="", flush=True)
        full += chunk
    print(f"\n\n총 {len(full)}자 생성 완료")

# ── 3. FastAPI + SSE 스트리밍 서버 ───────────────────────────────
app = FastAPI(title="LangChain 스트리밍 API")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
chain = (
    ChatPromptTemplate.from_template("시스템: 전문 AI 어시스턴트\n\n사용자: {question}")
    | model | StrOutputParser()
)

async def token_generator(question: str):
    """SSE 형식으로 토큰을 스트리밍합니다."""
    try:
        async for chunk in chain.astream({"question": question}):
            yield f"data: {chunk}\n\n"
        yield "data: [DONE]\n\n"
    except Exception as e:
        yield f"data: [ERROR] {e}\n\n"

@app.get("/stream")
async def stream_response(question: str = "AI란 무엇인가요?"):
    """SSE로 LLM 응답을 스트리밍합니다."""
    return StreamingResponse(
        token_generator(question),
        media_type="text/event-stream",
        headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"},
    )

@app.get("/stream/json")
async def stream_json(question: str = "AI란 무엇인가요?"):
    """JSON Lines 형식으로 스트리밍합니다."""
    import json

    async def gen():
        async for chunk in chain.astream({"question": question}):
            yield json.dumps({"delta": chunk}, ensure_ascii=False) + "\n"

    return StreamingResponse(gen(), media_type="application/x-ndjson")

if __name__ == "__main__":
    import uvicorn
    # asyncio.run(async_streaming())
    uvicorn.run(app, host="0.0.0.0", port=8000)
    # 접속: http://localhost:8000/stream?question=RAG를 설명해줘

from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.vectorstores import Chroma
from langchain_community.retrievers import BM25Retriever
from langchain.retrievers import EnsembleRetriever, MultiQueryRetriever
from langchain.retrievers.document_compressors import CohereRerank
from langchain.retrievers import ContextualCompressionRetriever
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_community.document_loaders import TextLoader

# ── 문서 로드 및 청킹 ────────────────────────────────────────────
documents = TextLoader("knowledge_base.txt", encoding="utf-8").load()
chunks = RecursiveCharacterTextSplitter(
    chunk_size=500, chunk_overlap=50,
    separators=["\n\n", "\n", ".", " "],
).split_documents(documents)

embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 1. Hybrid Search: BM25 + Vector 앙상블 ──────────────────────
bm25_retriever = BM25Retriever.from_documents(chunks)
bm25_retriever.k = 5

vectorstore = Chroma.from_documents(chunks, embeddings)
vector_retriever = vectorstore.as_retriever(search_kwargs={"k": 5})

# BM25 40% + Vector 60% 가중치 결합
ensemble_retriever = EnsembleRetriever(
    retrievers=[bm25_retriever, vector_retriever],
    weights=[0.4, 0.6],
)

# ── 2. Cohere Reranker: 검색 결과 재정렬 ────────────────────────
# pip install cohere
reranking_retriever = ContextualCompressionRetriever(
    base_compressor=CohereRerank(model="rerank-multilingual-v3.0", top_n=3),
    base_retriever=ensemble_retriever,
)

# ── 3. HyDE: 가상 문서 임베딩 ────────────────────────────────────
hyde_chain = (
    ChatPromptTemplate.from_template(
        "다음 질문에 대한 가상의 답변 문서를 작성하세요:\n\n질문: {question}\n\n문서:"
    )
    | model | StrOutputParser()
    | (lambda text: vectorstore.similarity_search(text, k=4))
)

# ── 4. Multi-query Retriever ─────────────────────────────────────
# LLM이 원본 질문에서 3개의 다른 질문을 생성해 더 넓게 검색
multi_query_retriever = MultiQueryRetriever.from_llm(
    retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),
    llm=model,
)

# ── 5. 완성된 고급 RAG 체인 ──────────────────────────────────────
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """정확한 정보를 제공하는 AI입니다.
주어진 컨텍스트만 사용해 답변하세요. 컨텍스트에 없는 내용은 모른다고 말하세요.

컨텍스트:
{context}"""),
    ("human", "{question}"),
])

def format_docs(docs) -> str:
    return "\n\n---\n\n".join(
        f"[출처: {doc.metadata.get('source', 'unknown')}]\n{doc.page_content}"
        for doc in docs
    )

# Hybrid + Rerank + RAG 완전 파이프라인
advanced_rag_chain = (
    {"context": reranking_retriever | format_docs, "question": RunnablePassthrough()}
    | rag_prompt | model | StrOutputParser()
)

if __name__ == "__main__":
    question = "LangChain에서 메모리를 관리하는 방법은?"
    print("답변:", advanced_rag_chain.invoke(question))

    import logging
    logging.getLogger("langchain.retrievers.multi_query").setLevel(logging.INFO)
    docs = multi_query_retriever.invoke(question)
    print(f"\nMulti-query 검색 문서 수: {len(docs)}")

import os
from langchain_openai import ChatOpenAI, OpenAIEmbeddings
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import JsonOutputParser

# ── 1. LangSmith 트레이싱 설정 ───────────────────────────────────
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__your_api_key"
os.environ["LANGCHAIN_PROJECT"] = "production-rag-v2"
# 설정 후 모든 체인 실행이 자동으로 LangSmith에 기록됨

model = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# ── 2. RAGAS로 RAG 품질 평가 ─────────────────────────────────────
# pip install ragas
from ragas import evaluate
from ragas.metrics import (
    faithfulness,      # 답변이 컨텍스트에 충실한지 (환각 방지)
    answer_relevancy,  # 답변이 질문과 관련 있는지
    context_recall,    # 필요한 정보가 컨텍스트에 포함됐는지
    context_precision, # 컨텍스트가 관련 정보만 담고 있는지
)
from datasets import Dataset

# 도메인 전문가가 검수한 QA 쌍으로 평가 데이터셋 구성
eval_dataset = Dataset.from_dict({
    "question": [
        "LangChain의 LCEL이란 무엇인가요?",
        "ReAct 에이전트의 동작 방식을 설명해주세요.",
        "RAG와 Fine-tuning의 차이점은?",
    ],
    "answer": [
        "LCEL은 LangChain Expression Language로, | 연산자로 컴포넌트를 조합하는 방식입니다.",
        "ReAct는 Reasoning과 Acting을 결합한 패턴으로, Thought→Action→Observation을 반복합니다.",
        "RAG는 외부 지식을 검색해 응답하고, Fine-tuning은 모델 가중치를 직접 업데이트합니다.",
    ],
    "contexts": [
        ["LCEL(LangChain Expression Language)은 파이프 연산자(|)를 사용해 LLM 컴포넌트를 연결합니다."],
        ["ReAct 패턴은 Reasoning(추론)과 Acting(행동)을 결합해 에이전트가 단계적으로 문제를 해결합니다."],
        ["RAG는 Retrieval-Augmented Generation으로 외부 문서를 검색해 LLM 입력에 추가합니다."],
    ],
    "ground_truth": [
        "LCEL은 파이프 연산자로 컴포넌트를 조합하는 LangChain의 표현 언어입니다.",
        "ReAct는 Thought-Action-Observation 루프로 에이전트가 추론하고 행동합니다.",
        "RAG는 검색 기반 생성이고 Fine-tuning은 모델 파라미터를 학습시킵니다.",
    ],
})

results = evaluate(
    dataset=eval_dataset,
    metrics=[faithfulness, answer_relevancy, context_recall, context_precision],
    llm=model,
    embeddings=OpenAIEmbeddings(),
)

print("\n=== RAGAS 평가 결과 ===")
df = results.to_pandas()
print(df[["question", "faithfulness", "answer_relevancy", "context_recall"]].to_string())
print(f"\n평균 Faithfulness:     {df['faithfulness'].mean():.3f}")
print(f"평균 Answer Relevancy: {df['answer_relevancy'].mean():.3f}")
print(f"평균 Context Recall:   {df['context_recall'].mean():.3f}")

# ── 3. LLM-as-Judge 자체 평가 루프 ──────────────────────────────
eval_chain = (
    ChatPromptTemplate.from_messages([
        ("system", """AI 답변 품질 평가자입니다.
질문, 참조 답변, AI 답변을 비교해 0~10점으로 평가하고 JSON으로 반환하세요.
형식: {{\"score\": 숫자, \"reason\": \"이유\"}}"""),
        ("human", "질문: {question}\n참조 답변: {reference}\nAI 답변: {prediction}"),
    ])
    | model | JsonOutputParser()
)

def evaluate_answer(question: str, reference: str, prediction: str) -> dict:
    """답변 품질을 0-10 척도로 자동 평가합니다."""
    return eval_chain.invoke({
        "question": question, "reference": reference, "prediction": prediction,
    })

if __name__ == "__main__":
    score = evaluate_answer(
        question="LCEL이란?",
        reference="파이프 연산자로 컴포넌트를 조합하는 LangChain 표현 언어",
        prediction="LCEL은 LangChain의 파이프라인 구성 방식입니다.",
    )
    print(f"\n평가 점수: {score['score']}/10")
    print(f"평가 이유: {score['reason']}")