1. 从零开始：用LangChain快速搭建大语言模型应用

最近在折腾大语言模型应用开发时，发现LangChain这个框架确实能极大提升开发效率。作为一个长期奋战在一线的开发者，我想分享下如何用LangChain快速搭建一个可用的LLM应用。不同于官方文档的学院派风格，我会用更接地气的方式，结合实战经验来讲解。

LangChain本质上是一个为大语言模型应用开发设计的框架，它把常见的开发模式抽象成组件和链条（Chain）。想象一下乐高积木，每个组件就像一块积木，而Chain就是把这些积木按照特定顺序连接起来的方式。这样做最大的好处是，开发者不用每次都从头造轮子，可以直接使用现成的模块快速搭建应用。

2. 环境准备与工具选型

2.1 基础环境配置

首先需要安装LangChain的核心包。建议使用Python 3.8及以上版本，太老的Python版本可能会有兼容性问题。安装命令很简单：

pip install langchain langchain-community

这里解释下这两个包的区别：

  • langchain是核心框架
  • langchain-community包含社区维护的各种第三方集成

注意：建议在虚拟环境中安装，避免污染全局Python环境。可以用venv或conda创建隔离环境。

2.2 模型提供商选择

LangChain支持多种大模型提供商，这里列举几个常用的：

# OpenAI (ChatGPT)
pip install langchain-openai

# Hugging Face
pip install langchain-huggingface

# 阿里云通义千问
pip install langchain-qwq

选择模型提供商时需要考虑几个因素：

  1. 模型性能：不同模型的智能程度差异很大
  2. API成本：商用API通常按token收费
  3. 响应速度：实时应用对延迟敏感
  4. 数据隐私：某些场景对数据出境有要求

以我的经验，如果是个人项目或测试，可以从通义千问开始，它的免费额度比较友好。如果是商业项目，GPT-4可能是更稳妥的选择。

3. 第一个LangChain应用实战

3.1 初始化模型连接

以通义千问为例，首先需要设置API密钥：

import os
from langchain_community.chat_models.tongyi import ChatTongyi

os.environ["DASHSCOPE_API_KEY"] = "你的API_KEY"
model = ChatTongyi(model="qwen-plus")

这里有几个关键点：

  1. API密钥建议通过环境变量传入，不要硬编码在代码中
  2. qwen-plus 是模型名称，不同提供商模型命名规则不同
  3. 首次使用需要到对应平台申请API密钥

3.2 构建提示词模板

LangChain提供了强大的提示词模板功能：

from langchain_core.prompts import ChatPromptTemplate

prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个精通{topic}的资深技术专家。"),
    ("user", "请用三句话解释一下什么是{concept}。")
])

这个模板有几个特点：

  1. 使用 {} 定义变量，实际调用时动态填充
  2. 支持多角色对话（system/user/ai）
  3. system消息用于设定AI的角色和背景

实战技巧：system提示词对输出质量影响很大。根据我的经验，明确指定AI的角色和专业领域，能显著提升回答的专业性。

3.3 创建处理链

LangChain的核心概念就是Chain（链）：

from langchain_core.output_parsers import StrOutputParser

output_parser = StrOutputParser()
chain = prompt | model | output_parser

这个 | 操作符看起来像Unix管道，确实概念也很相似 - 数据从左向右流动：

  1. 首先输入数据填充prompt模板
  2. 然后将完整prompt发送给模型
  3. 最后用解析器处理模型输出

3.4 调用模型获取响应

调用方式有两种，根据场景选择：

# 一次性获取完整响应（适合非实时场景）
response = chain.invoke({"topic": "Python", "concept": "列表"})
print(response)

# 流式输出（适合实时交互场景）
for chunk in chain.stream({"topic": "人工智能", "concept": "神经网络"}):
    print(chunk, end="", flush=True)

流式输出的优势：

  1. 用户体验更好，像打字机一样逐步显示
  2. 对于长响应可以提前看到部分结果
  3. 在某些平台上能减少响应感知延迟

4. 核心组件深度解析

4.1 模型封装详解

不同的模型提供商有不同的初始化方式：

# 通义千问
from langchain_community.chat_models.tongyi import ChatTongyi
model = ChatTongyi(model="qwen-plus")

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

