LangChain 开发框架入门

什么是 LangChain

LangChain 是目前最流行的 LLM 应用开发框架，提供了构建 AI 应用所需的各种组件和工具。

核心优势

• 模块化设计：各组件可独立使用，灵活组合
• 丰富的集成：支持 100+ 大模型和向量数据库
• 活跃的社区：GitHub 90k+ stars，生态繁荣
• 生产就绪：LangSmith 提供完整的调试和监控

适用场景

安装配置

Python 版本

# 安装核心包
pip install langchain langchain-openai

# 安装常用集成
pip install langchain-anthropic  # Claude
pip install langchain-community  # 社区集成
pip install langchain-chroma     # 向量数据库

JavaScript/TypeScript 版本

npm install langchain @langchain/openai @langchain/anthropic

环境变量

export OPENAI_API_KEY="your-openai-key"
export ANTHROPIC_API_KEY="your-anthropic-key"

核心概念

1. Chat Models（对话模型）

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic

# OpenAI
llm = ChatOpenAI(model="gpt-4o", temperature=0.7)

# Anthropic
llm = ChatAnthropic(model="claude-sonnet-4-20250514")

# 调用
response = llm.invoke("你好，介绍一下你自己")
print(response.content)

2. Prompt Templates（提示词模板）

from langchain_core.prompts import ChatPromptTemplate

# 简单模板
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一位{role}专家"),
    ("human", "{question}")
])

# 使用模板
chain = prompt | llm
response = chain.invoke({
    "role": "Python",
    "question": "如何读取JSON文件？"
})

3. Output Parsers（输出解析器）

from langchain_core.output_parsers import JsonOutputParser
from pydantic import BaseModel

class Recipe(BaseModel):
    name: str
    ingredients: list[str]
    steps: list[str]

parser = JsonOutputParser(pydantic_object=Recipe)

prompt = ChatPromptTemplate.from_messages([
    ("system", "返回JSON格式的食谱。{format_instructions}"),
    ("human", "给我一个{dish}的食谱")
])

chain = prompt | llm | parser
result = chain.invoke({
    "dish": "番茄炒蛋",
    "format_instructions": parser.get_format_instructions()
})

4. Chains（链）

使用 LCEL (LangChain Expression Language) 构建链：

from langchain_core.runnables import RunnablePassthrough

# 简单链
chain = prompt | llm | parser

# 复杂链
chain = (
    {"context": retriever, "question": RunnablePassthrough()}
    | prompt
    | llm
    | parser
)

RAG 实战

构建简单的 RAG 系统

from langchain_openai import OpenAIEmbeddings, ChatOpenAI
from langchain_chroma import Chroma
from langchain_community.document_loaders import TextLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough

# 1. 加载文档
loader = TextLoader("./data/document.txt")
documents = loader.load()

# 2. 分割文档
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200
)
chunks = splitter.split_documents(documents)

# 3. 创建向量存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(chunks, embeddings)

# 4. 创建检索器
retriever = vectorstore.as_retriever(search_kwargs={"k": 4})

# 5. 构建 RAG 链
template = """基于以下上下文回答问题：

上下文：
{context}

问题：{question}

回答："""

prompt = ChatPromptTemplate.from_template(template)
llm = ChatOpenAI(model="gpt-4o")

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

rag_chain = (
    {"context": retriever | format_docs, "question": RunnablePassthrough()}
    | prompt
    | llm
)

# 6. 查询
answer = rag_chain.invoke("文档中提到了什么内容？")
print(answer.content)

Agent 开发

创建工具调用 Agent

from langchain_openai import ChatOpenAI
from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.tools import tool

# 定义工具
@tool
def search(query: str) -> str:
    """搜索网络获取信息"""
    # 实现搜索逻辑
    return f"搜索结果：{query}"

@tool
def calculator(expression: str) -> str:
    """计算数学表达式"""
    return str(eval(expression))

tools = [search, calculator]

# 创建 Agent
llm = ChatOpenAI(model="gpt-4o")
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个有帮助的助手，可以使用工具完成任务。"),
    ("human", "{input}"),
    ("placeholder", "{agent_scratchpad}")
])

agent = create_tool_calling_agent(llm, tools, prompt)
executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

# 运行
result = executor.invoke({"input": "计算 123 * 456 等于多少"})
print(result["output"])

LangGraph 进阶

LangGraph 是 LangChain 的扩展，用于构建复杂的多步骤 Agent：

from langgraph.graph import StateGraph, END
from typing import TypedDict, Annotated
import operator

class AgentState(TypedDict):
    messages: Annotated[list, operator.add]
    next_step: str

# 定义节点
def research_node(state: AgentState):
    # 研究逻辑
    return {"messages": ["研究完成"], "next_step": "analyze"}

def analyze_node(state: AgentState):
    # 分析逻辑
    return {"messages": ["分析完成"], "next_step": "report"}

# 构建图
workflow = StateGraph(AgentState)
workflow.add_node("research", research_node)
workflow.add_node("analyze", analyze_node)
workflow.add_edge("research", "analyze")
workflow.add_edge("analyze", END)
workflow.set_entry_point("research")

app = workflow.compile()

最佳实践

1. 使用 LangSmith 调试

import os
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-langsmith-key"

2. 错误处理

from langchain_core.runnables import RunnableConfig

config = RunnableConfig(
    max_concurrency=5,
    recursion_limit=10,
    configurable={"timeout": 30}
)

result = chain.invoke(input, config=config)

3. 流式输出

for chunk in chain.stream({"question": "你好"}):
    print(chunk, end="", flush=True)

下一步

• LlamaIndex 教程 - 另一个流行的 RAG 框架
• Vercel AI SDK 教程 - Web AI 应用开发