LangChain 深度教學

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.runnables import RunnableParallel

# --- 基本 Chain：Prompt | LLM | Parser ---
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一個專業的技術翻譯師，請將以下英文翻譯成繁體中文。"),
    ("human", "{text}")
])

chain = prompt | llm | StrOutputParser()

# 單次呼叫
result = chain.invoke({"text": "AI Agents are autonomous programs."})
print(result)  # "AI Agent 是自主運作的程式。"

# 串流輸出（逐 token 印出，提升使用者體驗）
for chunk in chain.stream({"text": "Retrieval-Augmented Generation"}):
    print(chunk, end="", flush=True)

# 批次執行（並行呼叫，效率比 for 迴圈高約 3-5 倍）
results = chain.batch([
    {"text": "Machine learning"},
    {"text": "Neural network"},
    {"text": "Reinforcement learning"}
])

# 非同步執行
import asyncio
async def run():
    result = await chain.ainvoke({"text": "Deep learning"})
    return result

# --- 並行執行兩條 Chain（同時呼叫）---
parallel = RunnableParallel(
    translation=chain,
    summary=(ChatPromptTemplate.from_template("請用一句話摘要：{text}") | llm | StrOutputParser())
)
out = parallel.invoke({"text": "Large language models are transformer-based neural networks."})
print(out["translation"])
print(out["summary"])

# --- ConversationChain：帶記憶的對話 ---
from langchain.memory import ConversationSummaryBufferMemory
from langchain.chains import ConversationChain

memory = ConversationSummaryBufferMemory(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    max_token_limit=2000,  # 超過 2000 token 自動摘要
    return_messages=True
)
conv_chain = ConversationChain(
    llm=ChatOpenAI(model="gpt-4o-mini"),
    memory=memory,
    verbose=False
)
conv_chain.predict(input="我是一名後端工程師，想學 AI Agent 開發。")
resp = conv_chain.predict(input="根據我的背景，該從哪個框架入手？")
print(resp)

from langchain_community.document_loaders import WebBaseLoader, PyPDFLoader, DirectoryLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_community.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain_core.prompts import ChatPromptTemplate

# ===== 階段一：建立向量索引（離線執行）=====

# 1. 載入文件（支援 Web、PDF、Markdown、CSV、Notion、Google Drive 等）
loader = PyPDFLoader("./company_knowledge_base.pdf")
docs = loader.load()

# 2. 切分段落（chunk_overlap 確保語意不被截斷於段落邊界）
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,           # 每段最多 1000 字元
    chunk_overlap=200,         # 前後重疊 200 字元，保留跨段語意
    separators=["\n\n", "\n", "。", "，", " "]  # 優先在段落/句子邊界切分
)
chunks = splitter.split_documents(docs)
print(f"共切分為 {len(chunks)} 個段落")

# 3. 生成 Embedding 並存入向量資料庫
embeddings = OpenAIEmbeddings(model="text-embedding-3-small")
vectorstore = FAISS.from_documents(chunks, embeddings)
vectorstore.save_local("./faiss_index")   # 持久化儲存，避免重複建索引

# ===== 階段二：RAG 查詢（線上即時執行）=====

# 4. 載入已建好的索引
vectorstore = FAISS.load_local(
    "./faiss_index", embeddings,
    allow_dangerous_deserialization=True
)

# 5. 建立 Retriever（MMR 策略：兼顧相關性與多樣性，避免重複段落）
retriever = vectorstore.as_retriever(
    search_type="mmr",
    search_kwargs={"k": 4, "fetch_k": 20, "lambda_mult": 0.7}
)

# 6. 自訂 RAG Prompt（明確指示 LLM 只依據參考資料回答，不自行捏造）
rag_prompt = ChatPromptTemplate.from_messages([
    ("system", """你是一個專業的企業知識庫助理。請根據以下參考資料回答使用者問題。
回答時務必：
- 只使用參考資料中的資訊，不要加入訓練資料中的外部知識
- 如果資料中找不到答案，明確回答「根據現有資料無法回答此問題」
- 引用具體的資料來源（例如：「根據第 3 頁的說明…」）

參考資料：
{context}"""),
    ("human", "{question}")
])

# 7. 建立 RetrievalQA Chain（return_source_documents=True 可追溯來源）
rag_chain = RetrievalQA.from_chain_type(
    llm=ChatOpenAI(model="gpt-4o-mini", temperature=0),
    retriever=retriever,
    chain_type_kwargs={"prompt": rag_prompt},
    return_source_documents=True
)

# 8. 查詢並顯示答案與來源
result = rag_chain.invoke({"query": "公司的退款政策是什麼？"})
print("答案：", result["result"])
print("\n參考來源：")
for doc in result["source_documents"]:
    src = doc.metadata.get("source", "未知")
    page = doc.metadata.get("page", "?")
    print(f"  - {src}（第 {page} 頁）：{doc.page_content[:80]}…")

import os

# 設定 LangSmith 環境變數（只需這幾行，其餘程式碼完全不用修改）
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "ls__your_api_key_here"
os.environ["LANGCHAIN_PROJECT"] = "my-rag-project"   # 專案名稱，用於分組管理

# 之後所有 Chain / Agent 的呼叫都會自動記錄到 LangSmith Dashboard
# 前往 https://smith.langchain.com 即可查看完整執行 trace

# --- 手動建立 Run（為特定邏輯加上追蹤標記）---
from langsmith import traceable

@traceable(name="custom-rag-pipeline", tags=["production", "v2"])
def run_rag_query(question: str) -> str:
    result = rag_chain.invoke({"query": question})
    return result["result"]

answer = run_rag_query("LangChain 支援哪些向量資料庫？")