LangChain环境搭建与模型调用实战指南-AI智能范式网

LangChain环境搭建与模型调用实战指南

Wong Kosheng·2026-08-20·阅读 22 GEO检测（免费）

LangChain大模型调用Python开发

|难度：普通|阅读约 20 分钟

1. LangChain基础环境搭建与模型调用实战

1.1 安装LangChain及相关依赖库

在开始使用LangChain之前，我们需要先搭建好Python开发环境。推荐使用Anaconda创建独立的Python环境，避免与其他项目产生依赖冲突。以下是具体操作步骤：

bash复制# 创建名为langchain_env的Python3.9环境
conda create -n langchain_env python=3.9 -y
conda activate langchain_env

# 安装LangChain核心库及常用扩展
pip install langchain langchain-community langchain-ollama dashscope chromadb -i https://pypi.tuna.tsinghua.edu.cn/simple

注意：使用清华镜像源可以显著加快国内用户的下载速度。如果遇到SSL证书问题，可以添加--trusted-host pypi.tuna.tsinghua.edu.cn参数。

安装完成后，建议验证关键库的版本：

python复制import langchain
print(langchain.__version__)  # 应显示0.1.0或更高版本

1.2 配置大模型访问密钥

不同的大模型平台需要不同的认证方式。以阿里云通义千问为例，我们需要设置API_KEY环境变量：

bash复制# Linux/MacOS
export DASHSCOPE_API_KEY="your_api_key_here"

# Windows
set DASHSCOPE_API_KEY="your_api_key_here"

对于本地运行的Ollama模型，则无需API_KEY，但需要确保已正确安装并启动了Ollama服务。

需要模型API调用？免费领10W Token，多模型网关一键接入 Claude、DeepSeek 等主流模型。

2. LangChain核心功能深度解析

2.1 大模型基础调用方式

LangChain提供了多种调用大模型的方式，满足不同场景需求：

2.1.1 同步调用（基础版）

python复制from langchain_community.llms.tongyi import Tongyi

# 初始化通义千问模型
qwen_model = Tongyi(model="qwen-max")
response = qwen_model.invoke("请用Python写一个快速排序算法")
print(response)

2.1.2 流式输出（适合长文本）

python复制from langchain_ollama import OllamaLLM

gemma_model = OllamaLLM(model="gemma3:4b")
for chunk in gemma_model.stream("解释量子计算的基本原理"):
    print(chunk, end="", flush=True)  # 实时输出结果

实操心得：流式输出特别适合生成长篇内容或代码，可以实时看到生成过程，避免长时间等待。

2.2 聊天模式高级应用

聊天模式相比普通调用能保持对话上下文，适合多轮交互场景：

python复制from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.messages import HumanMessage, SystemMessage

chat = ChatTongyi(model="qwen3-max")

# 定义对话历史
messages = [
    SystemMessage(content="你是一位资深Python开发专家"),
    HumanMessage(content="如何优化Python代码的执行效率?")
]

response = chat.invoke(messages)
print(response.content)

2.2.1 简化的消息格式

LangChain还支持更简洁的元组格式定义消息：

python复制messages = [
    ("system", "你是一位机器学习研究员"),
    ("human", "解释Transformer架构的核心创新"),
    ("ai", "Transformer的核心是自注意力机制..."),
    ("human", "这与RNN相比有什么优势?")
]

3. LangChain高级特性实战

3.1 嵌入模型应用

嵌入模型(Embedding)是将文本转换为向量表示的关键组件：

python复制from langchain_community.embeddings import DashScopeEmbeddings

embedder = DashScopeEmbeddings(model="text-embedding-v2")

# 单文本嵌入
query_vec = embedder.embed_query("深度学习")
print(f"向量维度: {len(query_vec)}")

# 批量嵌入
docs_vec = embedder.embed_documents([
    "神经网络",
    "机器学习",
    "计算机视觉"
])

3.2 提示词模板工程

3.2.1 Zero-shot提示模板

python复制from langchain_core.prompts import PromptTemplate

template = PromptTemplate.from_template(
    "根据问题'{question}'和数据库结构{schema}，生成SQL查询语句"
)

prompt = template.format(
    question="查找销售额最高的产品",
    schema="CREATE TABLE products(id INT, name TEXT, price FLOAT)"
)
print(prompt)

3.2.2 Few-shot提示模板

python复制from langchain_core.prompts import FewShotPromptTemplate

