快速开始 #

环境准备 #

安装 LangChain #

LangChain 采用模块化设计,核心包和各提供商的集成包分开安装:

bash
# 安装核心包
pip install langchain

# 安装 LangChain 生态系统
pip install langchain-core      # 核心接口
pip install langchain-community # 社区集成

# 安装特定模型提供商的包
pip install langchain-openai    # OpenAI
pip install langchain-anthropic # Anthropic Claude
pip install langchain-google-genai  # Google Gemini

配置 API Key #

bash
# 创建 .env 文件
touch .env

# 添加 API Key
echo "OPENAI_API_KEY=your-api-key-here" >> .env
python
# 在代码中加载环境变量
from dotenv import load_dotenv

load_dotenv()  # 加载 .env 文件

项目结构 #

text
my-langchain-project/
├── .env              # 环境变量
├── .gitignore        # Git 忽略文件
├── requirements.txt  # 依赖列表
├── main.py          # 主程序
└── README.md        # 项目说明
text
# .gitignore
.env
__pycache__/
*.pyc
.venv/

第一个程序 #

Hello World #

python
# hello.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage

# 创建模型实例
model = ChatOpenAI(model="gpt-4o-mini")

# 发送消息
message = HumanMessage(content="你好,请用一句话介绍 LangChain")
response = model.invoke([message])

print(response.content)

运行程序:

bash
python hello.py

输出示例:

text
LangChain 是一个用于开发大语言模型应用的开源框架,提供了模型集成、提示管理、链式调用等功能。

使用 LCEL 语法 #

LangChain Expression Language (LCEL) 是一种声明式语法,让组件组合更简洁:

python
# lcel_hello.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 定义提示模板
prompt = ChatPromptTemplate.from_template("给我讲一个关于{topic}的笑话")

# 创建模型
model = ChatOpenAI(model="gpt-4o-mini")

# 创建输出解析器
output_parser = StrOutputParser()

# 使用 LCEL 语法组合
chain = prompt | model | output_parser

# 调用链
result = chain.invoke({"topic": "程序员"})
print(result)
text
┌─────────────────────────────────────────────────────────────┐
│                    LCEL 数据流                               │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  {"topic": "程序员"}                                        │
│        │                                                    │
│        ▼                                                    │
│  ┌─────────┐                                                │
│  │ prompt  │  "给我讲一个关于程序员的笑话"                   │
│  └────┬────┘                                                │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────┐                                                │
│  │  model  │  LLM 生成回复                                  │
│  └────┬────┘                                                │
│       │                                                     │
│       ▼                                                     │
│  ┌─────────┐                                                │
│  │ parser  │  提取字符串内容                                │
│  └────┬────┘                                                │
│       │                                                     │
│       ▼                                                     │
│  "为什么程序员总是分不清万圣节和圣诞节?..."                 │
│                                                             │
└─────────────────────────────────────────────────────────────┘

基础用法 #

1. 聊天模型 #

python
# chat_model.py
from langchain_openai import ChatOpenAI
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage

# 创建模型
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)

# 单轮对话
response = model.invoke([HumanMessage(content="什么是 Python?")])
print("单轮对话:", response.content)

# 多轮对话
messages = [
    SystemMessage(content="你是一个友好的 Python 专家"),
    HumanMessage(content="Python 有什么优点?"),
    AIMessage(content="Python 有很多优点,比如语法简洁、生态丰富..."),
    HumanMessage(content="那缺点呢?"),
]
response = model.invoke(messages)
print("多轮对话:", response.content)

2. 提示模板 #

python
# prompt_template.py
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 基础模板
prompt = ChatPromptTemplate.from_template("告诉我关于{topic}的{count}个事实")
formatted = prompt.format(topic="Python", count="3")
print("格式化后:", formatted)

# 带系统消息的模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个{role},请用{style}的风格回答问题"),
    ("human", "{question}")
])

# 带历史消息的模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手"),
    MessagesPlaceholder(variable_name="chat_history"),
    ("human", "{question}")
])

