当前位置： 首页 > news >正文

langchain源码概览

news 2025/9/8 11:59:45

概述

LangChain 是一个用于构建大语言模型应用的强大框架，它通过可组合的组件和第三方集成简化AI应用开发，同时在底层技术发展时保持决策的前瞻性。

🌟 核心特点

🧩 模块化设计

独立的抽象组件，不绑定特定模型提供商

🔗 统一接口

通过Runnable接口提供一致的调用方式

📝 声明式编程

LCEL表达式语言简化组件组合

🚀 生产就绪

内置流式处理、异步支持、跟踪等企业级特性

项目架构

🏗️ 分层架构设计

LangChain 生态系统

├── langchain-core (核心抽象层)

│ ├── Runnable (通用调用接口)

│ ├── Language Models (语言模型抽象)

│ ├── Prompts (提示模板)

│ ├── Output Parsers (输出解析器)

│ ├── Retrievers (检索器接口)

│ ├── Vectorstores (向量存储)

│ ├── Messages (消息类型)

│ └── Tools (工具接口)

├── langchain (主要实现)

│ ├── Agents (智能代理)

│ ├── Chains (调用链)

│ ├── LLMs (大语言模型)

│ ├── Chat Models (对话模型)

│ ├── Document Loaders (文档加载器)

│ ├── Embeddings (嵌入模型)

│ ├── Memory (记忆机制)

│ ├── Retrievers (检索器实现)

│ └── Callbacks (回调系统)

├── partners/ (第三方集成)

│ ├── OpenAI

│ ├── Anthropic

│ ├── Hugging Face

│ ├── Chroma │

└── Ollama

├── langchain-cli (命令行工具)

└── text-splitters (文本分割器)

📁 目录结构解析

/libs/core/ - 核心抽象层
  • 职责: 定义所有基础接口和抽象
  • 特点: 轻量级依赖，无第三方集成
  • 关键模块:
    • runnables/ - Runnable接口实现
    • language_models/ - 语言模型基类
    • prompts/ - 提示模板系统
    • output_parsers/ - 输出解析器
/libs/langchain/ - 主要实现层
  • 职责: 具体功能实现和高级抽象
  • 关键模块:
    • agents/ - 智能代理系统
    • chains/ - 复合调用链
    • tools/ - 外部工具集成
    • memory/ - 对话记忆机制
/libs/partners/ - 第三方集成层
  • 职责: 各种服务商和工具的集成
  • 包含: OpenAI、Anthropic、HuggingFace等

核心概念详解

🔑 1. Runnable 接口 - 统一调用协议

Runnable 是整个LangChain的核心接口，为所有组件提供统一的调用方式：

# 基础调用方法
.invoke(input)          # 单次同步调用
.ainvoke(input)         # 单次异步调用
.batch(inputs)          # 批量处理
.stream(input)          # 流式输出
.astream(input)         # 异步流式输出# 高级功能
.with_config(config)    # 配置管理
.with_fallbacks([...])  # 故障转移
.with_retry()           # 重试机制
Runnable 的优势：
  • ✅ 统一接口 - 所有组件都有相同的调用方式
  • ✅ 自动优化 - 内置并行化和流式处理
  • ✅ 可组合性 - 轻松链接不同组件
  • ✅ 可观测性 - 自动跟踪和监控

🔗 2. LCEL (LangChain Expression Language)

LCEL 是一种声明式语言，用于组合LangChain组件：

# 基础语法
chain = prompt | model | output_parser# 并行处理
chain = {"context": retriever,"question": RunnablePassthrough()
} | prompt | model | output_parser# 条件分支
chain = RunnableBranch((lambda x: "math" in x["topic"], math_chain),(lambda x: "science" in x["topic"], science_chain),general_chain
)
LCEL 的特性：
  • 🔄 自动并行化 - 自动识别可并行的操作
  • 🌊 内置流式支持 - 实时输出中间结果
  • 📊 自动跟踪 - 完整的执行链路跟踪
  • ⚡ 异步支持 - 原生支持异步操作

🧩 3. 组件抽象层

语言模型层次结构

BaseLanguageModel (抽象基类)

├── LLM (文本输入输出)

│ ├── OpenAI

│ ├── Anthropic

│ └── HuggingFace

└── ChatModel (消息格式)

├── ChatOpenAI

├── ChatAnthropic

└── ChatOllama

提示模板层次结构

BasePromptTemplate

├── StringPromptTemplate

│ ├── PromptTemplate

│ └── FewShotPromptTemplate

└── BaseChatPromptTemplate

├── ChatPromptTemplate

└── MessagesPlaceholder

主要模块分析

🤖 1. Agents (智能代理)

位置: libs/langchain/langchain/agents/

