langchain

LangChain - 使用 Agent 和 RAG 构建大模型应用

最受欢迎的大模型应用构建框架。

何时使用 LangChain

在以下情况下使用 LangChain：

• 使用工具调用和推理（ReAct 模式）构建 Agent
• 实现 RAG（检索增强生成）工作流
• 需要轻松切换大模型提供商（OpenAI、Anthropic、Google）
• 创建带有对话记忆的聊天机器人
• 快速原型化大模型应用
• 使用 LangSmith 可观测性进行生产部署

指标：

• 272,000+ 个代码仓库使用 LangChain
• 500+ 种集成（模型、向量存储、工具）
• 3,800+ 位贡献者

替代方案：

• LlamaIndex：专注于 RAG，更适合文档问答
• LangGraph：适用于复杂的有状态工作流，提供更多控制
• Haystack：用于生产级搜索工作流
• Semantic Kernel：适用于 Microsoft 生态系统

快速入门

安装

# Core library (Python 3.10+)
pip install -U langchain

# With OpenAI
pip install langchain-openai

# With Anthropic
pip install langchain-anthropic

# Common extras
pip install langchain-community  # 500+ integrations
pip install langchain-chroma     # Vector store

基本大模型用法

from langchain_anthropic import ChatAnthropic

# Initialize model
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929")

# Simple completion
response = llm.invoke("Explain quantum computing in 2 sentences")
print(response.content)

创建 Agent（ReAct 模式）

from langchain.agents import create_agent
from langchain_anthropic import ChatAnthropic

# Define tools
def get_weather(city: str) -> str:
    """Get current weather for a city."""
    return f"It's sunny in {city}, 72°F"

def search_web(query: str) -> str:
    """Search the web for information."""
    return f"Search results for: {query}"

# Create agent (<10 lines!)
agent = create_agent(
    model=ChatAnthropic(model="claude-sonnet-4-5-20250929"),
    tools=[get_weather, search_web],
    system_prompt="You are a helpful assistant. Use tools when needed."
)

# Run agent
result = agent.invoke({"messages": [{"role": "user", "content": "What's the weather in Paris?"}]})
print(result["messages"][-1].content)

核心概念

1. 模型 - 大模型抽象

from langchain_openai import ChatOpenAI
from langchain_anthropic import ChatAnthropic
from langchain_google_genai import ChatGoogleGenerativeAI

# Swap providers easily
llm = ChatOpenAI(model="gpt-4o")
llm = ChatAnthropic(model="claude-sonnet-4-5-20250929")
llm = ChatGoogleGenerativeAI(model="gemini-2.0-flash-exp")

# Streaming
for chunk in llm.stream("Write a poem"):
    print(chunk.content, end="", flush=True)

2. 链 - 顺序操作

from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

# Define prompt template
prompt = PromptTemplate(
    input_variables=["topic"],
    template="Write a 3-sentence summary about {topic}"
)

# Create chain
chain = LLMChain(llm=llm, prompt=prompt)

# Run chain
result = chain.run(topic="machine learning")

3. Agent - 使用工具进行推理

ReAct（推理 + 行动）模式：

from langchain.agents import create_tool_calling_agent, AgentExecutor
from langchain.tools import Tool
from scripts.safe_arithmetic import calculate

# Define custom tool
calculator = Tool(
    name="Calculator",
    func=lambda x: str(calculate(x)),
    description="Useful for arithmetic with numbers and +, -, *, /."
)

# Create agent with tools
agent = create_tool_calling_agent(
    llm=llm,
    tools=[calculator, search_web],
    prompt="Answer questions using available tools"
)

# Create executor
agent_executor = AgentExecutor(agent=agent, tools=[calculator], verbose=True)

# Run with reasoning
result = agent_executor.invoke({"input": "What is 25 * 17 + 142?"})

4. 记忆 - 对话历史

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationChain

# Add memory to track conversation
memory = ConversationBufferMemory()

conversation = ConversationChain(
    llm=llm,
    memory=memory,
    verbose=True
)

