LangChain 快速开始

这页目标只有一个：让你在 15 分钟内跑通一个真正能用的 LangChain 应用，而不是看着示例代码发呆。

如果你是第一次接触 LangChain，建议按顺序过一遍；如果只是想找某个特定模式的代码，直接跳到对应部分。

---

#安装

bash复制
pip install langchain langchain-openai python-dotenv

然后创建 .env 文件：

bash复制
OPENAI_API_KEY="sk-proj-..."

.env 一定要加入 .gitignore，否则不小心 push 到 GitHub，OpenAI 机器人会在几分钟内发现并自动撤销你的 Key。

---

#第一个可运行的例子

最简单的三步 Chain——Prompt、LLM、Parser 用 | 串起来：

python复制
from dotenv import load_dotenv
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from langchain_core.output_parsers import StrOutputParser

load_dotenv()  # 这行放最前面，否则 API Key 加载时机可能出问题

chain = (
    ChatPromptTemplate.from_messages([
        ("system", "你是一个{role}，回答简洁，不超过 100 字。"),
        ("user", "{question}"),
    ])
    | ChatOpenAI(model="gpt-4o-mini")
    | StrOutputParser()
)

result = chain.invoke({"role": "Python 专家", "question": "什么是装饰器？"})
print(result)

能跑通这个，后面所有的例子都是在它基础上变形。

---

#换一个模型

LangChain 最实用的地方在这里：换模型改一行，其余逻辑一字不动。这也是为什么我在需要同时评估多个模型效果时，LangChain 比直接调 API 方便。

python复制
# OpenAI GPT
from langchain_openai import ChatOpenAI
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# Claude（需要先 pip install langchain-anthropic）
from langchain_anthropic import ChatAnthropic
llm = ChatAnthropic(model="claude-sonnet-4-20250514", temperature=0)

# Google Gemini（需要先 pip install langchain-google-genai）
from langchain_google_genai import ChatGoogleGenerativeAI
llm = ChatGoogleGenerativeAI(model="gemini-1.5-flash", temperature=0)

# 其余代码完全一样
chain = prompt | llm | StrOutputParser()

---

#返回结构化数据（不是纯字符串）

直接让模型"请输出 JSON"这种 Prompt 不稳定——模型心情不好时会在 JSON 前后套 markdown 代码块，或者字段名写错。用 Pydantic + JsonOutputParser 会可靠很多：

python复制
from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel, Field
from typing import List

class RecipeInfo(BaseModel):
    name: str = Field(description="菜品名称")
    ingredients: List[str] = Field(description="食材列表")
    steps: List[str] = Field(description="制作步骤")
    difficulty: str = Field(description="难度：简单/中等/困难")

parser = JsonOutputParser(pydantic_object=RecipeInfo)

chain = (
    ChatPromptTemplate.from_messages([
        ("system", "你是一个厨师助手，返回 JSON 格式的食谱。\n{format_instructions}"),
        ("user", "给我一个{dish}的食谱"),
    ]).partial(format_instructions=parser.get_format_instructions())
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | parser
)

result = chain.invoke({"dish": "番茄炒蛋"})
print(result["name"])          # 番茄炒蛋
print(result["difficulty"])    # 简单

---

#流式输出（前端实时显示）

python复制
chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

# 只改这一行，就从等全部结果变成边生成边输出
for chunk in chain.stream({"role": "讲故事的人", "question": "讲一个短故事"}):
    print(chunk, end="", flush=True)
print()  # 换行

做 Web 应用时基本都要用这个，用户看着光标一直不动会以为页面卡死了。

---

#多轮对话（带记忆）

python复制
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_community.chat_message_histories import ChatMessageHistory

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手。"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("user", "{question}"),
])

chain = prompt | ChatOpenAI(model="gpt-4o-mini") | StrOutputParser()

store = {}
def get_history(session_id: str):
    if session_id not in store:
        store[session_id] = ChatMessageHistory()
    return store[session_id]

chat = RunnableWithMessageHistory(
    chain,
    get_history,
    input_messages_key="question",
    history_messages_key="chat_history",
)

config = {"configurable": {"session_id": "demo"}}
print(chat.invoke({"question": "我叫小明"}, config=config))
print(chat.invoke({"question": "我叫什么名字？"}, config=config))  # 记得你叫小明

注意：这里用的是内存存储，进程重启历史就没了。生产环境需要接 Redis 或数据库——记忆这块有不少坑，详细见 Memory 记忆系统。

---

#简单的文档问答（RAG 雏形）

python复制
from langchain_openai import OpenAIEmbeddings
from langchain_community.vectorstores import Chroma
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_core.runnables import RunnablePassthrough

# 你的文档（字符串列表，或者用 TextLoader 加载文件）
texts = [
    "公司假期政策：员工每年有 15 天带薪年假，入职满 3 年后增加到 20 天。",
    "报销政策：差旅费用需在出差后 7 天内提交，超过 5000 元需要总监审批。",
    "工作时间：标准工时为周一到周五 9:00-18:00，弹性工作制需提前申请。",
]

# 向量化
vectorstore = Chroma.from_texts(texts, OpenAIEmbeddings())
retriever = vectorstore.as_retriever(search_kwargs={"k": 2})

# RAG Chain
def format_docs(docs):
    return "\n".join(doc.page_content for doc in docs)

rag_chain = (
    {
        "context": retriever | format_docs,
        "question": RunnablePassthrough(),
    }
    | ChatPromptTemplate.from_template(
        "基于以下文档回答问题，如果文档里没有就说不知道：\n\n{context}\n\n问题：{question}"
    )
    | ChatOpenAI(model="gpt-4o-mini", temperature=0)
    | StrOutputParser()
)

print(rag_chain.invoke("年假有多少天？"))

---

#并发处理多条输入

处理批量数据时，.batch() 比 for 循环快很多：

python复制
questions = [
    {"role": "历史老师", "question": "秦始皇统一六国是哪年？"},
    {"role": "地理老师", "question": "长江全长多少公里？"},
    {"role": "数学老师", "question": "什么是勾股定理？"},
]

# 并发处理，速度接近单次调用
results = chain.batch(questions)
for r in results:
    print(r)

---

#下一步去哪里

掌握基础后，按需深入：

---

官方文档：python.langchain.com↗ — 版本更新很快，遇到 API 报错先查这里