企业官网建设流程全解析

langchain-core是LangChain生态系统的基础抽象层，它定义了整个框架赖以依赖构建的核心，本身不包含任何第三方模型集成，所以我们从langchain-core开始。

langchain-core包含以下核心模块：

安装最新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( content="1. 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_id="call_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的工具，以a=2，b=3执行它。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(content="5",tool_call_id="call_2")用来回应call_1有什么问题?

答案：说明AIMessage调用工具的tool_calls发生错误，需要纠正，工具执行发生错误。

3、完整工具流

HumanMessage：2+3等于多少？ ↓ AIMessage：请求调用add(a=2,b=3),id=call_1 ↓ 程序：实际执行add(2,3) ↓ ToolMessage:结果为5，tool_call_id=call_1 ↓ AIMessage:2+3=5

注意：工具结果不是模型最后的回答。模型读取ToolMessage后，才可能生成给用户看的回答，也可能继续请求别的工具。

用langchain-core构造上面这一轮消息，并正确执行工具：

from langchain_core.messages import ( HumanMessage, AIMessage, ToolMessage, ) def add(a: int, b: int) -> int: return a + b messages = [ HumanMessage(content="2 + 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( content=str(result), tool_call_id=call["id"], ) ) # 这里同样模拟模型读完工具结果后的最终回答。 messages.append(AIMessage(content="2 + 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}."), ]) value=prompt.invoke({ "subject":"数学", "topic":"加法", }) for message in value.to_messages(): print(type(message).__name__,message.content)

1、ChatPromptTemplate：结构化可复用模板

理解它的用法：聊天模板+本次传入的信息->将内容组织好->交给LLM

from langchain_core.prompts import ChatPromptTemplate prompt = ChatPromptTemplate.from_messages([ ("system", "你是一位耐心的数学老师。"), ("human", "请解释：{question}"), ])

这个用法可以用下面这个表格概括：

所以在这里{"human","请解释:{question}"}还不是用户实际说过的话，而是“如何制作一条用户消息”的规则。from_messages()会按照列表顺序组织这些规则。

例题：上面的prompt已经包含一道具体的数学题了吗？

答案：没有。{question}还没有被填上。

2、invoke()是"填空"

现在填入一道题：

value=prompt.invoke({ "question":"2+3为什么等于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", optional=True), ("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() clean=RunnableLambda(clean_name)

clean的作用很简单：去掉名字两端的空格，再转为大写。

例题：

invoke(x)：处理一个输入->一个结果

result = clean.invoke(" xiaoming ") print(result) #答案 XIAO MING

batch([x,y])：处理多个独立输入->结果列表

result=clean.batch([" li ", "wang "]) print(result) #答案 ["LI","WANG"]

stream(x):逐块读取结果->通过循环读取结果块

for chunk in clean.stream(" zhang "): print("收到一块：", chunk) #答案 收到一块： ZHANG

ainvoke(x)：在异步代码里处理一个输入，要用await取得结果

async def main(): result = await clean.ainvoke(" chen ") print(result) asyncio.run(main()) #答案 CHEN

7、管道符 |

它表示：左边处理完，把结果交给右边

from langchain_core.runnables import RunnableLambda def add_one(x): print("加1收到：",x) return x+1 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；因为(5+1)*2=12

习题：如果把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 ")) # HELLO

8、并行组合RunnableParallel

把同一个输入分别交给几个步骤，最后放进同一个字典中。

运行这个完整程序：

from langchain_core.runnables import RunnableLambda,RunnableParallel plus_one = RunnableLambda(lambda x:x+1) times_two = RunnableLambda(lambda x:x*2) parallel=RunnableParallel( add=plus_one, multiply=times_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( total=lambda data: data["price"] * data["quantity"] ) print(add_total.invoke({"price": 10, "quantity": 3})) # {'price': 10, 'quantity': 3, 'total': 30}

10、StrOutputParser

from 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(content="5")经过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_1

13、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的核心知识点基本就这些了