# Multi-turn conversation
conversation.predict(input="Hi, I'm Alice")
conversation.predict(input="What's my name?")  # Remembers "Alice"

RAG（检索增强生成）

详情、代码示例和参数表：references/rag-retrieval-augmented-generation.md。适用时请阅读。

高级 Agent 模式

结构化输出

from langchain_core.pydantic_v1 import BaseModel, Field

# Define schema
class WeatherReport(BaseModel):
    city: str = Field(description="City name")
    temperature: float = Field(description="Temperature in Fahrenheit")
    condition: str = Field(description="Weather condition")

# Get structured response
structured_llm = llm.with_structured_output(WeatherReport)
result = structured_llm.invoke("What's the weather in SF? It's 65F and sunny")
print(result.city, result.temperature, result.condition)

并行执行工具

from langchain.agents import create_tool_calling_agent

# Agent automatically parallelizes independent tool calls
agent = create_tool_calling_agent(
    llm=llm,
    tools=[get_weather, search_web, calculator]
)

# This will call get_weather("Paris") and get_weather("London") in parallel
result = agent.invoke({
    "messages": [{"role": "user", "content": "Compare weather in Paris and London"}]
})

流式执行 Agent

# Stream agent steps
for step in agent_executor.stream({"input": "Research AI trends"}):
    if "actions" in step:
        print(f"Tool: {step['actions'][0].tool}")
    if "output" in step:
        print(f"Output: {step['output']}")

常见模式

多文档问答

from langchain.chains.qa_with_sources import load_qa_with_sources_chain

# Load multiple documents
docs = [
    loader.load("https://docs.python.org"),
    loader.load("https://docs.numpy.org")
]

# QA with source citations
chain = load_qa_with_sources_chain(llm, chain_type="stuff")
result = chain({"input_documents": docs, "question": "How to use numpy arrays?"})
print(result["output_text"])  # Includes source citations

带错误处理的自定义工具

from langchain.tools import tool

@tool
def risky_operation(query: str) -> str:
    """Perform a risky operation that might fail."""
    try:
        # Your operation here
        result = perform_operation(query)
        return f"Success: {result}"
    except Exception as e:
        return f"Error: {str(e)}"

# Agent handles errors gracefully
agent = create_agent(model=llm, tools=[risky_operation])

LangSmith 可观测性

import os

# Enable tracing
os.environ["LANGCHAIN_TRACING_V2"] = "true"
os.environ["LANGCHAIN_API_KEY"] = "your-api-key"
os.environ["LANGCHAIN_PROJECT"] = "my-project"

# All chains/agents automatically traced
agent = create_agent(model=llm, tools=[calculator])
result = agent.invoke({"input": "Calculate 123 * 456"})

# View traces at smith.langchain.com

向量存储

Chroma（本地）

from langchain_chroma import Chroma

vectorstore = Chroma.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    persist_directory="./chroma_db"
)

Pinecone（云端）

from langchain_pinecone import PineconeVectorStore

vectorstore = PineconeVectorStore.from_documents(
    documents=docs,
    embedding=OpenAIEmbeddings(),
    index_name="my-index"
)

FAISS（相似度搜索）

from langchain_community.vectorstores import FAISS

vectorstore = FAISS.from_documents(docs, OpenAIEmbeddings())
vectorstore.save_local("faiss_index")

# Load later
vectorstore = FAISS.load_local("faiss_index", OpenAIEmbeddings())

文档加载器

# Web pages
from langchain_community.document_loaders import WebBaseLoader
loader = WebBaseLoader("https://example.com")

# PDFs
from langchain_community.document_loaders import PyPDFLoader
loader = PyPDFLoader("paper.pdf")

# GitHub
from langchain_community.document_loaders import GithubFileLoader
loader = GithubFileLoader(repo="user/repo", file_filter=lambda x: x.endswith(".py"))

# CSV
from langchain_community.document_loaders import CSVLoader
loader = CSVLoader("data.csv")

文本分割器

