1. 引言为什么需要 LangChain Core APILangChain 是一个强大的框架用于构建基于大语言模型LLM的应用程序。而LangChain Core API是 LangChain 的核心基础库它提供了所有高级组件如 LangChain、LangGraph、LangServe所依赖的底层抽象和接口。掌握 Core API意味着你真正理解了 LangChain 的设计哲学和运行机制而不仅仅是会调用几个高级封装。本文将带你从零开始极其详细地学习 LangChain Core API 的每一个核心模块包括消息模型、提示模板、输出解析器、可运行对象Runnable、回调系统、工具调用、文档加载与处理、向量存储与检索、以及内存管理。每个模块都会配有完整的代码示例和原理说明。2. 环境准备与安装在开始之前请确保你的 Python 版本在 3.9 以上并安装以下依赖pip install langchain-core langchain-openai langchain-community # 如果你使用其他模型提供商请安装对应的包例如 # pip install langchain-anthropic langchain-google-genai设置你的 API 密钥以 OpenAI 为例import os os.environ[OPENAI_API_KEY] your-api-key-here3. 消息模型Message TypesLangChain Core API 中所有与大语言模型的交互都通过消息Message对象来表示。这是最基础的抽象理解它至关重要。3.1 消息类型体系LangChain 定义了多种消息类型每种类型对应对话中的不同角色SystemMessage系统消息用于设定 AI 助手的角色和行为。通常放在对话最开头。HumanMessage人类用户发送的消息。AIMessageAI 模型返回的消息。FunctionMessage函数调用返回的结果消息用于工具调用场景。ToolMessage工具调用返回的结果消息FunctionMessage 的替代推荐使用。ChatMessage自定义角色的消息当上述类型不满足时使用。from langchain_core.messages import ( SystemMessage, HumanMessage, AIMessage, ToolMessage, ChatMessage, ) 创建各种消息 system_msg SystemMessage(content你是一个专业的 Python 编程助手。) human_msg HumanMessage(content请用 Python 写一个快速排序算法。) ai_msg AIMessage(content好的以下是快速排序的实现\n\npython\ndef quick_sort(arr):\n ...\n) tool_msg ToolMessage(content函数执行结果为 42, tool_call_idcall_123) print(system_msg) print(human_msg) print(ai_msg) print(tool_msg)3.2 消息的属性和方法每个消息对象都有以下核心属性content消息的文本内容字符串。type消息类型字符串如 system、human、ai。additional_kwargs附加参数用于存储模型返回的额外信息如 token 使用量。response_metadata响应元数据通常由模型提供商返回。id消息的唯一标识符UUID 字符串。# 查看消息属性 msg AIMessage( content你好, additional_kwargs{usage: {prompt_tokens: 10, completion_tokens: 5}}, response_metadata{model: gpt-4, finish_reason: stop}, ) print(f内容: {msg.content}) print(f类型: {msg.type}) print(f附加参数: {msg.additional_kwargs}) print(f响应元数据: {msg.response_metadata}) print(f消息 ID: {msg.id})3.3 消息列表与对话历史在实际应用中我们通常使用消息列表List[BaseMessage]来表示完整的对话历史from langchain_core.messages import BaseMessage 构建对话历史 messages: list[BaseMessage] [ SystemMessage(content你是一个友好的 AI 助手。), HumanMessage(content今天天气怎么样), AIMessage(content抱歉我无法获取实时天气信息。), HumanMessage(content那你能做什么), ] 遍历消息 for msg in messages: print(f[{msg.type}] {msg.content[:50]}...)4. 提示模板Prompt Templates提示模板用于将用户输入和变量动态格式化为完整的提示文本或消息列表。4.1 ChatPromptTemplate最常用的提示模板它允许你定义一个消息列表模板其中可以包含占位符变量from langchain_core.prompts import ChatPromptTemplate 定义一个聊天提示模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个{role}专家。请用{language}回答用户的问题。), (human, {question}), ]) 格式化模板 formatted_messages prompt.format_messages( rolePython, language中文, question请解释什么是装饰器 ) for msg in formatted_messages: print(f[{msg.type}] {msg.content})4.2 消息占位符MessagesPlaceholder当你需要动态插入一组消息如对话历史时使用 MessagesPlaceholderfrom langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder prompt_with_history ChatPromptTemplate.from_messages([ (system, 你是一个有用的 AI 助手。), MessagesPlaceholder(variable_namehistory), (human, {input}), ]) 准备对话历史 from langchain_core.messages import AIMessage, HumanMessage history [ HumanMessage(content你好), AIMessage(content你好有什么可以帮助你的吗), ] 格式化 result prompt_with_history.format_messages( historyhistory, input请告诉我 Python 的列表推导式。 ) for msg in result: print(f[{msg.type}] {msg.content[:60]}...)4.3 字符串提示模板PromptTemplate对于简单的文本补全模型非聊天模型可以使用 PromptTemplatefrom langchain_core.prompts import PromptTemplate string_prompt PromptTemplate.from_template( 请用{language}写一段关于{topic}的代码要求代码风格{style}。 ) formatted string_prompt.format( languagePython, topic文件读写, style简洁高效 ) print(formatted)4.4 部分变量Partial Variables你可以预先填充部分变量生成一个部分应用的模板# 部分应用变量 partial_prompt ChatPromptTemplate.from_messages([ (system, 你是一个{role}专家。), (human, {question}), ]).partial(role数据科学) 现在只需要提供 question result partial_prompt.format_messages(question什么是过拟合) print(result[0].content) print(result[1].content)5. 输出解析器Output Parsers输出解析器负责将 LLM 返回的原始文本解析为结构化的数据格式。5.1 StrOutputParser最简单的解析器直接返回字符串内容from langchain_core.output_parsers import StrOutputParser parser StrOutputParser() result parser.invoke(这是一段文本内容。) print(result) # 输出: 这是一段文本内容。 print(type(result)) # class str5.2 PydanticOutputParser将 LLM 输出解析为 Pydantic 模型结构化数据from langchain_core.output_parsers import PydanticOutputParser from pydantic import BaseModel, Field from typing import List 定义数据结构 class Person(BaseModel): name: str Field(description人物的姓名) age: int Field(description人物的年龄) hobbies: List[str] Field(description人物的爱好列表) 创建解析器 parser PydanticOutputParser(pydantic_objectPerson) 获取格式指令用于提示 LLM 按格式输出 format_instructions parser.get_format_instructions() print(格式指令:) print(format_instructions) 解析 LLM 输出 llm_output {name: 张三, age: 28, hobbies: [阅读, 编程, 跑步]} parsed_result parser.invoke(llm_output) print(f解析结果: {parsed_result}) print(f姓名: {parsed_result.name}, 年龄: {parsed_result.age})5.3 CommaSeparatedListOutputParser将逗号分隔的文本解析为列表from langchain_core.output_parsers import CommaSeparatedListOutputParser list_parser CommaSeparatedListOutputParser() result list_parser.invoke(苹果, 香蕉, 橘子, 葡萄) print(result) # [苹果, 香蕉, 橘子, 葡萄]5.4 JsonOutputParser解析 JSON 格式的输出from langchain_core.output_parsers import JsonOutputParser json_parser JsonOutputParser() result json_parser.invoke({name: LangChain, version: 0.3}) print(result) # {name: LangChain, version: 0.3} print(type(result)) # class dict6. 可运行对象Runnable—— LangChain 的核心抽象Runnable 是 LangChain Core API 中最重要的概念。它定义了一个统一的接口让所有组件模型、提示模板、输出解析器、检索器等都可以通过相同的方式调用、组合和链式连接。6.1 Runnable 的基本接口每个 Runnable 对象都实现了以下核心方法invoke(input)同步调用输入单个值返回单个输出。ainvoke(input)异步版本的 invoke。batch(inputs)批量调用输入列表返回列表。abatch(inputs)异步版本的 batch。stream(input)流式调用逐个返回输出块。astream(input)异步版本的 stream。from langchain_core.runnables import RunnableLambda 创建一个简单的 Runnable def add_one(x: int) - int: return x 1 runnable RunnableLambda(add_one) 调用方式 print(runnable.invoke(5)) # 6 print(runnable.batch([1, 2, 3])) # [2, 3, 4] 流式调用 for chunk in runnable.stream([10, 20, 30]): print(chunk) # 11, 21, 316.2 管道操作符|—— 构建链Runnable 最强大的特性是可以通过|操作符进行组合形成处理管道from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_openai import ChatOpenAI 创建组件 prompt ChatPromptTemplate.from_messages([ (system, 你是一个{role}专家。), (human, {question}), ]) model ChatOpenAI(modelgpt-4, temperature0) parser StrOutputParser() 构建链prompt - model - parser chain prompt | model | parser 调用链 result chain.invoke({ role: Python, question: 请解释 Python 中的生成器。 }) print(result)6.3 RunnablePassthrough 与 RunnableParallel这两个组件用于数据传递和并行处理from langchain_core.runnables import RunnablePassthrough, RunnableParallel RunnablePassthrough原样传递数据 passthrough RunnablePassthrough() print(passthrough.invoke({key: value})) # {key: value} RunnableParallel并行执行多个 Runnable def upper_case(text: str) - str: return text.upper() def reverse_case(text: str) - str: return text[::-1] parallel RunnableParallel( upperRunnableLambda(upper_case), reverseRunnableLambda(reverse_case), ) result parallel.invoke(hello world) print(result) # {upper: HELLO WORLD, reverse: dlrow olleh}6.4 RunnableMapRunnableMap 是 RunnableParallel 的别名用于将输入映射到多个分支from langchain_core.runnables import RunnableMap 将输入拆分为多个处理分支 chain RunnableMap({ original: RunnablePassthrough(), upper: RunnableLambda(lambda x: x.upper()), length: RunnableLambda(lambda x: len(x)), }) result chain.invoke(LangChain) print(result) {original: LangChain, upper: LANGCHAIN, length: 9}6.5 RunnableLambda 与 RunnableGenerator用于将普通函数或生成器函数包装为 Runnablefrom langchain_core.runnables import RunnableLambda, RunnableGenerator 普通函数 def multiply_by_two(x: int) - int: return x * 2 runnable_func RunnableLambda(multiply_by_two) print(runnable_func.invoke(21)) # 42 生成器函数用于流式处理 def token_generator(text: str): for word in text.split(): yield word.upper() runnable_gen RunnableGenerator(token_generator) for token in runnable_gen.stream(hello world from langchain): print(token) 输出: HELLO WORLD FROM LANGCHAIN6.6 RunnableBinding用于给 Runnable 绑定额外的参数而不改变原始 Runnablefrom langchain_core.runnables import RunnableBinding 假设我们有一个模型 model ChatOpenAI(modelgpt-4) 绑定额外的参数 bound_model RunnableBinding( boundmodel, kwargs{stop: [\n, 用户]}, # 设置停止词 config{tags: [my-tag]}, # 设置标签 ) 使用绑定的模型 result bound_model.invoke(请写一首诗) print(result)6.7 配置运行时参数with_config / with_retryfrom langchain_core.runnables import RunnableConfig 配置运行时参数 chain_with_config chain.with_config( RunnableConfig( tags[production], metadata{version: 1.0}, max_concurrency5, ) ) 配置重试机制 from langchain_core.runnables import RunnableRetry chain_with_retry chain.with_retry( stop_after_attempt3, wait_exponential_jitterTrue, )6.8 事件回调astream_events用于监控链的执行过程import asyncio async def monitor_chain(): async for event in chain.astream_events( {role: Python, question: 什么是闭包}, versionv2 ): kind event[event] if kind on_chat_model_stream: content event[data][chunk].content if content: print(content, end) elif kind on_chain_start: print(f\n开始执行: {event[name]}) asyncio.run(monitor_chain())7. 回调系统Callbacks回调系统允许你在链执行的各个阶段插入自定义逻辑用于日志记录、监控、调试等。7.1 BaseCallbackHandlerfrom langchain_core.callbacks import BaseCallbackHandler from typing import Any, Dict, List class MyCallbackHandler(BaseCallbackHandler): def on_llm_start( self, serialized: Dict[str, Any], prompts: List[str], **kwargs: Any ) - None: print(fLLM 开始处理提示词数量: {len(prompts)}) def on_llm_end(self, response, **kwargs: Any) - None: print(fLLM 处理完成生成文本长度: {len(response.generations[0][0].text)}) def on_chain_start( self, serialized: Dict[str, Any], inputs: Dict[str, Any], **kwargs: Any ) - None: print(f链开始执行输入: {inputs}) def on_chain_end(self, outputs: Dict[str, Any], **kwargs: Any) - None: print(f链执行完成输出: {outputs}) 使用回调 callback_handler MyCallbackHandler() chain prompt | model | parser result chain.invoke( {role: Python, question: 什么是列表推导式}, config{callbacks: [callback_handler]} )7.2 回调管理器CallbackManagerfrom langchain_core.callbacks import CallbackManager 创建回调管理器 callback_manager CallbackManager([MyCallbackHandler()]) 在链中使用 chain prompt | model | parser result chain.invoke( {role: Python, question: 什么是装饰器}, config{callbacks: callback_manager} )8. 工具调用Tool CallingLangChain Core API 提供了标准的工具定义和调用接口让 LLM 能够调用外部函数。8.1 定义工具from langchain_core.tools import tool tool def get_current_time() - str: 获取当前时间。 from datetime import datetime return datetime.now().strftime(%Y-%m-%d %H:%M:%S) tool def calculate(expression: str) - str: 计算数学表达式。 try: result eval(expression) return f计算结果: {result} except Exception as e: return f计算错误: {str(e)} tool def search_web(query: str) - str: 搜索网络信息模拟。 return f关于{query}的搜索结果这里是一些相关信息... print(get_current_time.name) # get_current_time print(get_current_time.description) # 获取当前时间。 print(get_current_time.args) # 参数模式8.2 工具绑定到模型# 将工具绑定到模型 model_with_tools model.bind_tools([get_current_time, calculate, search_web]) 调用模型模型会自动决定是否调用工具 response model_with_tools.invoke(现在几点了) print(response) print(f工具调用: {response.tool_calls}) 如果模型决定调用工具 if response.tool_calls: for tool_call in response.tool_calls: tool_name tool_call[name] tool_args tool_call[args] print(f调用工具: {tool_name}, 参数: {tool_args})8.3 完整的工具调用循环from langchain_core.messages import ToolMessage 工具映射 tools_map { get_current_time: get_current_time, calculate: calculate, search_web: search_web, } 完整的工具调用循环 messages [ SystemMessage(content你是一个有用的助手可以使用工具来回答问题。), HumanMessage(content计算 25 * 4 100 等于多少), ] response model_with_tools.invoke(messages) messages.append(response) 处理工具调用 for tool_call in response.tool_calls: tool tools_map[tool_call[name]] tool_result tool.invoke(tool_call[args]) messages.append(ToolMessage(contenttool_result, tool_call_idtool_call[id])) 将工具结果返回给模型 final_response model_with_tools.invoke(messages) print(final_response.content)9. 文档加载与处理Document Loaders Transformers9.1 文档对象Documentfrom langchain_core.documents import Document 创建文档对象 doc Document( page_content这是文档的文本内容。, metadata{ source: example.txt, author: 张三, date: 2024-01-01, page: 1, } ) print(f内容: {doc.page_content}) print(f元数据: {doc.metadata}) print(f文档 ID: {doc.id})9.2 文本分割器Text Splittersfrom langchain_core.documents import Document from langchain_text_splitters import ( RecursiveCharacterTextSplitter, CharacterTextSplitter, TokenTextSplitter, ) 递归字符文本分割器推荐 recursive_splitter RecursiveCharacterTextSplitter( chunk_size200, # 每个块的最大字符数 chunk_overlap20, # 块之间的重叠字符数 separators[\n\n, \n, 。, , , , , ], length_functionlen, ) 准备长文本 long_text LangChain 是一个用于构建大语言模型应用的框架。 它提供了丰富的工具和抽象帮助开发者快速构建复杂的 AI 应用。 Core API 是 LangChain 的核心库提供了所有基础组件。 掌握 Core API 是深入学习 LangChain 的关键。 本文将从消息模型开始逐步介绍提示模板、输出解析器、 可运行对象、回调系统、工具调用等核心概念。 每个部分都配有详细的代码示例和原理解释。 分割文档 documents recursive_splitter.create_documents([long_text]) print(f分割后文档数量: {len(documents)}) for i, doc in enumerate(documents): print(f文档 {i1}: {doc.page_content[:50]}...)9.3 文档转换器Document Transformersfrom langchain_core.document_transformers import ( LongContextReorder, EmbeddingsRedundantFilter, ) from langchain_core.documents import Document 长上下文重排序解决 LLM 对中间内容关注度下降的问题 reorder LongContextReorder() docs [ Document(page_content最重要的信息), Document(page_content次要信息), Document(page_content最不重要的信息), ] reordered_docs reorder.transform_documents(docs) for doc in reordered_docs: print(doc.page_content)10. 向量存储与检索Vector Stores Retrievers10.1 嵌入模型Embeddingsfrom langchain_openai import OpenAIEmbeddings 创建嵌入模型 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) 生成文本嵌入 text LangChain Core API 学习 vector embeddings.embed_query(text) print(f向量维度: {len(vector)}) print(f向量前5个值: {vector[:5]}) 批量嵌入 texts [文本1, 文本2, 文本3] vectors embeddings.embed_documents(texts) print(f批量向量数量: {len(vectors)})10.2 向量存储InMemoryVectorStorefrom langchain_core.vectorstores import InMemoryVectorStore 创建内存向量存储 vector_store InMemoryVectorStore(embeddings) 添加文档 docs [ Document(page_contentLangChain 是一个 LLM 应用框架, metadata{topic: langchain}), Document(page_contentPython 是一种编程语言, metadata{topic: python}), Document(page_content向量数据库用于存储和检索向量, metadata{topic: database}), ] vector_store.add_documents(docs) 相似度搜索 results vector_store.similarity_search(LLM 框架, k2) for doc in results: print(f内容: {doc.page_content}, 相似度: {doc.metadata.get(score, N/A)}) 带分数的相似度搜索 results_with_score vector_store.similarity_search_with_score(编程语言, k2) for doc, score in results_with_score: print(f内容: {doc.page_content}, 分数: {score})10.3 检索器Retrieversfrom langchain_core.retrievers import BaseRetriever from typing import List 将向量存储转换为检索器 retriever vector_store.as_retriever( search_typesimilarity, # 或 mmr (最大边际相关性) search_kwargs{k: 2} ) 使用检索器 retrieved_docs retriever.invoke(什么是 LangChain) for doc in retrieved_docs: print(f检索结果: {doc.page_content}) 自定义检索器 class MyCustomRetriever(BaseRetriever): def _get_relevant_documents(self, query: str) - List[Document]: # 自定义检索逻辑 return [ Document(page_contentf关于{query}的检索结果1), Document(page_contentf关于{query}的检索结果2), ] custom_retriever MyCustomRetriever() results custom_retriever.invoke(测试查询) print(results)11. 内存管理Memory内存模块用于在多次对话之间保持状态和上下文。11.1 BaseMemory 与对话内存from langchain_core.memory import BaseMemory from langchain.memory import ConversationBufferMemory 对话缓冲内存 memory ConversationBufferMemory( return_messagesTrue, # 返回消息对象而非字符串 memory_keyhistory, # 在提示模板中使用的变量名 ) 保存上下文 memory.save_context( {input: 你好}, {output: 你好有什么可以帮助你的吗} ) memory.save_context( {input: 请介绍一下 Python。}, {output: Python 是一种高级编程语言...} ) 加载内存 loaded_variables memory.load_memory_variables({}) print(loaded_variables[history])11.2 对话摘要内存from langchain.memory import ConversationSummaryMemory 摘要内存对长对话进行压缩 summary_memory ConversationSummaryMemory( llmmodel, return_messagesTrue, memory_keyhistory, ) summary_memory.save_context( {input: 你好}, {output: 你好我是 AI 助手。} ) summary_memory.save_context( {input: 今天天气真好。}, {output: 是的适合出去散步。} ) 查看摘要 summary summary_memory.load_memory_variables({}) print(summary[history])11.3 将内存集成到链中from langchain_core.runnables import RunnablePassthrough from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder 带内存的提示模板 prompt_with_memory ChatPromptTemplate.from_messages([ (system, 你是一个有用的 AI 助手。), MessagesPlaceholder(variable_namehistory), (human, {input}), ]) 创建带内存的链 def get_memory(): return ConversationBufferMemory(return_messagesTrue, memory_keyhistory) memory get_memory() 构建链 chain ( RunnablePassthrough.assign( historylambda x: memory.load_memory_variables({})[history] ) | prompt_with_memory | model | parser ) 第一次对话 result1 chain.invoke({input: 你好}) memory.save_context({input: 你好}, {output: result1}) print(f第一次: {result1}) 第二次对话带历史 result2 chain.invoke({input: 我刚才说了什么}) memory.save_context({input: 我刚才说了什么}, {output: result2}) print(f第二次: {result2})12. 完整实战构建一个 RAG 问答系统下面我们将综合运用以上所有知识构建一个完整的 RAG检索增强生成问答系统from langchain_core.documents import Document from langchain_core.prompts import ChatPromptTemplate from langchain_core.output_parsers import StrOutputParser from langchain_core.runnables import RunnablePassthrough, RunnableParallel from langchain_core.vectorstores import InMemoryVectorStore from langchain_openai import OpenAIEmbeddings, ChatOpenAI from langchain_text_splitters import RecursiveCharacterTextSplitter 1. 准备文档 documents [ Document( page_contentLangChain 是一个用于构建大语言模型应用的框架。它提供了丰富的工具和抽象 帮助开发者快速构建复杂的 AI 应用。Core API 是 LangChain 的核心库 提供了所有基础组件包括消息模型、提示模板、输出解析器等。, metadata{source: langchain_intro.md} ), Document( page_contentRunnable 是 LangChain Core API 中最重要的概念。它定义了一个统一的接口 让所有组件都可以通过相同的方式调用、组合和链式连接。 Runnable 支持 invoke、batch、stream 等多种调用方式。, metadata{source: runnable.md} ), Document( page_content向量数据库用于存储和检索向量嵌入。常见的向量数据库包括 Chroma、 Pinecone、Weaviate 等。InMemoryVectorStore 是一个轻量级的内存向量存储实现。, metadata{source: vectorstore.md} ), ] 2. 文本分割 text_splitter RecursiveCharacterTextSplitter( chunk_size200, chunk_overlap20, ) chunks text_splitter.split_documents(documents) 3. 创建向量存储和检索器 embeddings OpenAIEmbeddings(modeltext-embedding-3-small) vector_store InMemoryVectorStore(embeddings) vector_store.add_documents(chunks) retriever vector_store.as_retriever(search_kwargs{k: 2}) 4. 创建提示模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个专业的 AI 助手。请基于以下上下文信息回答用户的问题。\n\n上下文\n{context}), (human, {question}), ]) 5. 创建模型和解析器 model ChatOpenAI(modelgpt-4, temperature0) parser StrOutputParser() 6. 构建 RAG 链 def format_docs(docs): return \n\n.join(doc.page_content for doc in docs) rag_chain ( RunnableParallel( contextretriever | format_docs, questionRunnablePassthrough(), ) | prompt | model | parser ) 7. 测试 RAG 系统 questions [ 什么是 LangChain, Runnable 有哪些调用方式, InMemoryVectorStore 是什么, ] for question in questions: print(f\n问题: {question}) answer rag_chain.invoke(question) print(f回答: {answer}) print(- * 50)13. 总结与进阶学习路径通过本文的学习你已经掌握了 LangChain Core API 的以下核心模块消息模型SystemMessage、HumanMessage、AIMessage 等消息类型及其属性。提示模板ChatPromptTemplate、MessagesPlaceholder、PromptTemplate 及部分变量。输出解析器StrOutputParser、PydanticOutputParser、JsonOutputParser 等。可运行对象Runnableinvoke、batch、stream、管道操作符、RunnableParallel、RunnableLambda 等。回调系统BaseCallbackHandler、CallbackManager。工具调用tool 装饰器、bind_tools、完整的工具调用