3. 输出解析器 #

python
# output_parser.py
from langchain_core.output_parsers import StrOutputParser, JsonOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_openai import ChatOpenAI
from pydantic import BaseModel, Field

# 字符串解析器
str_parser = StrOutputParser()

# JSON 解析器
class Person(BaseModel):
    name: str = Field(description="人物姓名")
    age: int = Field(description="人物年龄")
    occupation: str = Field(description="职业")

json_parser = JsonOutputParser(pydantic_object=Person)

# 使用解析器
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手,请以 JSON 格式输出。"),
    ("human", "{query}\n\n{format_instructions}")
])

model = ChatOpenAI(model="gpt-4o-mini")

chain = prompt | model | json_parser

result = chain.invoke({
    "query": "告诉我一个虚构人物的信息",
    "format_instructions": json_parser.get_format_instructions()
})

print(result)
# {'name': '张三', 'age': 30, 'occupation': '软件工程师'}

4. 简单链 #

python
# simple_chain.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

# 定义组件
model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template(
    "你是一个{style}风格的作家。请写一段关于{topic}的短文。"
)
parser = StrOutputParser()

# 组合成链
chain = prompt | model | parser

# 调用
result = chain.invoke({
    "style": "幽默",
    "topic": "程序员的生活"
})

print(result)

流式输出 #

对于长文本生成,流式输出可以提供更好的用户体验:

python
# streaming.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("写一篇关于{topic}的短文")
parser = StrOutputParser()

chain = prompt | model | parser

# 流式输出
for chunk in chain.stream({"topic": "人工智能"}):
    print(chunk, end="", flush=True)

print()  # 换行
text
┌─────────────────────────────────────────────────────────────┐
│                    流式输出过程                              │
├─────────────────────────────────────────────────────────────┤
│                                                             │
│  时间线:                                                    │
│                                                             │
│  t0: "人工"                                                 │
│  t1: "智能"                                                 │
│  t2: "("                                                   │
│  t3: "AI"                                                   │
│  t4: ")"                                                   │
│  t5: "是"                                                   │
│  t6: "计算"                                                 │
│  t7: "机"                                                   │
│  ...                                                        │
│                                                             │
│  用户可以实时看到文字逐字出现                                │
│                                                             │
└─────────────────────────────────────────────────────────────┘

批量处理 #

当需要处理多个输入时,使用批量处理更高效:

python
# batch.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("用一句话解释:{concept}")
parser = StrOutputParser()

chain = prompt | model | parser

# 批量处理
concepts = [
    {"concept": "机器学习"},
    {"concept": "深度学习"},
    {"concept": "自然语言处理"},
]

results = chain.batch(concepts)

for concept, result in zip(concepts, results):
    print(f"{concept['concept']}: {result}")
    print("-" * 50)

异步处理 #

对于高并发场景,使用异步 API:

python
# async_demo.py
import asyncio
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("用一句话介绍:{name}")
parser = StrOutputParser()

chain = prompt | model | parser

async def main():
    # 异步调用
    result = await chain.ainvoke({"name": "Python"})
    print("异步结果:", result)
    
    # 异步批量
    results = await chain.abatch([
        {"name": "JavaScript"},
        {"name": "Go"},
        {"name": "Rust"},
    ])
    print("批量结果:", results)
    
    # 异步流式
    print("流式输出:")
    async for chunk in chain.astream({"name": "LangChain"}):
        print(chunk, end="", flush=True)
    print()

asyncio.run(main())

使用不同模型 #

LangChain 支持多种模型提供商,切换非常简单:

python
# different_models.py

# OpenAI
from langchain_openai import ChatOpenAI
openai_model = ChatOpenAI(model="gpt-4o-mini")

# Anthropic Claude
from langchain_anthropic import ChatAnthropic
claude_model = ChatAnthropic(model="claude-3-sonnet-20240229")

# Google Gemini
from langchain_google_genai import ChatGoogleGenerativeAI
gemini_model = ChatGoogleGenerativeAI(model="gemini-pro")