examples = [
    {"question": "列出价格超过100元的产品", "sql": "SELECT * FROM products WHERE price > 100"},
    {"question": "统计每个类别的产品数量", "sql": "SELECT category, COUNT(*) FROM products GROUP BY category"}
]

few_shot_template = FewShotPromptTemplate(
    examples=examples,
    example_prompt=PromptTemplate.from_template("Q: {question}\nSQL: {sql}"),
    prefix="你是一个SQL专家，请根据示例回答",
    suffix="问题: {input_question}\nSQL:",
    input_variables=["input_question"]
)

4. LangChain链式编程实战

4.1 基础链式调用

python复制from langchain_core.prompts import ChatPromptTemplate
from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.output_parsers import StrOutputParser

prompt = ChatPromptTemplate.from_template("你是一位{role}，请回答：{question}")
model = ChatTongyi(model="qwen3-max")
parser = StrOutputParser()

chain = prompt | model | parser

response = chain.invoke({
    "role": "资深厨师",
    "question": "如何做出完美的牛排"
})
print(response)

4.2 带历史记录的对话链

python复制from langchain_core.runnables.history import RunnableWithMessageHistory
from langchain_core.chat_history import InMemoryChatMessageHistory

store = {}

def get_session_history(session_id):
    if session_id not in store:
        store[session_id] = InMemoryChatMessageHistory()
    return store[session_id]

conversation_chain = RunnableWithMessageHistory(
    chain,
    get_session_history,
    input_messages_key="question",
    history_messages_key="history"
)

response = conversation_chain.invoke(
    {"role": "营养师", "question": "早餐应该吃什么?"},
    config={"configurable": {"session_id": "user123"}}
)

5. 向量数据库与文档处理

5.1 文档加载与处理

python复制from langchain_community.document_loaders import CSVLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter

loader = CSVLoader(file_path="data/products.csv")
documents = loader.load()

splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", "。", "！", "？"]
)

splits = splitter.split_documents(documents)

5.2 Chroma向量数据库实战

python复制from langchain_chroma import Chroma

vector_store = Chroma.from_documents(
    documents=splits,
    embedding=DashScopeEmbeddings(),
    persist_directory="./chroma_db"
)

# 相似度搜索
results = vector_store.similarity_search("笔记本电脑", k=3)
for doc in results:
    print(doc.page_content)

6. 生产环境最佳实践

6.1 性能优化技巧

批量处理：尽量使用embed_documents而不是多次调用embed_query
缓存机制：对频繁查询的结果实现本地缓存
异步调用：对于高并发场景使用async/await

python复制import asyncio
from langchain_community.chat_models.tongyi import ChatTongyi

async def async_query():
    model = ChatTongyi(model="qwen3-max")
    return await model.ainvoke("异步查询测试")

asyncio.run(async_query())

6.2 错误处理与重试

python复制from tenacity import retry, stop_after_attempt, wait_exponential

@retry(stop=stop_after_attempt(3), wait=wait_exponential(multiplier=1, min=4, max=10))
def reliable_invoke(model, input_text):
    try:
        return model.invoke(input_text)
    except Exception as e:
        print(f"调用失败: {str(e)}")
        raise

7. 典型应用场景案例

7.1 智能客服系统

python复制def build_customer_service_chain(knowledge_base_path):
    loader = PyPDFLoader(knowledge_base_path)
    docs = loader.load()
    
    vector_store = Chroma.from_documents(
        docs, 
        DashScopeEmbeddings()
    )
    
    retriever = vector_store.as_retriever()
    
    prompt = ChatPromptTemplate.from_messages([
        ("system", "你是一个专业客服，请根据资料回答问题。资料：{context}"),
        ("human", "{question}")
    ])
    
    return {
        "context": retriever,
        "question": RunnablePassthrough()
    } | prompt | ChatTongyi() | StrOutputParser()

7.2 数据分析助手

python复制def data_analysis_assistant():
    template = """你是一个数据分析专家，请根据以下数据摘要回答问题：
    {data_summary}
    
    问题：{question}
    """
    
    prompt = PromptTemplate.from_template(template)
    
    return {
        "data_summary": load_and_summarize_data,
        "question": RunnablePassthrough()
    } | prompt | ChatTongyi() | JsonOutputParser()

在实际项目中，LangChain的这些功能可以组合使用，构建出功能强大的AI应用。建议从简单场景开始，逐步增加复杂度，同时注意监控API调用成本和性能指标。