# Hugging Face
from langchain_community.chat_models.huggingface import ChatHuggingFace
model = ChatHuggingFace(model="meta-llama/Llama-2-7b-chat-hf")

关键参数说明：

  • temperature ：控制输出的随机性（0-1）
  • max_tokens ：限制响应长度
  • top_p ：核采样参数，影响多样性

经验之谈：temperature设为0.7左右通常能在确定性和创造性之间取得不错平衡。对于需要精确答案的场景，可以降到0.3以下。

4.2 提示词工程技巧

好的提示词是获得优质回答的关键。LangChain的提示词模板支持多种高级功能：

from langchain_core.prompts import ChatPromptTemplate, FewShotChatMessagePromptTemplate

# 少量示例提示词
examples = [
    {"input": "2+2", "output": "4"},
    {"input": "5*3", "output": "15"}
]

example_prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("ai", "{output}")
])

few_shot_prompt = FewShotChatMessagePromptTemplate(
    example_prompt=example_prompt,
    examples=examples
)

这种少量示例(few-shot)方式特别适合：

  1. 定义特定的回答格式
  2. 教会模型特定领域的推理方式
  3. 处理模型容易出错的边缘情况

4.3 输出解析器应用

除了简单的字符串解析，LangChain还支持结构化输出：

from langchain_core.output_parsers import CommaSeparatedListOutputParser

parser = CommaSeparatedListOutputParser()
chain = prompt | model | parser

result = chain.invoke({"topic": "编程语言", "concept": "主流语言"})
# 返回的是列表而不是字符串

常用输出解析器类型：

  1. StrOutputParser - 普通文本
  2. CommaSeparatedListOutputParser - 逗号分隔列表
  3. StructuredOutputParser - 复杂JSON结构
  4. RegexParser - 用正则表达式提取内容

5. 实战中的常见问题与解决方案

5.1 超时与重试机制

网络请求难免会遇到超时，LangChain提供了自动重试功能：

from langchain_community.chat_models.tongyi import ChatTongyi
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 safe_invoke():
    model = ChatTongyi(model="qwen-plus", request_timeout=30)
    return model.invoke("请用中文回答")

response = safe_invoke()

这个配置表示：

  • 最多重试3次
  • 等待时间指数增长（4s, 8s, 16s）
  • 单次请求超时30秒

5.2 速率限制处理

调用商用API常会遇到速率限制，解决方案：

from langchain_community.chat_models.tongyi import ChatTongyi
from langchain_core.callbacks import CallbackManager
from langchain_core.callbacks.streaming_stdout import StreamingStdOutCallbackHandler

model = ChatTongyi(
    model="qwen-plus",
    streaming=True,
    callback_manager=CallbackManager([StreamingStdOutCallbackHandler()]),
    max_retries=5,
    retry_min_seconds=1,
    retry_max_seconds=10
)

关键参数：

  • max_retries ：最大重试次数
  • retry_min/max_seconds ：重试间隔范围
  • streaming ：启用流式可以降低超时概率

5.3 本地模型部署方案

如果对数据隐私要求高，可以考虑本地部署：

from langchain_community.chat_models.huggingface import ChatHuggingFace
from transformers import AutoModelForCausalLM, AutoTokenizer

model_name = "meta-llama/Llama-2-7b-chat-hf"
tokenizer = AutoTokenizer.from_pretrained(model_name)
model = AutoModelForCausalLM.from_pretrained(model_name)

hf_model = ChatHuggingFace(model=model, tokenizer=tokenizer)

本地部署的优缺点：

  • 优点：数据不出内网，长期成本可能更低
  • 缺点：需要GPU资源，推理速度较慢

6. 性能优化与高级技巧

6.1 缓存机制应用

重复查询相同内容时，使用缓存可以显著节省成本：

from langchain.cache import InMemoryCache
from langchain.globals import set_llm_cache

set_llm_cache(InMemoryCache())

# 第一次调用会真实请求API
response1 = chain.invoke({"topic": "Python", "concept": "装饰器"})

# 第二次相同输入会直接从缓存返回
response2 = chain.invoke({"topic": "Python", "concept": "装饰器"})