智能代理是LangChain的高级功能，能够根据输入动态决定执行哪些操作：

核心组件：
  • Agent - 决策制定者
  • AgentExecutor - 执行器和安全控制
  • Tools - 可用工具集合
  • Memory - 对话历史记忆
常见代理类型：
# ReAct代理 - 推理和行动
agent = create_react_agent(llm, tools, prompt)# OpenAI Functions代理 - 函数调用
agent = create_openai_functions_agent(llm, tools, prompt)# Tool Calling代理 - 工具调用
agent = create_tool_calling_agent(llm, tools, prompt)

⛓️ 2. Chains (调用链)

位置: libs/langchain/langchain/chains/

Chains将多个组件组合成更复杂的应用：

核心链类型：
LLMChain - 基础链
from langchain.chains import LLMChainchain = LLMChain(llm=model,prompt=prompt,output_parser=parser
)
RetrievalQA - 检索问答链
from langchain.chains import RetrievalQAqa_chain = RetrievalQA.from_chain_type(llm=model,chain_type="stuff",  # map_reduce, refine, map_rerankretriever=vectorstore.as_retriever()
)
ConversationalRetrievalChain - 对话检索链
from langchain.chains import ConversationalRetrievalChainconversation_chain = ConversationalRetrievalChain.from_llm(llm=model,retriever=vectorstore.as_retriever(),memory=memory
)

🧠 3. Memory (记忆机制)

位置: libs/langchain/langchain/memory/

为对话提供上下文记忆：

记忆类型：
# 对话缓冲记忆
from langchain.memory import ConversationBufferMemory
memory = ConversationBufferMemory()# 对话摘要记忆
from langchain.memory import ConversationSummaryMemory
memory = ConversationSummaryMemory(llm=model)# 对话知识图谱记忆
from langchain.memory import ConversationKGMemory
memory = ConversationKGMemory(llm=model)

🔍 4. Retrievers (检索器)

位置: libs/langchain/langchain/retrievers/

用于从外部数据源检索相关信息：

检索器类型：
# 向量存储检索器
retriever = vectorstore.as_retriever(search_type="similarity",search_kwargs={"k": 4}
)# 多查询检索器
from langchain.retrievers import MultiQueryRetriever
retriever = MultiQueryRetriever.from_llm(retriever=base_retriever,llm=model
)# 压缩检索器
from langchain.retrievers import ContextualCompressionRetriever
retriever = ContextualCompressionRetriever(base_compressor=compressor,base_retriever=base_retriever
)

代码示例与最佳实践

🚀 1. 基础链式调用

from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_anthropic import ChatAnthropic# 组件定义
prompt = ChatPromptTemplate.from_template("你是一个专业的{domain}专家。请回答：{question}"
)
model = ChatAnthropic(model="claude-3-sonnet")
parser = StrOutputParser()# LCEL链式组合
chain = prompt | model | parser# 调用
result = chain.invoke({"domain": "机器学习","question": "什么是梯度下降？"
})
print(result)

🛠️ 2. 自定义工具和代理

from langchain_core.tools import tool
from langchain.agents import AgentExecutor, create_tool_calling_agent
from langchain_core.prompts import ChatPromptTemplate# 定义自定义工具
@tool
def calculate_area(length: float, width: float) -> float:"""计算矩形面积"""return length * width@tool
def get_weather(city: str) -> str:"""获取指定城市的天气信息"""# 实际应用中这里会调用天气APIreturn f"{city}今天晴，温度25°C"# 创建代理提示
prompt = ChatPromptTemplate.from_messages([("system", "你是一个有用的助手，可以使用提供的工具回答问题。"),("human", "{input}"),("placeholder", "{agent_scratchpad}"),
])# 工具列表
tools = [calculate_area, get_weather]# 创建并运行代理
agent = create_tool_calling_agent(model, tools, prompt)
agent_executor = AgentExecutor(agent=agent,tools=tools,verbose=True,handle_parsing_errors=True
)# 测试代理
result = agent_executor.invoke({"input": "请帮我计算一个长5米、宽3米的房间面积，然后告诉我北京的天气"
})

📚 3. RAG检索增强生成系统

from langchain_community.document_loaders import TextLoader
from langchain_community.vectorstores import Chroma
from langchain_community.embeddings import OpenAIEmbeddings
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.chains import RetrievalQA# 1. 文档加载和分割
loader = TextLoader("knowledge_base.txt")
documents = loader.load()text_splitter = RecursiveCharacterTextSplitter(chunk_size=1000,chunk_overlap=200
)
texts = text_splitter.split_documents(documents)# 2. 向量化存储
embeddings = OpenAIEmbeddings()
vectorstore = Chroma.from_documents(documents=texts,embedding=embeddings,persist_directory="./chroma_db"
)# 3. 创建检索QA链
qa_chain = RetrievalQA.from_chain_type(llm=model,chain_type="stuff",retriever=vectorstore.as_retriever(search_kwargs={"k": 4}),return_source_documents=True
)# 4. 查询
query = "LangChain的核心特性是什么？"
result = qa_chain.invoke({"query": query})print("答案:", result["result"])
print("来源文档:", result["source_documents"])

