一句話理解
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 | parserRunnableParallel(並行執行)
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 | parserStreaming(串流輸出)
# 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 沒回傳合法 JSON | 加 format_instructions,或用 with_structured_output |
| API 費用爆炸 | temperature=1 亂跑 | 固定 temperature=0 |
相關筆記
- LangChain — 模組地圖
- LangChain-RAG — 加入知識庫
- LangChain-Memory — 對話記憶
- LangChain-Agent — 自主工具呼叫