LangChain支持的缓存后端：

  1. InMemoryCache - 内存缓存，重启失效
  2. SQLiteCache - 本地数据库缓存
  3. RedisCache - Redis缓存，适合分布式场景

6.2 异步并发调用

对于批量处理场景，异步调用可以大幅提升效率：

import asyncio
from langchain_core.runnables import RunnableLambda

async def batch_invoke():
    chain = prompt | model | output_parser
    inputs = [
        {"topic": "Python", "concept": "生成器"},
        {"topic": "Go", "concept": "协程"},
        {"topic": "Java", "concept": "Stream API"}
    ]
    return await chain.abatch(inputs)

results = asyncio.run(batch_invoke())

异步调用的注意事项：

  1. 检查API提供商的并发限制
  2. 合理控制并发量，避免触发速率限制
  3. 错误处理更复杂，需要单独捕获每个任务异常

6.3 监控与日志记录

生产环境需要完善的监控：

from langchain.callbacks.tracers import LangChainTracer
from langchain.callbacks.manager import collect_runs

tracer = LangChainTracer()

with collect_runs() as cb:
    response = chain.invoke(
        {"topic": "DevOps", "concept": "CI/CD"},
        config={"callbacks": [tracer]}
    )
    
print(cb.traced_runs[0].outputs)

监控的关键指标：

  1. 请求延迟
  2. Token使用量
  3. 错误率
  4. 速率限制触发次数

7. 项目架构设计建议

7.1 模块化设计

大型项目应该采用模块化设计：

my_llm_app/
├── chains/
│   ├── qa_chain.py
│   └── summarization_chain.py
├── models/
│   └── llm_manager.py
├── prompts/
│   ├── system_prompts.py
│   └── few_shot_examples.py
└── main.py

各模块职责：

  • chains/ ：存放各种业务链
  • models/ ：模型初始化与管理
  • prompts/ ：提示词模板管理
  • main.py ：应用入口

7.2 配置管理

建议使用配置文件管理参数：

# config.yaml
models:
  tongyi:
    api_key: ${TONGYI_API_KEY}
    model_name: qwen-plus
    temperature: 0.7

prompts:
  tech_expert:
    system: "你是一个精通{topic}的资深技术专家"
    user: "请用三句话解释{concept}"

然后在代码中加载配置：

import yaml
from langchain_core.prompts import ChatPromptTemplate

with open("config.yaml") as f:
    config = yaml.safe_load(f)

prompt = ChatPromptTemplate.from_messages([
    ("system", config["prompts"]["tech_expert"]["system"]),
    ("user", config["prompts"]["tech_expert"]["user"])
])

7.3 测试策略

LLM应用测试的挑战：

  1. 输出具有不确定性
  2. 评估标准主观性强
  3. 运行成本高

建议的测试方案：

  1. 对确定性逻辑（如输入处理）写单元测试
  2. 对模型输出进行抽样人工评估
  3. 使用评估框架如RAGAS进行自动化评估
def test_chain_output_format():
    response = chain.invoke({"topic": "Test", "concept": "Test"})
    assert isinstance(response, str)
    assert len(response.split("。")) == 3  # 检查是否三句话

8. 从Demo到生产的关键考量

当应用要从演示版升级到生产环境时，有几个关键点需要考虑：

  1. 错误处理与重试 ：生产环境必须有完善的错误处理机制，特别是对于网络不稳定的情况。建议实现指数退避的重试策略。

  2. 性能监控 ：需要监控API调用延迟、错误率和token使用情况。可以使用Prometheus等工具采集指标。

  3. 成本控制 ：商用API按token计费，需要：

    • 实现使用量告警
    • 考虑缓存常用查询结果
    • 对长文本进行适当截断

  4. 安全防护 ：

    • 对用户输入进行严格的过滤和清理
    • 防范提示词注入攻击
    • 敏感数据脱敏处理

  5. 可观测性 ：

    • 记录完整的请求/响应日志（注意隐私）
    • 实现追踪链式调用过程的能力
    • 设置合理的日志保留策略
# 生产级错误处理示例
from tenacity import retry, stop_after_attempt, wait_exponential
from langchain_core.exceptions import LangChainError