# Recursive (recommended for general text)
from langchain.text_splitter import RecursiveCharacterTextSplitter
splitter = RecursiveCharacterTextSplitter(
    chunk_size=1000,
    chunk_overlap=200,
    separators=["\n\n", "\n", " ", ""]
)

# Code-aware
from langchain.text_splitter import PythonCodeTextSplitter
splitter = PythonCodeTextSplitter(chunk_size=500)

# Semantic (by meaning)
from langchain_experimental.text_splitter import SemanticChunker
splitter = SemanticChunker(OpenAIEmbeddings())

最佳实践

1. 从简单开始 - 大多数情况下使用 create_agent()
2. 启用流式传输 - 为较长的响应提供更好的用户体验
3. 添加错误处理 - 工具可能会失败，请妥善处理
4. 使用 LangSmith - 对调试 Agent 至关重要
5. 优化分块大小 - RAG 使用 500-1000 个字符
6. 为提示词添加版本管理 - 跟踪生产环境中的更改
7. 缓存嵌入 - 嵌入成本高昂，尽可能缓存
8. 监控成本 - 使用 LangSmith 跟踪令牌用量

性能基准

操作	延迟	备注
简单大模型调用	~1-2s	取决于提供商
使用 1 个工具的 Agent	~3-5s	ReAct 推理开销
RAG 检索	~0.5-1s	向量搜索 + 大模型
为 1000 份文档生成嵌入	~10-30s	取决于模型

LangChain 与 LangGraph 的比较

特性	LangChain	LangGraph
最适用于	快速构建 Agent、RAG	复杂工作流
抽象级别	高	低
起步所需代码	<10 行	~30 行
控制能力	简单	完全控制
有状态工作流	有限	原生支持
循环图	不支持	支持
人在回路	基础	高级

在以下情况下使用 LangGraph：

• 需要带循环的有状态工作流
• 需要精细控制
• 构建多 Agent 系统
• 构建具有复杂逻辑的生产应用

参考资料

• Agent 指南 - ReAct、工具调用、流式传输
• RAG 指南 - 文档加载器、检索器、问答链
• 集成指南 - 向量存储、LangSmith、部署

资源

• GitHub：https://github.com/langchain-ai/langchain ⭐ 119,000+
• 文档：https://docs.langchain.com
• API 参考：https://reference.langchain.com/python
• LangSmith：https://smith.langchain.com（可观测性）
• 版本：0.3+（稳定版）
• 许可证：MIT

Agent 操作流程

1. 检查环境。 确认框架版本、模型提供商、API 密钥和速率限制。
2. 明确输入。 根据数据或用户确认格式、标识符和参数。对于任何会影响结果的值，不要猜测，应主动询问。
3. 先运行小规模版本。 使用已知输入测试单次调用或调用链，并检查原始输出。
4. 根据上述说明和参考资料执行完整任务。
5. 验证结果。 使用少量带标签的数据进行评估；根据架构检查结构化输出；记录提示词和响应。
6. 汇报。 说明运行了什么（版本、命令、参数）、检查了什么，以及仍有哪些不确定之处。

如果发生这种情况	请这样做
输出不符合预期架构	添加验证和重试、收紧架构，或简化提示词。
已安装版本中没有这些说明里的某个函数、标志或端点	查看已安装版本自身的文档（help()、--help、官方文档），调整做法并告知用户。切勿编造 API。
必需的输入、标识符或参数含糊不清	询问用户，或在运行前明确说明所采用的假设。

诚信准则

• 切勿编造结果、参数、标识符、引文或统计数据。如果无法运行或验证某项内容，请直言说明。
• 未经用户同意，切勿将私人或敏感数据发送到外部 API。
• 请将此处的特定版本细节视为可能已过时，并根据已安装版本的官方文档进行确认。
• 对于会产生费用、占用共享 GPU 或云配额、涉及个人或患者数据，或无法撤销的操作，请先征求同意。

相关技能

• llamaindex：用于构建带 RAG 的大模型应用的数据框架。
• autogpt-agents：用于构建和部署持续运行 Agent 的自主 AI Agent 平台。
• chroma：用于 AI 应用的开源嵌入数据库。