langchain-core是LangChain生态系统的基础抽象层它定义了整个框架赖以依赖构建的核心本身不包含任何第三方模型集成所以我们从langchain-core开始。langchain-core包含以下核心模块模块说明Runnables统一调用协议与语法LCELLanguage Models语言模型与聊天模型的基础接口Prompts提示模板及相关工具Messages聊天模型的输入/输出消息类型Output Parsers解析模型输出Documents文档抽象用于检索工作流Retrievers文档检索器接口Vector Stores向量存储接口Embeddings嵌入模型抽象Tools工具积累与定义CachesLLM调用缓存机制Callbacks回调处理器与管理Rate limiters速度限制工具安装最新langchain-core包pip install -U langchain-core一、一次对话的数据结构基础导入from langchain_core.messages import( SystemMessage, HumanMessage, AIMessage, ToolMessage, )SystemMessage系统指令# 创建系统消息 system_msg SystemMessage( content你是一名RAG技术面试官只考察检索增强生成相关知识。回答必须分点每题不超过3条。不知道就说超出考察范围。 ) # 打印完整运行时结构 print( SystemMessage 完整结构 ) print(system_msg.model_dump()) #输出 SystemMessage 完整结构 {type: system, content: 你是一名RAG技术面试官只考察检索增强生成相关知识。回答必须分点每题不超过3条。不知道就说超出考察范围。}HumanMessage用户输入# 创建用户消息 human_msg HumanMessage( content讲一下RAG和微调的区别, # 可选附加元数据仅程序内部追踪用不会发给模型 metadata{user_id: 001, timestamp: 2026-09-27} ) print( HumanMessage 完整结构 ) print(human_msg.model_dump()) HumanMessage 完整结构 {type: human, content: 讲一下RAG和微调的区别, metadata: {user_id: 001, timestamp: 2026-09-27}}AIMessage模型输出场景一ai_msg AIMessage( content1. RAG是外挂外部资料成本低实时更新2. 微调是更新模型参数成本高知识固化3. 两者常结合使用。 ) print( AIMessage普通对话完整结构 ) print(ai_msg.model_dump()) AIMessage普通对话完整结构 {type: ai, content: 1. RAG是外挂外部资料成本低实时更新2. 微调是更新模型参数成本高知识固化3. 两者常结合使用。, tool_calls: []}场景二ai_msg_with_tool AIMessage( content, # 文本内容可以为空模型先决定调用工具 tool_calls[ { name: search_rag_doc, args: {query: RAG切分策略}, id: call_001 # 工具调用唯一ID用于对应后续ToolMessage }, ], ) print( AIMessage带工具调用完整结构 ) print(ai_msg_with_tool.model_dump()) AIMessage带工具调用完整结构 {type: ai, content: , tool_calls: [{name: search_rag_doc, args: {query: RAG切分策略}, id: call_001}]}ToolMessage工具结果tool_msg ToolMessage( content常见切分策略固定字符切分、递归字符切分、语义切分、结构感知切分。, tool_call_idcall_001 # 必须和对应AIMessage里的工具调用ID完全一致 ) print( ToolMessage 完整结构 ) print(tool_msg.model_dump()) ToolMessage 完整结构 {type: tool, content: 常见切分策略固定字符切分、递归字符切分、语义切分、结构感知切分。, tool_call_id: call_001}注意模型请求调用工具时调用信息在AIMessage.tool_calls工具执行用相同的tool_call_id返回ToolMessage。1、AIMessage.tool_calls里装的是什么它通常是一个列表。每项包含工具名name、参数args和本次调用的标识id例题看到下面的信息接下来该做什么AIMessage( content, tool_calls[ { name:add, args:{a:2,b:3}, id:call_1, }, ], )答案找到名为add的工具以a2b3执行它。content为空很正常因为此时模型要调用工具还没有给出最后的回答。习题tool_calls有两项能只处理第一项就直接让模型回答吗答案通常不行。应该处理这一轮每个工具的调用并且返回对应结果。2、ToolMessage为什么需要tool_call_id?因为同一条AIMessage可以请求多个工具。tool_call_id告诉模型“这个结果属于哪一次调用。”它的值必须对应请求中的id。例题模型同时请求call_1:add(2,3)和call_2:add(10,20),得到5和30.如何对应答案结果5的ToolMessage.tool_call_id是call_1;结果30的是call_2。习题ToolMessage(content5,tool_call_idcall_2)用来回应call_1有什么问题?答案说明AIMessage调用工具的tool_calls发生错误需要纠正工具执行发生错误。3、完整工具流HumanMessage23等于多少 ↓ AIMessage请求调用add(a2,b3),idcall_1 ↓ 程序实际执行add(2,3) ↓ ToolMessage:结果为5tool_call_idcall_1 ↓ AIMessage:235注意工具结果不是模型最后的回答。模型读取ToolMessage后才可能生成给用户看的回答也可能继续请求别的工具。用langchain-core构造上面这一轮消息并正确执行工具from langchain_core.messages import ( HumanMessage, AIMessage, ToolMessage, ) def add(a: int, b: int) - int: return a b messages [ HumanMessage(content2 3 等于多少), ] # 这里手动模拟模型发出的请求真实应用中它由模型返回。 request AIMessage( content, tool_calls[{ name: add, args: {a: 2, b: 3}, id: call_1, }], ) messages.append(request) for call in request.tool_calls: if call[name] add: result add(**call[args]) messages.append( ToolMessage( contentstr(result), tool_call_idcall[id], ) ) # 这里同样模拟模型读完工具结果后的最终回答。 messages.append(AIMessage(content2 3 5。)) for message in messages: print(message.type, message.content)二、聊天提示词组织消息from langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system,你是{subject}老师。), (human,请解释{topic}.), ]) valueprompt.invoke({ subject:数学, topic:加法, }) for message in value.to_messages(): print(type(message).__name__,message.content)1、ChatPromptTemplate结构化可复用模板理解它的用法聊天模板本次传入的信息-将内容组织好-交给LLMfrom langchain_core.prompts import ChatPromptTemplate prompt ChatPromptTemplate.from_messages([ (system, 你是一位耐心的数学老师。), (human, 请解释{question}), ])这个用法可以用下面这个表格概括写法含义system生成一条系统角色的消息human生成一条用户角色的消息{question}这里留一个待填写的位置所以在这里{human,请解释:{question}}还不是用户实际说过的话而是“如何制作一条用户消息”的规则。from_messages()会按照列表顺序组织这些规则。例题上面的prompt已经包含一道具体的数学题了吗答案没有。{question}还没有被填上。2、invoke()是填空现在填入一道题valueprompt.invoke({ question:23为什么等于5 })invoke将字典中的键值question对应模板中的{question},将内容填充进去填充完毕得到的是ChatPromptValue。用.to_messages()能看到其中具体的信息。messages value.to_messages() for message in messages: print(type(message).__name__, message.content)结果相当于SystemMessage 你是一位耐心的数学老师。 HumanMessage 请解释2 3 为什么等于 5注意这里没有AIMessage。prompt.invoke()的含义是执行模板填充工作不是调用AI回答问题。虽然两个操作都可能叫invoke但是执行的是不同对象。习题如果改为prompt.invoke({question:什么是乘法})会改变系统消息吗答案不会。只会改变含有{question}的用户消息。3、为什么不能只拼成一大段字符串你当然可以写成text 系统你是数学老师。用户什么是乘法但是这只是一段字符串。这个区别在多轮对话和工具调用时尤其重要用户说的话、AI 说的话、工具返回的结果不能随意混成一条消息。模板的价值是保留消息角色和排列顺序同时让内容可复用。4、to_messages()的返回形式 to_messages() 整体返回形式 [ SystemMessage(content你是RAG技术助手只能基于给定资料回答。), HumanMessage(content什么是文本切分), AIMessage(content我需要先检索知识库。), ToolMessage(content文本切分是把长文档拆成小段方便检索匹配。), HumanMessage(content常用策略有哪些) ]5、MessagesPlaceholder在聊天模板预留一个位置用来插入多条已经存在的消息它本身不产生消息也不保存历史更不会自动调用模型。from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import HumanMessage, AIMessage prompt ChatPromptTemplate.from_messages([ (system, 你是简洁的中文助手。), MessagesPlaceholder(history, optionalTrue), (human, {question}), ]) value prompt.invoke({ history: [ HumanMessage(content你好), AIMessage(content你好), ], question: 什么是 Runnable, }) print([m.type for m in value.to_messages()]) # [system, human, ai, human]from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from langchain_core.messages import AIMessage # 第 1 步定义消息排列方式 prompt ChatPromptTemplate.from_messages([ (system, 你是一个友好的助手。), MessagesPlaceholder(history), (human, {question}), ]) # 第 2 步第一次提问。还没有历史所以传空列表 history [] first prompt.invoke({ history: history, question: 我叫小明。, }) print(第一次准备好的消息) for message in first.to_messages(): print(message.type, :, message.content) # 第 3 步模拟助手已经回答并由我们自己保存这一轮对话 history.append(first.to_messages()[-1]) # 保存“我叫小明。” history.append(AIMessage(content你好小明。)) # 模拟助手回答 # 第 4 步第二次提问把保存的历史传给同一个模板 second prompt.invoke({ history: history, question: 我叫什么, }) print(\n第二次准备好的消息) for message in second.to_messages(): print(message.type, :, message.content)每一次执行invoke时候都会把前一次的history传入进去第一次是[]的第二次把上一次对话最后一次HumanMessage和AIMessage传入进去。6、Runnable接口RunnableLambda是langchain-core中LCEL体系的核心基础组件作用是把任意普通python函数包装成标准的Runnable接口对象从而可以接入LCEL的链式调用和Prompt、模型、解析器等其他Runnable组件拼接。from langchain_core.runnables import RunnableLambda def clean_name(name:str)-str: return name.strip().upper() cleanRunnableLambda(clean_name)clean的作用很简单去掉名字两端的空格再转为大写。例题invoke(x)处理一个输入-一个结果result clean.invoke( xiaoming ) print(result) #答案 XIAO MINGbatch([x,y])处理多个独立输入-结果列表resultclean.batch([ li , wang ]) print(result) #答案 [LI,WANG]stream(x):逐块读取结果-通过循环读取结果块for chunk in clean.stream( zhang ): print(收到一块, chunk) #答案 收到一块 ZHANGainvoke(x)在异步代码里处理一个输入要用await取得结果async def main(): result await clean.ainvoke( chen ) print(result) asyncio.run(main()) #答案 CHEN7、管道符 |它表示左边处理完把结果交给右边from langchain_core.runnables import RunnableLambda def add_one(x): print(加1收到,x) return x1 def multiply_two(x): print(乘2收到,x) return x*2 step1 RunnableLambda(add_one) step2 RunnableLambda(multiply_two) chain step1 | step2 answer chain.invoke(3) print(最终结果,answer) #结果 加 1 收到 3 乘 2 收到 4 最终结果 8例题chain.invoke(5)的结果是多少答案12因为(51)*212习题如果把chain改成step2 | step1结果还是8吗答案不是而是7代码题把“ hello ”先去掉两端空格再转成大写得到“HELLO”。答案from langchain_core.runnables import RunnableLambda strip_spaces RunnableLambda(lambda text: text.strip()) make_upper RunnableLambda(lambda text: text.upper()) chain strip_spaces | make_upper print(chain.invoke( hello )) # HELLO8、并行组合RunnableParallel把同一个输入分别交给几个步骤最后放进同一个字典中。运行这个完整程序from langchain_core.runnables import RunnableLambda,RunnableParallel plus_one RunnableLambda(lambda x:x1) times_two RunnableLambda(lambda x:x*2) parallelRunnableParallel( addplus_one, multiplytimes_two ) print(parallel.invoke(3)) #结果 {add: 4, multiply: 6}9、RunnablePassthrough保留原来的数据同时添加一个新的计算结果from langchain_core.runnables import RunnablePassthrough keep RunnablePassthrough() print(keep.invoke(你好)) # 你好 #结果 你好from langchain_core.runnables import RunnablePassthrough add_total RunnablePassthrough.assign( totallambda data: data[price] * data[quantity] ) print(add_total.invoke({price: 10, quantity: 3})) # {price: 10, quantity: 3, total: 30}10、StrOutputParserfrom langchain_core.messages import AIMessage from langchain_core.output_parsers import StrOutputParser message AIMessage(content你好小明。) parser StrOutputParser() result parser.invoke(message) print(result) # 你好小明。 print(type(result)) # class langchain_core.messages.base.TextAccessor #结果 你好小明。 class langchain_core.messages.base.TextAccessor给出了可当作字符串使用的文本结果解析器的方便之处在于能作为统一步骤接进 Runnable 链例题AIMessage(content5)经过StrOutputParser后得到数字5吗答案不是得到可当作字符串使用的文本结果5习题AIMessage(content完成)经过解析后还能从结果上读取tool.calls吗答案不能。解析后的结果是可当作字符串使用的文本结果不再是原来的AIMessage注意注意注意如果某段代码要求严格内置str再用str(result)转换即可11、JsonOutputParser上一步的StrOutputParser是取出文本这一步是要把符合JSON格式的文本解析成Python数据。from langchain_core.messages import AIMessage from langchain_core.output_parsers import JsonOutputParser message AIMessage( content{name: 小明, passed: true} ) parser JsonOutputParser() result parser.invoke(message) print(result) print(type(result))12、tool把普通函数封装成工具from langchain_core.tools import tool from langchain_core.messages import AIMessage # 1. 定义工具 tool def add(a: int, b: int) - int: 计算两个整数的和。 return a b # 2. 直接调用工具 print(add.invoke({a: 2, b: 3})) # 5 # 3. 手动模拟一条模型发出的工具调用请求 request AIMessage( content, tool_calls[{ name: add, args: {a: 2, b: 3}, id: call_1, }], ) # 4. 把这次请求交给工具执行 result add.invoke(request.tool_calls[0]) print(type(result).__name__) # ToolMessage print(result.content) # 5 print(result.tool_call_id) # call_113、Document正文内容来源等附加信息from langchain_core.documents import Document doc Document( page_content退款申请需要在购买后七天内提交。, metadata{ source: 售后政策.md, page: 1, }, ) print(doc.page_content) print(doc.metadata) #输出 退款申请需要在购买后七天内提交。 {source: 售后政策.md, page: 1}langchain-core的核心知识点基本就这些了