💬 4. 对话式应用

from langchain.memory import ConversationBufferMemory
from langchain.chains import ConversationalRetrievalChain# 设置记忆
memory = ConversationBufferMemory(memory_key="chat_history",return_messages=True
)# 创建对话检索链
conversation_chain = ConversationalRetrievalChain.from_llm(llm=model,retriever=vectorstore.as_retriever(),memory=memory,verbose=True
)# 多轮对话
print("开始对话（输入'quit'退出）：")
while True:question = input("你: ")if question.lower() == 'quit':breakresponse = conversation_chain.invoke({"question": question})print(f"AI: {response['answer']}")

🔄 5. 流式处理示例

async def streaming_example():"""展示流式处理能力"""# 创建流式链chain = prompt | model | StrOutputParser()print("流式输出：")async for chunk in chain.astream({"topic": "人工智能的未来"}):print(chunk, end="", flush=True)print("\n")# 运行流式示例
import asyncio
asyncio.run(streaming_example())

文章转载自：

http://qh3ktJiu.rkwwy.cn
http://rwIcXcrA.rkwwy.cn
http://rWC6oUUm.rkwwy.cn
http://7qS2UxxO.rkwwy.cn
http://YcI5ShFX.rkwwy.cn
http://l6GCYcP4.rkwwy.cn
http://duEIuhPZ.rkwwy.cn
http://dMErY8r6.rkwwy.cn
http://h5Y0JQ9E.rkwwy.cn
http://KdRKkQnf.rkwwy.cn
http://5nR5fwdN.rkwwy.cn
http://KrT8ByNQ.rkwwy.cn
http://uJiYJHJX.rkwwy.cn
http://NKVcb8PG.rkwwy.cn
http://GD5YpgmC.rkwwy.cn
http://66xJHafw.rkwwy.cn
http://JNdMAiLq.rkwwy.cn
http://ASpEoumq.rkwwy.cn
http://iZxoKyUX.rkwwy.cn
http://MMIL2T3Z.rkwwy.cn
http://t26kdkmL.rkwwy.cn
http://Y2AVWXIP.rkwwy.cn
http://8hFnDLIm.rkwwy.cn
http://nFcM7xrW.rkwwy.cn
http://8gjlsgfN.rkwwy.cn
http://JRtFBGzK.rkwwy.cn
http://4umcxnDH.rkwwy.cn
http://a55Og5b2.rkwwy.cn
http://QaeY5xVg.rkwwy.cn
http://fr6KI2y9.rkwwy.cn
http://www.dtcms.com/a/370828.html

相关文章：

  • 07-任务调度器的挂起和恢复
  • dask.dataframe.shuffle.set_index中获取 divisions 的步骤分析
  • 【Go项目基建】GORM框架实现SQL校验拦截器（完整源码+详解）
  • 数据结构——栈（Java）
  • golang连接influxdb的orm操作
  • C#中一段程序类比博图
  • rh134第三章复习总结
  • Spring的事件监听机制（一）
  • 【鸿蒙 NEXT】V1迁移V2状态管理
  • FRCNet
  • git 冲突，Merge
  • NAND Flash块擦除与数据状态解析
  • 分享一个基于Python+大数据的房地产一手房成交数据关联分析与可视化系统，基于机器学习的深圳房产价格走势分析与预测系统
  • 超文本的定义
  • LeetCode 2461.长度为K子数组中的最大和
  • 【机器学习入门】6.2 朴素贝叶斯分类器详解：从理论到西瓜数据集实战
  • STM32F4芯片RS485使用记录
  • java面向对象之this关键字的内存原理
  • 【FastDDS】Layer Transport ( 05-Shared Memory Transport）
  • AI工具深度测评与选型指南 - AI工具测评框架及方法论
  • Kernel中的cgroup2介绍
  • Iconify AI：免费商用AI图标生成工具，高效解决开发图标需求
  • MySQL 基础架构（一）：SQL语句的执行之旅
  • STM32-----SPI
  • 洛谷 P1591 阶乘数码-普及-
  • DEEP THINK WITH CONFIDENCE-Meta-基于置信度的深度思考
  • Qt 基础教程合集（完）
  • swagger接口文档规范化(苍穹外卖)
  • 【微知】dmesg如何将dmesg消息查看日志等级？(dmesg -x； prefix)
  • 基于STM32智能阳台监控系统