@retry(
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=1, min=4, max=10),
    retry=retry_if_exception_type(LangChainError)
)
def safe_invoke(chain, input_dict):
    try:
        return chain.invoke(input_dict)
    except LangChainError as e:
        log_error(e)
        raise

9. 扩展应用场景

LangChain不仅限于简单的问答场景，还可以支持更复杂的应用：

9.1 文档问答系统

from langchain_community.document_loaders import WebBaseLoader
from langchain_text_splitters import RecursiveCharacterTextSplitter
from langchain_community.vectorstores import FAISS
from langchain_core.runnables import RunnablePassthrough

# 加载并分割文档
loader = WebBaseLoader("https://example.com/doc")
docs = loader.load()
text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000, chunk_overlap=200)
splits = text_splitter.split_documents(docs)

# 创建向量存储
vectorstore = FAISS.from_documents(docs, embedding_model)
retriever = vectorstore.as_retriever()

# 构建问答链
qa_chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | model
    | output_parser
)

9.2 数据分析Agent

from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain.tools import Tool

def query_database(query: str) -> str:
    # 实现数据库查询逻辑
    return "查询结果"

tools = [
    Tool(
        name="db_query",
        func=query_database,
        description="用于查询业务数据库"
    )
]

agent = create_openai_tools_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent, tools=tools)

result = agent_executor.invoke({
    "input": "我们上个月销售额最高的产品是什么？"
})

9.3 自动化工作流

from langchain_core.runnables import RunnableBranch

def route_based_on_topic(input_dict):
    if "Python" in input_dict["topic"]:
        return python_chain
    else:
        return general_chain

branch_chain = RunnableBranch(
    (lambda x: "Python" in x["topic"], python_chain),
    general_chain
)

full_chain = {
    "topic": lambda x: x["topic"],
    "concept": lambda x: x["concept"]
} | branch_chain

10. 持续学习与资源推荐

要深入掌握LangChain，我推荐以下学习路径：

  1. 官方文档 ：虽然有些部分比较简略，但架构说明很值得细读

    • 重点阅读：Chain、Memory、Agent等核心概念

  2. 开源项目 ：

    • LangChain模板库：大量现成的应用示例
    • LangSmith：官方的调试和监控平台（目前处于beta）

  3. 进阶书籍 ：

    • 《LangChain in Action》：实战案例丰富
    • 《Prompt Engineering for Developers》：深入提示词工程

  4. 社区资源 ：

    • LangChain官方Discord频道
    • 相关技术论坛的LLM板块

  5. 实践建议 ：

    • 从简单应用开始，逐步增加复杂度
    • 建立自己的代码片段库
    • 参与开源项目贡献

最后分享一个我在实际项目中的体会：LangChain虽然降低了开发门槛，但要构建真正可用的生产级应用，仍然需要对底层原理有扎实的理解。建议在使用框架的同时，也要学习Transformer架构、注意力机制等基础知识，这样才能在遇到问题时更快找到解决方案。

Logo
AI Agent技术社区

Agent 垂直技术社区，欢迎活跃、内容共建。

更多推荐

cover

AI Agent 多角色协作总是重复问需求？用单一事实源串起文档、代码和测试

avatar AI Agent技术社区

零成本在终端里搭建 AI 编程智能体（Open Interpreter + 智谱免费模型）

作为嵌入式开发者，我一直在寻找一种能的 AI 工具。Claude Code 虽然强大，但需要付费 API 且国内网络不稳定；GitHub Copilot 依赖 IDE，不够灵活。后来我发现了——一个开源的终端 AI Agent，支持多种大模型后端。配合智谱 AI 的免费模型，我实现了，能读写文件、执行 Shell 命令、管理 QEMU 虚拟机，极大提升了内核和驱动学习效率。这篇文章记录完整的搭建过

avatar AI Agent技术社区

AIOps 项目做了一个 SRE 实验排障 AI Agent

of=/dev/sdX、iptables -F、nft flush ruleset、systemctl stop sshd|network|firewalld、chmod -R 777 /、chown -R、curl …| sh、wget …| bash、kill -9 1、shutdown、reboot、setenforce 0 等。项目代号 SRE Lab Doctor，仓库 andersth

avatar AI Agent技术社区