Langchain 的基本用法
发布于 1 天前 作者 pangguoming 10 次浏览 来自 LangChain

langchain 的各种花式写法实在是太多,直接看复杂代码让人头晕。文本基于 0.3 版本记录一下基础语法,重点是 ChatModel、PromptTemplate 和 OutputParser,以及如何通过这三种对象搭基础 RunnableSequence,进行一次完整的 LLM 调用。这种小功能单元是 Agent 系统的基础结构。

关于 langchain 的更多细节,可以参考官方网站:Introduction | ️ LangChain LangChain Python API Reference 【我基于的是最新的 v0.3,从该版本起,很多传统写法已废弃,详见 Deprecations and Breaking Changes | ️ LangChain】

1. 开发环境

langchain 的核心是 langchain-core 和 langchain,如果用到一些社区功能,还要加上 langchain-community,搭建 Agent 用的是 langgraph。这些包可以先全部装上。

pip install langchain-core langchain langchain-community langgraph

2. 加载 LLM 模型

langchain 中的 LLM 是通过 API 来访问的,目前支持将近 80 种不同平台的 API,详见 Chat models | ️ LangChain

这些模型都是会话模型 ChatModel,因此命名都以前缀 Chat- 开始,比如 ChatOPenAI 和 ChatDeepSeek 等。这些模型分两种,一种由 langchain 官方提供,需要安装对应的依赖包(比如 langchain-openai),一种由社区提供,在 langchain-community 中实现。具体的安装和使用方式,可以从上面的链接中查看。

无论哪种模型,都需要配置密钥 api_key 和模型版本 model_name,以阿里云的百炼平台为例:

import os
# 通过环境变量配置密钥.
os.environ["DASHSCOPE_API_KEY"] = "自己的密钥"

from langchain_community.chat_models.tongyi import ChatTongyi
llm = ChatTongyi(model_name="qwen-max") # 也可以通过参数 api_key="..." 配置密钥.

# 调用模型.
llm.invoke("你好")
如果是由 langchain 提供的官方模型,还可以使用统一的初始化接口 init_chat_model:

from langchain.chat_models import init_chat_model

os.environ["OPENAI_API_KEY"] = "XXXXXXXXXXXXXXXXXX"
gpt_4o = init_chat_model("gpt-4o", model_provider="openai")

3. RunnableSequence

这里调用 llm 的方式是 Invoke(),这是 Runnable 对象的通用方法。在 langchain 里,几乎所有的核心功能类都是 Runnable 对象。Runnable 对象可以看作是一种“管道”或者“工作流”,它们拥有统一的调用接口,我们就不需要去记忆每种类型的入口函数到底叫什么名字,总之都是 invoke()。Runnable 对象的详细手册可以参看 Runnable interface | ️ LangChain,这里只介绍高频使用的:

invoke:把输入转换为输出,即调用函数; batch:并行计算; stream:流式处理; 每一种 Runnable 对象,都已经预置了自身的输入类型和输出类型,即 invoke() 函数的输入和输出。可以通过成员函数 get_input_schema / get_input_jsonschema、get_output_schema / get_output_jsonschema、config_schema / get_config_jsonschema 查询。比如:

llm.get_input_jsonschema() invoke、batch、stream,都有对应的异步版本 ainvoke、abatch、astream,大大减轻了开发工作量。只要开发了一个 Runnable 对象,就能轻松变成并行版 batch() 和异步版 ainvoke()。

多个 Runnable 可以顺序拼接成“链式”的 RunnableSequence,比如:

chain = llm | xxx | yyy | zzz

拼接成的 chain 也是 Runnable 对象,能 invoke(),能 batch(),能 stream()。这种写法叫做 LCEL(LangChain Expression Language),看习惯就好了。

4. 基础消息类型 Message

Messages | ️ LangChain 继续说回 llm.invoke()。这个函数返回的是一个 AIMessage 对象。Message 是langchain 中 ChatModel 的核心输入输出类型,根据 role 可以分为 SystemMessage、HumanMessage、AIMessage 和 ToolMessage。每种 Message 的核心是内部属性 content。

