一句話理解

LCEL(LangChain Expression Language)用 | 把 Prompt → LLM → OutputParser 串成一條 Chain,像 Unix pipe 一樣直觀。


核心三元件

PromptTemplate → LLM → OutputParser
     |              |         |
   組出問題       問 AI     解析回答

基本 Chain

from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
 
# 1. 定義 Prompt Template
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一個專業的{role},請用繁體中文回答。"),
    ("human", "{question}")
])
 
# 2. 定義 LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
 
# 3. 定義 OutputParser
parser = StrOutputParser()
 
# 4. 用 | 串起來(LCEL)
chain = prompt | llm | parser
 
# 5. 執行
result = chain.invoke({
    "role": "後端工程師",
    "question": "什麼是 REST API?"
})
print(result)

PromptTemplate 種類

# 單純文字
from langchain_core.prompts import PromptTemplate
prompt = PromptTemplate.from_template("請翻譯:{text}")
 
# Chat(有 system/human/ai role)
from langchain_core.prompts import ChatPromptTemplate
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是翻譯專家"),
    ("human", "請把這段話翻成{lang}{text}")
])
 
# Few-shot(給範例讓 AI 學格式)
from langchain_core.prompts import FewShotChatMessagePromptTemplate
 
examples = [
    {"input": "happy", "output": "開心"},
    {"input": "sad", "output": "難過"},
]
example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}")
])
few_shot_prompt = FewShotChatMessagePromptTemplate(
    examples=examples,
    example_prompt=example_prompt
)

OutputParser 種類

from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.pydantic_v1 import BaseModel
 
# 純文字
parser = StrOutputParser()
 
# JSON → dict
parser = JsonOutputParser()
chain = prompt | llm | parser
result = chain.invoke({"text": "..."})  # 回傳 dict
 
# Pydantic(結構化輸出)
class ProductInfo(BaseModel):
    name: str
    price: float
    category: str
 
parser = JsonOutputParser(pydantic_object=ProductInfo)
# prompt 需包含 {format_instructions}
prompt = ChatPromptTemplate.from_template(
    "從以下文字擷取商品資訊:{text}\n{format_instructions}"
)
chain = prompt | llm | parser

RunnableParallel(並行執行)

from langchain_core.runnables import RunnableParallel
 
# 同時執行兩條 chain,結果合併
parallel = RunnableParallel(
    summary=summary_chain,
    keywords=keywords_chain
)
 
result = parallel.invoke({"text": "..."})
print(result["summary"])   # 摘要
print(result["keywords"])  # 關鍵字

RunnableLambda(插入自訂邏輯)

from langchain_core.runnables import RunnableLambda
 
# 在 chain 中插入一段 Python 函式
def preprocess(text):
    return text.strip().lower()
 
chain = RunnableLambda(preprocess) | prompt | llm | parser

Streaming(串流輸出)

# stream 逐 token 輸出,適合前端即時顯示
for chunk in chain.stream({"question": "介紹 Python"}):
    print(chunk, end="", flush=True)
 
# 非同步版本
async for chunk in chain.astream({"question": "介紹 Python"}):
    print(chunk, end="", flush=True)

常見錯誤

錯誤原因解法
KeyError in prompt變數名稱和 template 裡的不一樣確認 invoke({}) 的 key 名稱
JSON 解析失敗LLM 沒回傳合法 JSONformat_instructions,或用 with_structured_output
API 費用爆炸temperature=1 亂跑固定 temperature=0

相關筆記