# 本地模型 (Ollama)
from langchain_community.chat_models import ChatOllama
local_model = ChatOllama(model="llama3")

# 统一接口调用
models = {
    "OpenAI": openai_model,
    "Claude": claude_model,
    "Gemini": gemini_model,
    "Local": local_model,
}

for name, model in models.items():
    response = model.invoke("说一句问候语")
    print(f"{name}: {response.content}")

错误处理 #

python
# error_handling.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.exceptions import OutputParserException

model = ChatOpenAI(model="gpt-4o-mini")
prompt = ChatPromptTemplate.from_template("{input}")
parser = StrOutputParser()

chain = prompt | model | parser

try:
    result = chain.invoke({"input": "Hello"})
    print(result)
except Exception as e:
    print(f"发生错误: {type(e).__name__}: {e}")

# 使用 with_retry 自动重试
chain_with_retry = chain.with_retry(
    stop_after_attempt=3,
    wait_exponential_jitter=True,
    retry_if_exception_type=(Exception,)
)

完整示例:简单问答机器人 #

python
# qa_bot.py
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
from langchain_core.output_parsers import StrOutputParser
from langchain_core.messages import HumanMessage, AIMessage
from langchain.memory import ConversationBufferMemory

# 初始化
model = ChatOpenAI(model="gpt-4o-mini", temperature=0.7)
memory = ConversationBufferMemory(return_messages=True)

# 创建提示模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个友好的助手,请用中文回答问题。"),
    MessagesPlaceholder(variable_name="history"),
    ("human", "{input}")
])

# 创建链
chain = prompt | model | StrOutputParser()

def chat(user_input: str) -> str:
    # 获取历史消息
    history = memory.load_memory_variables({})["history"]
    
    # 调用链
    response = chain.invoke({
        "history": history,
        "input": user_input
    })
    
    # 保存对话
    memory.save_context(
        {"input": user_input},
        {"output": response}
    )
    
    return response

# 交互循环
if __name__ == "__main__":
    print("问答机器人已启动,输入 'quit' 退出")
    print("-" * 40)
    
    while True:
        user_input = input("你: ").strip()
        
        if user_input.lower() == "quit":
            print("再见!")
            break
        
        if not user_input:
            continue
        
        response = chat(user_input)
        print(f"AI: {response}")
        print("-" * 40)

运行效果:

text
问答机器人已启动,输入 'quit' 退出
----------------------------------------
你: 你好
AI: 你好!很高兴见到你,有什么我可以帮助你的吗?
----------------------------------------
你: 我刚才说了什么?
AI: 你刚才说"你好"。
----------------------------------------
你: quit
再见!

常见问题 #

1. API Key 配置 #

python
# 方式一:环境变量
import os
os.environ["OPENAI_API_KEY"] = "your-key"

# 方式二:直接传入
model = ChatOpenAI(
    model="gpt-4o-mini",
    api_key="your-key"
)

# 方式三:使用 .env 文件
from dotenv import load_dotenv
load_dotenv()

2. 代理设置 #

python
import os

# HTTP 代理
os.environ["HTTP_PROXY"] = "http://127.0.0.1:7890"
os.environ["HTTPS_PROXY"] = "http://127.0.0.1:7890"

# 或在模型初始化时设置
model = ChatOpenAI(
    model="gpt-4o-mini",
    openai_proxy="http://127.0.0.1:7890"
)

3. 模型参数 #

python
model = ChatOpenAI(
    model="gpt-4o-mini",
    temperature=0.7,      # 创造性 (0-2)
    max_tokens=1000,      # 最大输出长度
    top_p=0.9,           # 核采样
    frequency_penalty=0,  # 频率惩罚
    presence_penalty=0,   # 存在惩罚
    timeout=30,          # 请求超时
    max_retries=2,       # 最大重试次数
)

下一步 #

现在你已经掌握了 LangChain 的基础用法,接下来学习:

最后更新:2026-03-30