当直接给 llm 输入字符串时,默认解析为 HumanMessage 类型。这里有很多种花式写法,以下的四种等价:

from langchain_core.messages import HumanMessage

llm.invoke("你好呀")
llm.invoke(HumanMessage(content="你好呀"))
# 以下两种写法只支持列表
llm.invoke([{"role": "user", "content": "你好呀"})
llm.invoke([("user", "你好呀")])

如果要传入会话历史,就使用 Message 列表的形式:

from langchain_core.messages import SystemMessage, HumanMessage

messages = [SystemMessage(content="你是我的私人助理"), HumanMessage(content="你好呀")]
llm.invoke(messages)

5. 输入模板 PromptTemplate

Prompt Templates | ️ LangChain 5.1. 带占位的基础模板 实际调用 llm 时,每次请求需要在固定 prompt 中拼接用户定制项,这就需要 PromptTemplate 来实现。langchain 用的占位符是 {},和 Python 的 f-string 一样。

from langchain_core.prompts import PromptTemplate

# 模板定义.
prompt_template = PromptTemplate.from_template("今天是星期{day}")
# 返回类型.
# PromptTemplate(input_variables=['day'], template='今天是星期{day}')
PromptTemplate 的两个核心属性是 input_variables 和 template,因此也可以通过直接赋值来初始化:

prompt_template = PromptTemplate(template='今天是星期{day}', input_variables=['day'])
PromptTemplate 也是 Runnable 对象,因此也使用 invoke() 作为入口。输入是参数字典,输出是 StringPromptValue 对象:

prompt_template.invoke({"day": "三"})
# 返回类型.
# StringPromptValue(text='今天是星期三')

还有一种模板是 ChatPromptTemplate,可以处理 Message 列表:

from langchain_core.prompts import ChatPromptTemplate

prompt_template = ChatPromptTemplate([
    ("system", "你是我的私人助理"),
    ("user", "今天是星期{day}")
])
# 返回类型.
# ChatPromptTemplate(input_variables=['day'], messages=[SystemMessagePromptTemplate(prompt=PromptTemplate(input_variables=[], template='你是我的私人助理')), HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['day'], template='今天是星期{day}'))])

从返回类型可以看到,ChatPromptTemplate 内部实际是一个 PromptTemplate 的列表。调用 invoke() 执行后,会转化为 ChatPromptValue,它的内部是 Message 列表:

prompt_template.invoke({"day": "三"})
# 返回类型.
# ChatPromptValue(messages=[SystemMessage(content='你是我的私人助理'), HumanMessage(content='今天是星期三')])

如果要在prompt中变动的一条完整 Message,可以使用消息占位符 MessagesPlaceholder,或者直接将 role 指定为 placeholder。两种写法等价:

from langchain_core.prompts import MessagesPlaceholder

# 第一种:使用 MessagesPlaceholder 占位.
prompt_template = ChatPromptTemplate([
    ("system", "你是我的私人助理"),
    MessagesPlaceholder("msgs")
])

# 第二种:直接使用 placeholder 作为消息的 role.
prompt_template = ChatPromptTemplate([
    ("system", "你是我的私人助理"),
    ("placeholder", "{msgs}")
])

# 返回类型.
# ChatPromptTemplate(input_variables=['msgs'], input_types={'msgs': typing.List[typing.Union[langchain_core.messages.ai.AIMessage, langchain_core.messages.human.HumanMessage, langchain_core.messages.chat.ChatMessage, langchain_core.messages.system.SystemMessage, langchain_core.messages.function.FunctionMessage, langchain_core.messages.tool.ToolMessage]]}, messages=[SystemMessagePromptTemplate(prompt=PromptTemplate(input_variables=[], template='你是我的私人助理')), MessagesPlaceholder(variable_name='msgs')])

ChatPromptTemplate 同样使用 invoke 接口来调用,注意占位符 msgs 对应的必须是列表:

# 第一种写法.
prompt_template.invoke({"msgs": [HumanMessage(content="你好!")]})
# 第二种写法.
prompt_template.invoke({"msgs": [("user", "你好!")]})
# 返回类型.
# ChatPromptValue(messages=[SystemMessage(content='你是我的私人助理'), HumanMessage(content='你好!')])

invoke 之后,ChatPromptTemplate 就转化成了 ChatPromptValue,也就是 Message 的列表。

5.2. Message拼接转模板 当两个 Message 使用 + 进行拼接时,会自动转换为 ChatPromptValue。因此,定义模板也可以直接往 Message 上去拼字符串。

# Message 直接拼接.
SystemMessage(content="你是我的私人助理") + HumanMessage(content="今天的天气不太好")
# 返回类型.
# ChatPromptTemplate(input_variables=[], messages=[SystemMessage(content='你是我的私人助理'), HumanMessage(content='今天的天气不太好')])
# 字符串拼接.
SystemMessage(content="你是我的私人助理") + "今天是星期{day}"
# 返回类型.
# ChatPromptTemplate(input_variables=['day'], messages=[SystemMessage(content='你是我的私人助理'), HumanMessagePromptTemplate(prompt=PromptTemplate(input_variables=['day'], template='今天是星期{day}'))])

5.3. 模板的部分格式化化 PromptTemplate 和 ChatPromptTemplate 的一个重要特点,是支持部分占位符的格式化。这一功能在复杂系统中尤其有用。比如,基类模板共有 a 和 b 两个占位符,派生模板 A 可以只使用 a,派生模板 B 可以只使用 b。部分格式化通过 partial() 来实现,先来看一个最简单的例子:

prompt_template = PromptTemplate.from_template("现在是 {year} 年 {month} 月")
# 输出.
# PromptTemplate(input_variables=['month', 'year'], template='现在是 {year} 年 {month} 月')
# 部分格式化.
partial_prompt = prompt_template.partial(year="2025")
# PromptTemplate(input_variables=['month'], partial_variables={'year': '2025'}, template='现在是 {year} 年 {month} 月')
在调用了 partial() 后,PromptTemplate 的 input_variables 由 ['month', 'year'] 减少到 ['month'],而 partial_variables 中增加了 {'year': '2025'}。

partial_prompt.invoke({"month": "3"})
# 输出.
# StringPromptValue(text='现在是 2025 年 3 月')

借着这个例子,介绍一些 langchain 中关于类型实例化的一种通用写法。比如上面的代码,当你看到输出内容是StringPromptValue(text=‘现在是 2025 年 3 月’),这往往就是该类型的是初始化方法,可以通过原命令直接构造。比如:

from langchain_core.prompt_values import StringPromptValue

# 构造.
StringPromptValue(text='现在是 2025 年 3 月')
# 输出. 一模一样.
# StringPromptValue(text='现在是 2025 年 3 月')

了解这种写法之后,看各种类型会觉得清晰一些。继续说回模板的部分参数化。如果不直接调用 partial(),而是直接做 partial_variables 的赋值,那么也可以将占位符指定为函数:

from datetime import datetime

def _get_datetime():
    now = datetime.now()
    return now.strftime("%m/%d/%Y, %H:%M:%S")

# 直接定义模板.
prompt_template = PromptTemplate(
    template="今天的日期是 {date},天气是 {wheather}",
    input_variables=["wheather"],
    partial_variables={"date": _get_datetime},
)
prompt_template.invoke({"wheather": "小雨"})
# 输出.
# StringPromptValue(text='今天的日期是 02/27/2025, 20:00:00,天气是 小雨')

6. 结构化输出 OutputParser

Output parsers | ️ LangChain 前面已经介绍了prompt 和 model,剩下的就是 model 的输出解析。通常来说,模型返回的输出是一个 AIMessage 对象,本质上还是字符串文本。在构造应用时,常常需要对大模型的输出做结构化,比如转换为 JSON 或者 Enum,以便于下游使用。这时就需要 OutputParser 来支持。

每种数据类型有自己对应的 OutputParser,比如 JSON 对应着 JsonOutputParser。无论哪种类型,都有一个公共方法 get_format_instructions(),可以返回对应格式的约束描述(schema)。实际上,结构化输出有两个步骤:

在 prompt 中注入 get_format_instructions(),告诉 llm 要生成什么类型的结果; 在生成之后,进行对应格式的转化; 6.1. 生成格式约束 以图书信息为例子,首先要基于 pydantic 构造 schema:

from pydantic import BaseModel, Field

class Book(BaseModel):
    name: str = Field(description="书籍名称")
    year: int = Field(description="出版年份")

然后基于 schema,构造 OutputParser:

from langchain_core.output_parsers import JsonOutputParser

parser = JsonOutputParser(pydantic_object=Book)
parser.get_format_instructions()
# 输出
# 'The output should be formatted as a JSON instance that conforms to the JSON schema below.\n\nAs an example, for the schema {"properties": {"foo": {"title": "Foo", "description": "a list of strings", "type": "array", "items": {"type": "string"}}}, "required": ["foo"]}\nthe object {"foo": ["bar", "baz"]} is a well-formatted instance of the schema. The object {"properties": {"foo": ["bar", "baz"]}} is not well-formatted.\n\nHere is the output schema:\n```\n{"properties": {"name": {"description": "书籍名称", "title": "Name", "type": "string"}, "year": {"description": "出版年份", "title": "Year", "type": "integer"}}, "required": ["name", "year"]}\n```'

调用 get_format_instructions() 后,我们得到了一个很长的字符串。仔细看,这是一段关于格式约束的 prompt,它要求模型生成 JSON,然后告诉模型什么是 schema,然后对目标 schema 做了字段提示。当然,如果你不使用 OutputParser,也完全可以自己给 llm 写这么一段 prompt。

有了这段格式约束,就可以通过 PromptTemplate 的 partial_variables 注入进总体 prompt 中去。比如:

prompt = PromptTemplate(
    template="请帮我从出版列表中解析图书的信息。\n{format_instructions}\n出版列表:{query}\n",
    input_variables=["query"],
    partial_variables={"format_instructions": parser.get_format_instructions()},
)

6.2. 生成结构解析 如果 llm 依从指令,生成了 JSON 格式的内容,就可以做下一步的结构化了:

response = "{\"name\": \"The World\", \"year\": 2025}"
parser.invoke(response)
# 输出.
# {'name': 'The World', 'year': 2025}

字符串的 response 就被转换成了 dict(也就是 JSON)。需要注意,这一步和前面定义的 schema 没有任何关系,只是单纯地做 JSON 转换。也就是说,只要是 JSON 字符串,无论是什么字段,都能转换成功。schema 只是用来生成格式约束来注入 prompt 中,并不参与生成后的格式解析。

6.3. 格式纠错 如果 llm 生成的结果不是标准 JSON ,比如缺了引号或者加了别的字,可以用带纠错的 OutputFixingParser 来修复。OutputFixingParser 在解析输入时,如果直接成功会直接返回结构化对象,如果不成功则调用 llm 做默认1次的修复。

from langchain.output_parsers import OutputFixingParser

bad_response = "json {'name': 'The World', 'year': 2025}"
fix_parser = OutputFixingParser.from_llm(parser=parser, llm=llm)
fix_parser.parse(bad_response)
# 输出.
# {'name': 'The World', 'year': 2025}

和普通的 OutputParser 一样,OutputFixingParser 也是 schema 无关的,它只是做「格式」的修复。

7. 链 chain

一次基本的大模型调用,prompt、model 和 parser 都有了,而且都是 Runnable 对象,那么就能串联成 chain:

chain = prompt | llm | parser
chain.invoke({"query": "xxxxxxxxxxxxxx"})

由于 prompt 是 chain 的入口,那么 chain.invoke() 的输入参数就和 prompt 一致;parser 是 chain 的出口,那么 chain.invoke() 的输出格式就和 parser 一致。

有了这三部分,就能实现大部分的基础 llm 调用。langchain 还支持很多其他的范式,后续再讲。

回到顶部