使用 LangChain 构建 AI 应用

[snxq]

这是遵循 LangChain 官方文档进行实践的学习项目。本篇是该系列的第一篇，简单介绍如何通过 LangChain 来构建一个完整的包含记忆功能的聊天机器人。

一、配置 Python 环境

# 使用 brew 安装 asdf（asdf 是一个环境多版本管理工具）
brew install asdf
# 使用 asdf 安装 python 
asdf install python 3.13.7
# 设置 python 当前版本为 3.13.7
asdf set -u python 3.13.7
# 使用当前版本的 python 在项目根目录新建虚拟环境
python -m venv .env
# 进入到虚拟环境中
source ./.env/bin/activate
# 安装 langchain 相关依赖包
pip install "langgraph>0.2.27" langchain-core
pip install -Uq "langchain[openai]"
pip install langchain-ollama langchain-community transformers

二、使用 ollama 搭建本地大模型

# 使用 deepseek-r1 模型
ollama pull deepseek-r1:latest

三、开始构建聊天机器人

# 初始化聊天大模型
from langchain.chat_models import init_chat_model

model = init_chat_model("deepseek-r1", model_provider="ollama")

# 定义对话状态 Schema
# 使用 TypedDict 定义 State
#    messages 为对话的消息序列，'add_messages' 声明为 追加合并的策略，用于langgraph.graph.StateGraph 读取。
#    language 为声明的语言
from typing import Annotated, TypedDict, Sequence
from langgraph.graph import add_messages
from langchain_core.messages import BaseMessage

class State(TypedDict):
	messages: Annotated[Sequence[BaseMessage], add_messages]
	language: str

# 预定义了一个带有固定 system 短句的提示语.
# MessagesPlaceholder 的含义是将历史消息代入进去，形成完整的 prompt.
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

prompt_template = ChatPromptTemplate.from_messages([
	(
		"system",
		"你是一个非常厉害的程序员, 擅长使用 golang 来解决问题",
	),
	MessagesPlaceholder(variable_name="messages"),
])

# 定义一个消息消息修剪器，用于控制喂给大模型的上下文长度过长，并且可以设置最大 token 的数量。
# trim_messages 配合 MessagesPlaceholder, 对其中的 messages 进行修剪。
from langchain_core.messages import trim_messages

trimmer = trim_messages(
	# messages 占用的 tokens 数量不能超过 1000
	# AI 实战建议: 预算要留出空间给系统提示、固定模板和模型输出。例如模型上下文是 8k，你可能给历史留 2k~4k，再给系统提示和本轮输入留一部分，还要给模型输出留余量。
	max_tokens=1000,
	# 采用 保留最近消息优先 的策略。即从对话尾部开始叠加，超过部分从旧消息开始裁剪。
	strategy="last",
	# model 为上述定义的 model 变量, 这里指 token 的数量由 model 来统计。
	token_counter=model,
	# 保留 system 提示语
	include_system=True,
	# 不允许消息截断，要裁剪消息也是一整条消息裁剪。
	allow_partial=False, 
	# 对话轮次起点设置为 人类 消息。
	start_on="human",
)

# 定义模型调用函数。将整个消息列表修剪后，传入到 prompt 模板中，进行大模型调用。
def call_model(state: State):
	trimmed_messages = trimmer.invoke(state["messages"])
	prompts = prompt_template.invoke({
		"messages": trimmed_messages,
	})
	response = model.invoke(prompts)
	return {"messages": response}

# 定义工作流图，使用 langgraph 提供的 有状态工作流编排器 StateGraph。
# 对话代理的各个步骤抽象成图中的“节点”，节点之间用“边”连接；状态在节点间流动并被合并，从而构成可控的推理流程。
from langgraph.graph import START, StateGraph
from langgraph.checkpoint.memory import MemorySaver

# State 指上面定义的对话状态 State，也就是 TypedDict
workflow = StateGraph(state_schema=State)
# 定义图的起点
workflow.add_edge(START, "model")
# 注册名为 "model" 的节点，并关联到业务函数 call_model：它封装了“整理上下文 → 拼接 Prompt → 调模型 → 返回 AI 消息”的核心逻辑。
workflow.add_node("model", call_model)

# 编译阶段会冻结图结构、装配 reducers、准备检查点逻辑。
# 这里检查点使用内存，也就是说程序重新运行后，保存的信息也会丢失。正式环境可以将消息持久化到其他地方。
app = workflow.compile(checkpointer=MemorySaver())

# 最后是与用户交互的部分。
from langchain_core.messages import AIMessage, HumanMessage

# thread_id 在使用任意检查点器(比如 MemorySaver) 时都用于区分会话线程。
config = {"configurable": {"thread_id": "abcd123"}}
while True:
	user_input = input("> User: ")
	if user_input == "exit":
		break
	# 将用户输入转化为 HumanMessage 的列表, HumanMessage 为 BaseMessage 的子类。
	input_messages = [HumanMessage(content=user_input)]
	# 使用 app.stream 调用流式返回
	for chunk, metadata in app.stream(
		{"messages": input_messages, "language": "中文"},
		config=config,
		stream_mode="messages"
	):
		if isinstance(chunk, AIMessage):
			print(chunk.content, end="")