langchain（一）
大语⾔模型 （LLM） 在各种与语⾔相关的任务（例如⽂本⽣成、翻译、摘要、问答等）中表现出⾊。现代 LLM 通常通过聊天模型接⼝访问，该接⼝将消息列表作为输⼊（LangChain消息列表），并返回消息作为输出（AIMessage），而不是使 ⽤纯⽂本。这⾥需要注意LLM与 LangChain 中 聊天模型 的关系：•在 LangChain 的官方文档中，认为 LLM⼤多数是纯⽂本补全模型。这些纯⽂
环境说明
python 3.13
requirements.txt文件如下
langchain-openai==0.3.33
langchain==0.3.27
langchain-deepseek==0.1.4
langchain-ollama==0.3.6
langchain-tavily==0.2.12
langchain-chroma==0.2.5
langchain-community==0.3.22
nltk==3.9.2
langchain-redis==0.2.4
unstructured==0.18.15
markdown==3.9
redisvl==0.10.0安装包命令 pip install -r requirements.txt
快速上手
编码
1、申请API key并配置本地环境
以openai为例，https://platform.openai.com/ 需要科学上网，
然后再环境变量中配置这个key
这里配置为环境变量之后，后续编码就能自动读取环境变量中的key了。
将 API Key 配置在环境变量中主要是为了保证其隐私性。由于 api key ⽐较隐私，为避免在程序中暴 露，可提前将各个 api key 配置在环境变量中，这样在程序中就可以通过获取对应环境变量拿到 api key，保证了 api key 的隐私。
2、开始编码
首先安装OpenAI包 pip install -U langchain-openai
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableSequence
from langchain_openai import ChatOpenAI
from langchain_core.messages import HumanMessage, SystemMessage
# 通过LangChain接入LLM
# 1、定义OpenAI模型 组件
# 默认从系统环境变量读取 OPENAI_API_KEY
model = ChatOpenAI(model="gpt-4o-mini")
# 2、定义消息
# 用户发出的消息 在langchain中定义为了HumanMessage
# 系统提示消息 SystemMessage
# AI响应消息 AIMessage
message = [
# 系统提示消息，只用定义一遍就行了，通常作为第一条提示消息传入，用于启动AI的行为
SystemMessage(content= "请帮我进行翻译，由英文翻译成中文！"),
HumanMessage(content= "good morning，Damn！")
]
# 3、调用大模型
# result = model.invoke(message)
# 大模型输出
# print(result)
# 链式体现在哪里呢
# 4、定义输出解析器 组件
parser = StrOutputParser()
# print(parser.invoke(result))
# 5、定义链，链的顺序不能乱放，执行顺序是从左向右依次执行的，前一个组件的输出作为下一个组件的输入
# 方式一
chain = model | parser
# 方式二
# chain = RunnableSequence(first=model, last=parser)
# 方式三
# chain = model.pipe(parser)
# 6、执行链
print(chain.invoke(message))
langchain的相关概念
Runnable接口
在上面的代码中，可以看到定义的model还是parser都能执行invoke方法，这里都离不开一个概念叫Runnable。
Runnable 接⼝是使用 LangChain Components（组件）的基础。
Components（组件）：⽤来帮助当我们在构建应⽤程序时，提供了⼀系列的核⼼构建块，例如语 ⾔模型、输出解析器、检索器、编译的 LangGraph 图等。
Runnable 定义了⼀个标准接口，允许 Runnable 组件：
• Invoked（调用）: 单个输⼊转换为输出。
• Batched（批处理）: 多个输⼊被有效地转换为输出。
• Streamed（流式传输）: 输出在⽣成时进⾏流式传输。
• Inspected（检查）: 可以访问有关 Runnable 的输⼊、输出和配置的原理图信息。
• Composed（组合）: 可以组合多个 Runnable，以使⽤ LCEL 协同⼯作以创建复杂的管道。
因此，在上面的代码中，我们定义的语⾔模型（model）、输出解析器（StrOutputParser）都是 Runnable 接⼝的实例！他们都使⽤了 Invoked（调⽤）的能力。
另外，之前说过，原生接入LLM，存在模型切换的问题。但是目前使用Runnable定义模型组件的时候，只用改变下模型定义的接口就行了（上面是model=ChatOpenAI(model="gpt-4o-mini")， 换成model=ChatGoogleGenerativeAI()），定义出来的模型组件依旧是Runnable接口，后续使用调用模型的过程完全不用改，这样切换模型就很方便了。
LCEL
LCEL（LangChain Expression Language），采用声明性方法，从现有 Runnable 对象构建新的 Runnable 对象。
通过 LCEL 构建出的新的 Runnable 对象，被称为 RunnableSequence ，表⽰可运⾏序列。 RunnableSequence 就是⼀种 链。通过调试 就能发现，chain 的类型就是 RunnableSequence
重要的是， RunnableSequence 也是 Runnable 接⼝的实例 ，它实现了完整的 Runnable 接⼝， 因此它可以⽤与任何其他 Runnable 相同的姿势使⽤，也就是上面也能调用invoke方法的原因。
chain = model | parser
chain.invoke(messages) # 链是 Runnable 接⼝实例，允许invoke调⽤可以看到，LCEL 其实是⼀种编排解决方案（将上一个Runnable实例的执行结果，传输给下一个Runnable实例的输入），它使 LangChain 能够以优化的⽅式处理链的运⾏时执 行。任何两个 Runnable 实例都可以“链”在⼀起成序列。上⼀个可运⾏对象的 .invoke（） 调用 的输出作为输⼊传递给下⼀个可运⾏对象。⽅法就是使⽤ | （管道/运算符）：
它通过两个 Runnable 对象去创建⼀个 RunnableSequence 。实际上 LangChain 重载了 | 运算符，使⽤ | 运算符就相当于：
from langchain_core.runnables import RunnableSequence
chain = RunnableSequence(first=model, last=parser)除此之外，可以使⽤ .pipe ⽅法代替。这也相当于 | 运算符：
chain = model.pipe(parser)在 Unix/Linux 系统中， pipe() 系统调⽤和 | 管道操作符都⽤于实现进程间通信，这⾥同样也 是迁移过来的用法。
聊天模型
定义聊天模型
大语⾔模型 （LLM） 在各种与语⾔相关的任务（例如⽂本⽣成、翻译、摘要、问答等）中表现出⾊。 现代 LLM 通常通过聊天模型接⼝访问，该接⼝将消息列表作为输⼊（LangChain消息列表），并返回消息作为输出（AIMessage），而不是使 ⽤纯⽂本。
这⾥需要注意 LLM 与 LangChain 中 聊天模型 的关系：
• 在 LangChain 的官方文档中，认为 LLM ⼤多数是纯⽂本补全模型。这些纯⽂本模型封装的 API 接 受⼀个字符串提⽰作为输⼊，并输出⼀个字符串补全结果（实际上 LLM 还包括多模态输⼊）。 OpenAI 的 GPT-5 就是作为 LLM 来实现的。
• LangChain 中的 聊天模型 通常由 LLM 提供⽀持，但经过专⻔调整以⽤于对话。关键在于，它们 不是接受单个字符串作为输⼊，⽽是接受聊天消息列表，并返回⼀条 AI 消息作为输出。
方式1：ChatOpenAI
class langchain_openai.chat_models.base.ChatOpenAI 是 LangChain 为 OpenAI 的 聊天模型（如 gpt-5 , gpt-5-mini ）提供的具体实现类。
其继承了 class langchain_openai.chat_models.base.BaseChatOpenAI ，且 BaseChatOpenAI 实现了标准的 Runnable 接⼝。
ChatOpenAI常用初始化参数如下
|
参数名
|
参数描述
|
|
model
|
要使⽤的 OpenAI 模型的名称
|
|
temperature
|
采样温度，温度值越⾼，AI 回答越天马行空；温度越低，回答越保守靠谱。温度范围一般是0~1，或者0~2。
|
|
max_tokens
|
要⽣成的最大令牌数
|
|
timeout
|
请求超时时间
|
|
max_retries
|
最⼤重试次数
|
|
openai_api_key / api_key
|
OpenAI API 密钥。如果未传⼊，将从环境变量中读取
OPENAI_API_KEY 。
|
|
base_url
|
API 请求的基本 URL。
|
|
organization
|
OpenAI 组织 ID。如果未传⼊，将从 env var OPENAI_ORG_ID 中读取。
|
# 1、定义OpenAI模型 组件
model = ChatOpenAI(
model="gpt-4o-mini",
# temperature=2, # 采样温度，温度越高，AI回复的内容越天马行空，温度越低，答案越保守。温度范围一般是0~1或者0~2
# max_tokens=10, # 最大token数，token是衡量文本的基本单位
# timeout=None, # 请求超时时间
# max_retries=2, # 最大重试次数
# api_key="...", # API key填写位置，如果不填 默认是从环境变量中读取
# base_url="...", # 填写本次调用大模型的前置路径
# organization="...", # openai组织 ID
# other params...
)invoke调用
该⽅法是将单个输 ⼊转换为对应的输出。例如对于聊天模型来说，就是根据用户的问题输入，输出相应的答案。
invoke方法定义如下：
abstractmethod invoke(
input: Input,
config: RunnableConfig | None = None,
**kwargs: Any,
) → Output请求参数：
• input ：输⼊⼀个 Runnable 实例
• config （默认空）：⽤于 Runnable 的配置。
返回值：
• 返回⼀个 Runnable 实例
常用参数如下
| 参数名 | 参数描述 |
|
configurable
|
通过 .configurable_fields（）在此 Runnable 或⼦ Runnable 上配置的属
性的运⾏时值。使用更多的字段。
|
|
run_id
|
针对此调⽤运⾏的跟踪器的唯⼀标识符。如果未提供，将⽣成新的 UUID。
|
|
run_name
|
此调⽤的跟踪器运⾏的名称。默认为类的名称。
|
|
metadata
|
此次调⽤和任何⼦调⽤的元数据。键是字符串，值是 JSON。
类型：dict[str， Any]
|
方式2：init_chat_model
ChatOpenAI ⽤于明确创建 OpenAI 聊天模型的实例。init_chat_model() 是⼀个⼯ ⼚函数，它可以初始化多种⽀持的聊天模型（如 OpenAI、Anthropic、FireworksAI 等），不仅仅是 OpenAI 的聊天模型。
| 参数名 | 参数描述 |
|
model_provider
|
模型提供⽅。⽀持的 model_provider 值和相应的集成包有：
• openai -> langchain-openai
• anthropic -> langchain-anthropic
• google_genai -> langchain-google-genai
• ollama -> langchain-ollama
• deepseek -> langchain-deepseek
• ...
如果未指定，将尝试从模型推断 model_provider。
|
|
configurable_fields
|
设置哪些模型参数是可配置的。若配置为：
• None: 没有可配置的字段。
• 'any' ：所有字段都是可配置的，类似 api_key 、 base_url 等可
以在运⾏时更改。
• Union[List[str], Tuple[str, …]]：指定的字段是可配置的。
|
|
config_prefix
|
• 配置为⾮空字符串，则模型将在运⾏时通过查找
config["configurable"]["{config_prefix}_{param}"] 字段
设置配置项 。
• 配置为空字符串，那么模型将可以通过 config["configurable"]["
{param}"] 字段设置配置项 。
|
其余很多字段和上面的ChatOpenAI的参数相同，这里只说明了ChatOpenAI没有的参数。
# 1、基本用法
# LangChain 封装了更上层的方法，让我们初始化模型
from langchain_core.messages import SystemMessage, HumanMessage
gpt_model = init_chat_model("gpt-4o-mini", model_provider="openai", temperature=0.3)
# deepseek_model = init_chat_model("deepseek-chat", model_provider="deepseek")
print(f"gpt-4o-mini:{gpt_model.invoke('你是谁哦').content}")init_chat_model() 返回值说明
函数返回⼀个与指定的 model_name 和 model_provider 相对应的 BaseChatModel （如 ChatOpenAI , ChatAnthropic 等）。注意要是模型可配置，则返回⼀个聊天模型模拟器，该模 拟器在传⼊配置后，于运行时才会初始化底层模型。
# 2、定义可配置的模型（模型模拟器）
config_model = init_chat_model(temperature=0.3)
message = [
SystemMessage(content= "请补全一段故事，20个字以内："),
HumanMessage(content= "一只狗__?")
]
# invoke()方法完善可配置模型的参数，包括具体模型和传入消息等
result = config_model.invoke(input=message, config={
"configurable" : {"model" : "gpt-4o-mini"}
})
print(f"config_model:{result.content}")
# 3、定义可配置的模型（默认参数）
# 在定义模型的时候，给定了默认参数，在后面调用的过程中，对默认参数进行修改
# 希望得到两版输出，原版输出，精简版输出
model = init_chat_model("gpt-4o-mini",
model_provider="openai",
temperature=0.3,
max_tokens=1024,
configurable_fields=("max_tokens", "model", "model_provider", ),
config_prefix="first" # 这里的前缀是 在后面对默认参数进行修改的时候， 需要在默认参数前面加上
)
message = [
SystemMessage(content= "请补全一段故事，100个字以内："),
HumanMessage(content= "一只狗__?")
]
result = model.invoke(message
, config={
"configurable" : {
"first_max_tokens" : 10,
# ""
}
})
print(result.content)调用工具
⼯具调⽤根本作⽤是让⼤语⾔模型（LLM）具备与外部世界交互的能力。
LLM 本⾝是⼀个封闭的知识系统，其能力受限于其训练数据（存在滞后性）和内在的文本生成逻辑。 它⽆法执⾏直接计算、查询实时信息、操作数据库或调⽤任何外部 API。⼯具调⽤打破了这层壁垒，其 作用具体体现在：
1. 扩展能力边界：模型可以借助⼯具完成它自身无法完成的任务，如执行数学计算、搜索网络、查询 数据库等。
2. 保证信息实时性：通过调⽤搜索⼯具或数据库查询⼯具，LLM 可以获取最新的、训练数据中不存在 的信息，避免回答过时或“⼀本正经地胡说⼋道”。
3. 处理复杂任务：将⼀个复杂的用户请求（如“分析我上个⽉的消费趋势”）分解成多个步骤，并依 次调⽤不同的⼯具（如“从数据库获取数据” -> “⽤ Python 进⾏数据分析” -> “生成图表”） 来协同完成。协调这件事这更体现在 Agent 智能体上。
4. 连接现有系统：可以将企业内部已有的系统、API 和数据库封装成⼯具，让 LLM 成为⼀个⽤⾃然语 ⾔驱动的统⼀接⼝，极⼤地提升了⾃动化和集成能⼒。
在 LangChain 中，聊天模型提供了额外的功能：工具调用。它能使 LLM 与外部服务、API 和数据库进 ⾏交互。⼯具调⽤还可⽤于从⾮结构化数据中提取结构化信息并执⾏各种其他任务。
创建工具
使用@tool装饰器创建工具
在 LangChain 中，实现了⼀个 @tool 装饰器来创建⼯具， @tool 装饰器是⾃定义⼯具的最简单方 法。
# 定义一个工具
# 方式一，依赖函数内部写好 文档字符串
@tool
def add(a: int, b: int) -> int:
"""两数相加
Args：
a：第一个整数
b：第二个整数
"""
return a + b
print(add.invoke({"a": 2, "b": 3}))
print(add.name)
print(add.description)
print(add.args)可以看出，⼯具通过 @tool 加 Python函数 实现，其中：
• 该装饰器默认使用函数名称作为工具名称。
• 该装饰器将使用函数的文档字符串作为工具的描述。
因此，函数名、类型提⽰和⽂档字符串都是传递给⼯具 Schema 的⼀部分，不可缺失。定义好的描述 是使模型良好运⾏的重要部分。
工具Schema是就描述工具结构的声明格式（具有哪些字段，字段的值类型是怎么样的），简明扼要地描述数据的表面结构，并根据数据自动验证数据很容易。
工具Schema的属性包括工具名称、工具描述和工具参数。在工具函数中添加描述文档字符串，就是为了传递给工具Schema的。对于⼯具schema，它将从函数名、类型提示和文档字符串中获取相关属性，以此 来声明⼀个⼯具，包括其名称、描述、输⼊参数、输出类型等等。这⾥需要说明的是，若是简单定义 ⼯具，如上述⽰例，工具 schema 需要解析 Google 风格的文档字符串去获取【参数描述】。
工具名称、工具描述和工具参数是一个不能少的，在文档字符串中。
工具最终是要被LLM调用的，提示词可以决定LLM调用输出结果的优劣。对于工具来说：1、工具的名称可以让LLM知道有哪些工具，可以调用哪些工具。2、工具描述实际上就是在写提示词，告诉模型工具的能力。3、工具参数可以让模型知道怎么调。
模式1：依赖 Pydantic 类
若使用 @tool 定义⼯具时，没有提供文档字符串，则会报错。此时，在 LangChain 中，可以使用 Pydantic 类，提供运行时数据验证和类型检查。通过 Field(description="...") 添加字段描述，LangChain 会⾃动提取。
# 方式二，依赖 Pydantic 类
class AddInput(BaseModel):
"""两数相加"""
a: int = Field(..., description="第一个整数") # ... 表示这个参数必填
b: int = Field(..., description="第二个整数")
@tool(args_schema=AddInput)
def add(a: int, b: int) -> int:
return a + b注意是代码中 @tool 的 args_schema 参数，它表示工具函数在未提供描述、文档字符串等需要传 递给⼯具 Schema 的内容时，依赖 Pydantic 类使⽤ args_schema 参数，定义并提供⼯具输⼊参数 的schema。默认为 None。
点击运行，不会报错，且将来运⾏时会进行数据验证。因此，我们再次印证了函数名、类型提示和文 档字符串都是传递给⼯具 Schema 的⼀部分，不可缺失。
模式2：依赖 Annotated
在 LangChain 中，可以依赖 Annotated 和⽂档字符串传递给⼯具 Schema
# 方式三，依赖依赖 Annotated
@tool
def add(
a: Annotated[int, ..., "第一个参数"],
b: Annotated[int, ..., "第二个参数"],
) -> int:
"""两数相加
Args：
a：第一个整数
b：第二个整数
"""
return a + b
print(add.invoke({"a": 2, "b": 3}))
print(add.name)
print(add.description)
print(add.args)使用StructuredTool类提供的函数创建工具
class langchain_core.tools.structured.StructuredTool 类⽤来初始化⼯具，其中 from_function 类⽅法通过给定的函数来创建并返回⼀个⼯具。from_function 类⽅法定义如 下：
classmethod from_function(
func: Callable | None = None,
coroutine: Callable[[...], Awaitable[Any]] | None = None,
name: str | None = None,
description: str | None = None,
return_direct: bool = False,
args_schema: type[BaseModel] | dict[str, Any] | None = None,
infer_schema: bool = True,
*,
response_format: Literal['content', 'content_and_artifact'] = 'content',
parse_docstring: bool = False,
error_on_invalid_docstring: bool = False,
**kwargs: Any,
) → StructuredTool关键参数说明：
• func：要设置的⼯具函数
• coroutine：协程函数，要设置的异步⼯具函数
• name：⼯具名称。默认为函数名称。
• description：⼯具描述。默认为函数⽂档字符串。
• args_schema：⼯具输⼊参数的schema。默认为 None。
• response_format：⼯具响应格式。默认为“content”。
◦ 如果配置为 “content” ，则⼯具的输出为 ToolMessage 的 content 属性。
▪ 对于 HumanMessage 、 AIMessage 已经见过，分别表⽰ 用户消息 和 AI消息响应 ， 对于 ToolMessage ，它表⽰对应⼯具⻆⾊所发出的消息。
◦ 如果配置为 “content_and_artifact” ，则输出应是与 ToolMessage 的 content 属性 与 artifact 属性相对应的⼆元组。
常规使用
对于⽤该类⽅法创建的⼯具，同样函数名、类型提⽰和⽂档字符串也都是传递给⼯具 Schema 的⼀部 分，不可缺失。
# 方式一
# 这里的函数中的文档字符串 必须写好
def add(a: int, b: int) -> int:
"""两数相加"""
return a + b
add_tool = StructuredTool.from_function(func=add)
print(add_tool.invoke({"a": 2, "b": 3}))加入配置，依赖 Pydantic 类
同样的，让⼯具函数不提供描述、文档字符串等需要传递给⼯具 Schema 的内容
# 方式二，通过指定参数替代工具函数中的文档字符串描述
class AddInput(BaseModel):
a: int = Field(description="第一个整数"),
b: int = Field(description="第二个参数")
def add(a: int, b: int) -> int:
return a + b
add_tool = StructuredTool.from_function(
func=add,
name= "ADD", # 工具名
description= "两数相加", # 工具描述
args_schema=AddInput, # 工具参数
)使用 args_schema 参数，依赖 Pydantic 类定义并提供⼯具输⼊参数的 schema 属性。
使用 description 参数，替代⽂档字符串中对于⼯具描述的 schema 属性。
加入 response_format 配置
如果希望我们的⼯具区分消息内容（content）和其他⼯件（artifact），让⼤模型读取 content ， 而⼀些⽤来构造 content 的原始数据保存下来，若后续有⼀些记录、分析的步骤，就可以派上⽤场 了，这就是 artifact 。 artifact 通常需要使⽤字典 Dict 或列表 List 保存。
# 方式三
class AddInput(BaseModel):
a: int = Field(description="第一个整数"),
b: int = Field(description="第二个参数")
# 这里的str是保留返回的结果，List里面就是保留返回的过程
def add(a: int, b: int) -> Tuple[str, List[int]]:
nums = [a, b]
content = f"{nums} 相加的结果是{a + b}"
return content, nums
add_tool = StructuredTool.from_function(
func=add,
name= "ADD", # 工具名
description= "两数相加", # 工具描述
args_schema=AddInput, # 工具参数
response_format="content_and_artifact", # 将内容和结果分开
)
print(add_tool.invoke({"a": 2, "b": 3})) # 手动调用工具的正常写法，显示的结果并不完整，只会显示content，过程并不会显示
# 模拟大模型调用工具姿势，聊天模型就可以通过content获取工具结果，通过artifact获取过程
print(add_tool.invoke( # 模拟调用大模型调用姿势
{
"name": "ADD",
"args": {"a": 3, "b": 4},
"type": "tool_call", # 必填，工具调用时，必填
"id": "111", # 必填，将工具调用请求和调用结果关联起来
}))我们需要在定义工具时指定 response_format="content_and_artifact" 参数，并确保我 们返回⼀个元组 (content, artifact)。若想要看到工具返回的元组，我们需要模拟大模型调用工具的姿势。
由于 LLM 大多理解⽂本，所以工具的主要输出 content 必须是结构良好、简洁的⽂本， 以便模型能够轻松理解和基于它进⾏推理、生成下⼀步的指令。
在链（Chain）中，工具调⽤之后的其他组件或函数，可能需要工具的原始且结构化数据（即 artifact ）来执行特定操作。这些数据可能是庞大的、且非文本的。这些数据不适合直接塞给模 型。因此， artifact 其实是为了给链中后续的组件或函数使用的，不被⼤模型所直接使⽤！！
绑定工具
为了实际将这些工具绑定到聊天模型，可以使⽤聊天模型的 .bind_tools() 方法。
@tool
def add(
a: Annotated[int, ..., "第一个参数"],
b: Annotated[int, ..., "第二个参数"],
) -> int:
"""两数相加"""
return a + b
@tool
def multiply(
a: Annotated[int, ..., "第一个参数"],
b: Annotated[int, ..., "第二个参数"],
) -> int:
"""两数相乘"""
return a * b
model = ChatOpenAI(model="gpt-4o-mini")
#绑定工具
tools = [add, multiply]
# 模型会根据 语义 相关性去 选择调用哪一个工具
model_with_tools = model.bind_tools(tools=tools)
# 强制选择工具，也就是说聊天 模型，就不会考虑语义相关性的问题了，直接选择随机选择一个工具
# model_with_tools = model.bind_tools(tools=tools, tool_choice="any")bind_tools方法如下
bind_tools(
tools: Sequence[dict[str, Any] | type | Callable | BaseTool],
*,
tool_choice: dict | str | Literal['auto', 'none', 'required', 'any'] | bool
| None = None,
strict: bool | None = None,
parallel_tool_calls: bool | None = None,
**kwargs: Any,
) → Runnable[PromptValue | str | Sequence[BaseMessage | list[str] |
tuple[str, str] | str | dict[str, Any]], BaseMessage]请求参数
• tools ：绑定到此聊天模型的⼯具定义列表。⽀持的类型为：字典、pydantic.BaseModel 类、 Python 函数和 BaseTool（如 @tool 装饰器创建的类）。
• tool_choice （默认空）：要求模型调⽤哪个⼯具。可以设置为：
◦ 形式为 '<<tool_name>>' 的 str：调⽤ <tool_name> ⼯具。
◦ 'auto' ：⾃动选择⼯具（包括无工具）。
◦ 'none' ：不调用⼯具。
◦ 'any' 或 'required' 或 True ：强制调用至少⼀个⼯具。
◦ False 或 None ：⽆效果，默认 OpenAI 的行为。
• strict （默认空）：
◦ 如果为 True ，则保证模型输出与⼯具定义中提供的 JSON Schema 完全匹配。输⼊也将根 据提供的 Schema 进⾏验证。
◦ 如果为 False ，则不会验证输⼊，也不会验证模型输出。
◦ 如果为 None ，则不会将 strict 参数传递给模型。
• parallel_tool_calls ：默认为 None ，允许并⾏⼯具使⽤。设置为 False 以禁⽤并⾏ ⼯具。
• kwargs(Any) ：任何附加参数都直接传递给 bind() 。
返回值
• 返回⼀个 Runnable 实例。
◦ 该实例⽀持多种格式输⼊：
▪ 原始提⽰ PromptValue
▪ 字符串： "上海天⽓如何？"
▪ 消息或消息列表：[HumanMessage(content="...")]
◦ 该实例的输出：
▪ 包含⼯具调⽤信息的 AIMessage
工具调用
通过 .bind_tools() ⽅法我们可知，它返回了⼀个 Runnable 实例，因此我们可以使⽤该 Runnable 实例，调⽤ .invoke() ⽅法，完成⼯具调用。
@tool
def add(
a: Annotated[int, ..., "第一个参数"],
b: Annotated[int, ..., "第二个参数"],
) -> int:
"""两数相加"""
return a + b
@tool
def multiply(
a: Annotated[int, ..., "第一个参数"],
b: Annotated[int, ..., "第二个参数"],
) -> int:
"""两数相乘"""
return a * b
model = ChatOpenAI(model="gpt-4o-mini")
#绑定工具
tools = [add, multiply]
# 模型会根据 语义 相关性去 选择调用哪一个工具
model_with_tools = model.bind_tools(tools=tools)
# 强制选择工具，也就是说聊天 模型，就不会考虑语义相关性的问题了，直接选择随机选择一个工具
# model_with_tools = model.bind_tools(tools=tools, tool_choice="any")
print(model_with_tools.invoke("2加3等于多少"))调用结果部分结果 分析
调用LLM，会将结果输出到content里，但是这里的content为空。additional_kwargs是传递工具调用的相关信息的字段，里面的tool_calls字段就是描述工具选择，可以看到模型会根据语义相关性和工具的schema匹配对应的函数。
这里的tool_calls是additional_kwargs字段同层级的，是专门执行工具的字段，可以发现tool_calls字段中的属性和之前模拟大模型调用工具姿势几乎一样。
print(add_tool.invoke( # 模拟调用大模型调用姿势
{
"name": "ADD",
"args": {"a": 3, "b": 4},
"type": "tool_call", # 必填
"id": "111", # 必填，将工具调用请求和调用结果关联起来
}))因此这里直接使用tool_calls就能调用工具。
ai_msg = model_with_tools.invoke("2加3等于多少")
print(add.invoke(ai_msg.tool_calls[0]))运行结果如下
思考：但是我们更希望 返回“2 加 3 等于5” 或者 “2 + 3 = 5”这样的字符串，而不是冰冷的结果。
而需要产生这样的结果需要将问题、AIMessage（包含选择的工具，及工具的调用姿势）和工具调用的结果输入给聊天模型分析然后输出。而聊天模型接受的输入分别为字符串和消息列表。问题-->HumanMessage，AIMessage就是一种消息，工具调用结果-->ToolMessage。进而就将问题、AIMessage和工具调用结果转换为了一个包含HumanMessage、AIMessage和ToolMessage的消息列表作为聊天模型的输入了。
"""
将问题转换为HumanMessage，
输入绑定工具的模型，转换为AIMessage，AIMessage包含了选择的工具，以及工具的调用姿势
再调用工具，调用工具分为两步，先选择工具，再调用工具。
再将调用工具转换为ToolMessage
分别将HumanMessage、AIMessage、ToolMessage构造出消息列表再传入聊天模型中，进行输出
"""
# 定义消息列表，添加要传递给聊天模型的消息
message = [
HumanMessage("2加3等于多少？ 7乘4等于多少"),
]
ai_msg = model_with_tools.invoke(message)
message.append(ai_msg)
# 构造ToolMessage，并添加到消息列表中
for tool_call in ai_msg.tool_calls:
selected_tool = {"add": add, "multiply": multiply}[tool_call["name"].lower()]
tool_msg = selected_tool.invoke(tool_call)
message.append(tool_msg)
print(model_with_tools.invoke(message).content)从流程与代码中可以看到，实际上我们调⽤了两次聊天模型：
• 第⼀次：仅将【 HumanMessage 】发送给聊天模型进⾏处理，结果返回了【包含⼯具调⽤的 AIMessage 】，并没有返回我们想要的结果。然后我们执⾏⼯具，得到 ToolMessage 。
• 第⼆次：将【 HumanMessage + AIMessage + ToolMessage 】消息记录发送给聊天模型进⾏ 处理，结果返回了【包含结果的 AIMessage 】
LangChain提供的工具
⼯具也不是全部都需要我们⾃⼰⼿搓，其实 LangChain 官⽅也已经给我们提供了很多现成的⼯具 （Tool）和⼯具包（Toolkit）。 LangChain 提供的⼯具见官方提供的工具。
写好的⼯具⼀般都是为了使⽤ LangChain 中集成的三⽅组件或⼯具⽽ 创造的，有搜索、数据库、⽹⻚浏览器等相关的⼯具。
TavilySearch
TavilySearch 类可以⽀持我们进⾏搜索，Tavily 是⼀个专⻔为 AI 设计的搜索引擎，专为智能体检索与 推理需求量⾝打造的⼯具。
Tavily 不仅提供了⾼度可编程的 API 接⼝，还具备显著优于传统搜索引擎的上下⽂相关性理解能⼒。 能够以结构化、可解析的形式返回搜索结果，便于将检索到的信息直接⽤于后续的推理、⽣成或任务 执⾏流程。Tavily官网
登录之后，需要新建API Keys。
在Langchain中接入该搜索工具，需要安装langchain-tavily 包：pip install -U langchain-tavily
另外需要将API Keys配在环境变量中，
然后就能在代码中使用了。
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch
# 定义模型
model = ChatOpenAI(model="gpt-4o-mini")
#定义搜索引擎 工具
tool = TavilySearch(max_result=4) # max_result 表示搜索结果的最大数
# 绑定工具
model_with_tools = model.bind_tools([tool])
# 定义消息列表
messages = [
HumanMessage("重庆今天的天气怎么样？")
]
ai_message = model_with_tools.invoke(messages)
messages.append(ai_message)
for tool_call in ai_message.tool_calls:
tool_message = tool.invoke(tool_call)
messages.append(tool_message)
print(model_with_tools.invoke(messages).content)运行结果如下
结构化输出
在 LangChain 中，聊天模型提供了额外的功能：结构化输出。⼀种使聊天模型以结构化格式（例如 JSON）进⾏响应的技术。例如，可能希望将模型输出存储在数据库中，并确保输出符合数据库模式。
这种需求激发了结构化输出的概念，其中可以指⽰模型使⽤特定的输出结构进⾏响应。
这样做的核⼼原因是：从“字符串”到“对象”的范式转换。想象⼀下，在没有这个功能之前，我们 调⽤聊天模型得到的是⼀个 AIMessage ，其内容是⼀个字符串。这个字符串对⼈类很友好，但对程序不友好。
聊天模型的 with_structured_output 方法则允许我们预先定义⼀个期望的数据结构，并要求大 模型必须按照这个结构返回信息。
with_structured_output()
要想使⽤结构化输出能⼒，LangChain 提供了⼀种⽅法 .with_structured_output() ，该⽅法 需要先定义输出结构，然后执⾏通过 .with_structured_output() 得到的 Runnable 实例。
with_structured_output() ⽅法定义：
with_structured_output(
schema: dict[str, Any] | type[_BM] | type | None = None,
*,
method: Literal['function_calling', 'json_mode', 'json_schema'] =
'json_schema',
include_raw: bool = False,
strict: bool | None = None,
**kwargs: Any,
) → Runnable[PromptValue | str | Sequence[BaseMessage | list[str] |
tuple[str, str] | str | dict[str, Any]], dict | _BM]请求参数
• schema ：表⽰输出结构。可以传⼊为：JSON、TypedDict、Pydantic、OpenAI 函数/⼯具
• method ：表⽰LLM的⽣成⽅法：
◦ json_schema （默认）：表⽰使⽤ OpenAI 的结构化输出 API。
◦ function_calling ：使⽤ OpenAI 的⼯具调⽤（以前称为函数调⽤）
◦ json_mode ：使⽤ OpenAI 的 JSON mode 。请注意，如果使⽤ json_mode ，则必须在 模型调⽤中包含将输出格式化为所需 schema 的说明。
• include_raw ：
◦ 如果为 False （默认），则仅返回解析的结构化输出。如果在模型输出解析过程中发⽣错 误，则会引发错误。
◦ 如果为 True ，则将返回原始模型响应（BaseMessage）和解析的模型响应。如果在输出解析 过程中发⽣错误，它也会被捕获并返回。最终输出始终是带有键 “raw” 、 “parsed” 和 “parsing_error” 的字典。
• strict ：
◦ 如果为 True ，保证模型输出与 schema 完全匹配。输⼊ schema 也将根据⽀持的 schema 进⾏验证。
◦ 如果为 False ，输⼊ schema 将不会被验证，模型输出也不会被验证。
◦ 如果为 None （默认），则不会将 strict 参数传递给模型。
• tools ：要绑定到聊天模型的⼯具列表。要求： method 为 'json_schema' 、 strict=True 、 include_raw=True 。则⽣成的 AIMessage 将在 'raw' 中包含⼯具调
• ⽤kw。args(Any) ：任何附加参数都直接传递给 bind() 。
返回值
返回⼀个 Runnable 实例。将来执⾏时：
• 如果 include_raw=False
◦ 且 schema是Pydantic类 ，则 Runnable 会输出 Pydantic 对象。
◦ 否则 Runnable 输出⼀个字典。
• 如果 include_raw=True ，则 Runnable 输出⼀个带有键的字典：
◦ 'raw' ：BaseMessage
◦ 'parsed' ：如果出现解析错误，则为 None，否则类型取决于如上所述的 schema 。
◦ 'parsing_error' ：Optional[BaseException]
返回 Pydantic 对象
model = ChatOpenAI(model="gpt-4o-mini",)
# Pydantic 对象
class Joke(BaseModel):
"""给用户讲的一个笑话"""
setup: str = Field(description="这个笑话的开头"),
punchline: str = Field(description="这个笑话的妙语"),
rating: Optional[int] = Field(default=None, description="从1-10分，给这个笑话评分") # default=None 表示默认为空
class Data(BaseModel):
"""获取关于笑话的数据列表"""
jokes: List[Joke]
structured_model = model.with_structured_output(Data) # 绑定结构化输出的model，返回的也是一个Runnable对象
result = structured_model.invoke("分别讲一个唱歌和跳舞的笑话")
print(result)输出如下
返回 TypedDict
先了解⼀下 TypedDict ，它⽤于为字典对象提供精确的、结构化的类型提⽰。它允许我们指定⼀个 字典中应该有哪些键，以及每个键对应的值的类型。
最清晰、最常⽤的定义⽅式，就是类似于定义⼀个类（Python 3.8+），如下所示：
from typing import TypedDict
class User(TypedDict):
name: str
age: int
email: str
is_active: bool = True # 默认值对于它非常重要的⼀个能⼒就是捕捉键名拼写错误与类型错误。如下
user1: User = {
"name": "Bob",
"age": 25,
"email": "bob@example.com"
}
# 类型检查器会捕获这些错误
bad_user: User = {
"name": "Dave",
"age": "forty", # 错误：应该是int
"emial": "dave@example.com" # 错误：拼写错误
}因此，我们也可以设置执行 Runnable 后的输出结果指定为 TypedDict 类，这将返回⼀个字典，且输 出后，会根据设定进⾏验证。
# TypedDict
class Joke(TypedDict):
"""给用户讲的一个笑话"""
setup: Annotated[str, ..., "这个笑话的开头"] # ...表示这个字段不能省
punchline: Annotated[str, ..., "这个笑话的妙语"]
rating: Annotated[Optional[int], None, "从1-10分，给这个笑话评分"]
model_with_structured = model.with_structured_output(Joke)
# include_raw=True 包含LLM完整的输出
model_with_structured = model.with_structured_output(Joke, include_raw=True)
print(model_with_structured.invoke("分别讲一个唱歌和跳舞的笑话"))输出如下
返回JSON
还可以让聊天模型直接返回 JSON，只不过为了声明 JSON，我们需要定义 JSON Schema，如下所 示：
json_schema = {
"title": "joke",
"description": "给⽤⼾讲⼀个笑话。",
"type": "object",
"properties": {
"setup": {
"type": "string",
"description": "这个笑话的开头",
},
"punchline": {
"type": "string",
"description": "这个笑话的妙语",
},
"rating": {
"type": "integer",
"description": "从1到10分，给这个笑话评分",
"default": None,
},
},
"required": ["setup", "punchline"],
}
structured_model = model.with_structured_output(json_schema) # 绑定结构化输出的model，返回的也是一个Runnable对象
result = structured_model.invoke("分别讲一个唱歌和跳舞的笑话")
print(result)输出如下
选择输出格式
model = ChatOpenAI(model="gpt-4o-mini")
"""
使用Pydantic嵌套的方式，让大语言模型选择（根据语义相关性）合适的输出结构
"""
# Pydantic 对象
class Joke(BaseModel):
"""给用户讲的一个笑话"""
setup: str = Field(description="这个笑话的开头"),
punchline: str = Field(description="这个笑话的妙语"),
rating: Optional[int] = Field(default=None, description="从1-10分，给这个笑话评分")
class Response(BaseModel):
"""用以对话的方式回应。"""
content: str = Field(description="用于对用户查询的会话响应")
class FinalResponse(BaseModel):
"""最终回复，选择合适的输出结构"""
final_output: Union[Joke, Response]
# 这里可以选择绑定Joke、Response和FinalResponse 三种
structured_model = model.with_structured_output(FinalResponse)
print(structured_model.invoke("讲一个关于跳舞的笑话"))
print(structured_model.invoke("你是谁"))输出如下
这里结构化输出的能力是聊天模型内部提供的，后面还有Langchain提供的具体组件来处理结构化输出。比如输出解析器。
结构化输出的使用场景
与工具相结合
from langchain_core.messages import HumanMessage
from langchain_openai import ChatOpenAI
from langchain_tavily import TavilySearch
# 定义模型
from pydantic import BaseModel, Field
model = ChatOpenAI(model="gpt-4o-mini")
#定义搜索引擎 工具
tool = TavilySearch(max_result=4) # max_result 表示搜索结果的最大数
# 绑定工具
model_with_tools = model.bind_tools([tool])
# 定义消息列表
messages = [
HumanMessage("重庆今天的天气怎么样？")
]
ai_message = model_with_tools.invoke(messages)
messages.append(ai_message)
for tool_call in ai_message.tool_calls:
tool_message = tool.invoke(tool_call)
messages.append(tool_message)
class SearchResult(BaseModel):
"""结构化搜索对象"""
query: str = Field(description="搜索查询")
findings: str = Field(description="查询结果摘要")
model_with_structure = model_with_tools.with_structured_output(SearchResult)
print(model_with_structure.invoke(messages))
流式输出
流式处理对于使基于 LLM 的应⽤程序能够响应最终用户至关重要。其通过逐步显示输出，甚⾄在完整 的响应准备就绪之前，流式传输可以显着改善用户体验。
我们之前直接使⽤ invoke 的调用方式属于非流式传输，看到的现象是聊天模型直接返回全量内容，若 模型思考时间较长，则我们等待的时间就越长。
stream() 同步传输
在 LangChain 聊天模型中，可以使⽤其 .stream() ⽅法，来同步⽣成流式响应的效果。 聊天模型的 .stream() 方法返回⼀个迭代器，该迭代器在生成输出时同步产⽣输出 消息块 。可以 使用 for 循环实时处理每个块。
model = ChatOpenAI(model="gpt-4o-mini")
# print(model.invoke("写一段关于春天的作文，1000字").content) # 非流式传输
# stream返回的是一个迭代器，产生的是消息块
chunks = []
for chunk in model.stream("写一段关于春天的作文，1000字"): # 实现流式传输
chunks.append(chunk)
# chunk类型 就是一个AIMessageChunk
print(chunk.content, end="|", flush=True) # 每个块的结果 使用 | 分割，再刷新
tmp_chunks = chunks[0] + chunks[1] + chunks[2] + chunks[3]
print(tmp_chunks)我们得到了⼀个叫做 AIMessageChunk 的东西，它代表 AIMessage 的⼀部分，也就是消息块。 消息块还可以直接相加。
astream() 异步传输
对于流式传输，通常我们可以选择异步调⽤。
异步相关概念
想象⼀个场景：你需要煮⼀壶⽔，同时还要给朋友发⼀条短信。我们分别⽤同步（传统）和异步两种 ⽅式来完成，以此对⽐并引⼊ 协程 和 事件循环 的概念。
同步（阻塞）方式：这就像是⼀个“死心眼”的⼈，做事必须⼀件⼀件来：
# 同步IO
def boil_water():
print("开始烧水了")
time.sleep(5)
print("烧水完成")
def send_message():
print("开始发消息")
time.sleep(2)
print("发消息完成")
def main():
# 1、烧水
boil_water()
# 2、发消息
send_message()
# 共耗时 7s
main()
执行结果
总耗时：7秒 问题： 在 boil_water 函数等待的5秒⾥，CPU 完全空闲，但却不能去做 send_message 任务， 效率低下。
异步方式：我们请出 asyncio 、 协程 和 事件循环 。
协程
• 多进程通常利⽤的是多核 CPU 的优势，同时执行多个计算任务。每个进程有自己独立的内存管 理，所以不同进程之间要进行数据通信比较麻烦。
• 多线程是在⼀个 cpu 上创建多个子任务，当某⼀个子任务休息的时候其他任务接着执⾏。多线程 的控制是由 python 自己控制的。线程存在数据同步问题，所以要有锁机制。
• 协程的实现是在⼀个线程内实现的，相当于流水线作业。由于线程切换的消耗比较大，所以对于 并发编程，可以优先使用协程。
协程，作为⼀种轻量级的并发编程模型，可以被视为用户态的“轻量级线程”。 与传统线程相比，协 程的核心优势在于其调度完全由用户空间掌控，避免了操作系统内核的频繁介⼊，从而显著降低了上 下文切换的开销。 在诸如⽹络数据刷新、资源加载、用户界⾯更新、以及 I/O 读写等场景下，如果并 发任务的计算量相对较小、对系统资源占⽤较低，则不必动⽤操作系统级别的线程。
协程的切换则由程序员和编程语⾔控制，程序员决定在何时暂停或恢复协程。协程是⼀个特殊的函 数，它可以在执⾏过程中暂停，并在稍后恢复执行。它⽤ async def 定义，并在需要暂停的地⽅使 用 await 。
事件循环
事件循环是 asyncio（Python 标准库中的模块，⽤于编写异步 I/O 操作的代码）的核⼼，你可以把它 想象成⼀个总调度员或⼀个⾼效的待办事项 (To-Do List) 管理员。
它的⼯作流程⾮常简单：
1. 它维护着⼀个任务列表（⽐如：煮⽔、发短信）。
2. 它不断地循环检查每个任务：
a. 如果任务处于 “等待I/O” 状态（⽐如等⽔开、等⽹络响应），就暂停它，⽴即去执⾏下⼀个 已经 “就绪” 的任务。
b. 如果任务的等待时间到了或者 I/O 操作完成了，事件循环就恢复执⾏这个任务。
# 异步IO
import asyncio
# 协程函数
async def boil_water_async():
print("开始烧水了")
# 关键！ await表示等待这个操作完成，但期间可以做别的事，就是会交出cpu的使用权
await asyncio.sleep(5)
print("烧水完成")
# 协程函数
async def send_message_async():
print("开始发消息")
await asyncio.sleep(2)
print("发消息完成")
# 协程函数
# 事件循环 的执行逻辑
async def main():
# 1、烧水（任务）
task1 = asyncio.create_task(boil_water_async())
# 2、发消息（任务）
task2 = asyncio.create_task(send_message_async())
await task1
await task2
# 共耗时 5s
# run方法 会创建一个事件循环，并运行一个指定的协程 main
asyncio.run(main())• 协程是 asyncio 的核心概念之⼀。它是⼀个特殊的函数，可以在执行过程中暂停，并在稍后恢 复执行。协程通过 async def 关键字定义，并通过 await 关键字暂停执⾏，等待异步操作完 成。
• 要运行⼀个协程，可以使用 asyncio.run() 函数。它会创建⼀个事件循环，并运⾏指定的协 程。事件循环是 asyncio 的核⼼组件，负责调度和执⾏协程。它不断地检查是否有任务需要执 行，并在任务完成后调⽤相应的回调函数。
具体使用
可以使⽤ .astream() ⽅法，来异步⽣成流式响应的效果，这专为⾮阻塞⼯作流程⽽设计。可以在 异步代码中使⽤它来实现相同的实时流式处理行为。
# 异步流式输出
async def async_stream():
print("开始异步调用")
async for chunk in model.astream("写一段关于春天的作文，50字"):
print(chunk.content, end="|", flush=True)
asyncio.run(async_stream())StrOutputParser解析模型输出
# 定义⼤模型
model = ChatOpenAI(model="gpt-4o-mini")
# 定义输出解析器
parser = StrOutputParser()
# 定义链
chain = model | parser
for chunk in chain.stream("写⼀段关于爱情的歌词，需要5句话"):
# 使用StrOutputParser，结果就是str
print(chunk, end="|", flush=True)自定义流式输出解析器
上面我们演示了如何让聊天模型进⾏流式输出。若此时我们希望修改上⼀步的输出样式（⼀个字或两 个字的输出），将输出改为⼀句话⼀句话的输出，同时保留流式处理功能。那么我们需要在链中使用 生成器函数，即可完成⾃定义流式输出的能⼒。
还记得之前说过，聊天模型的 .stream() ⽅法返回的是⼀个迭代器，该迭代器在⽣成输出时同步产 生输出 消息块 。那么我们的将实现的这些⽣成器的签名应该是 Iterator[Input] -> Iterator[Output] 。或者对于异步生成器： AsyncIterator[Input] -> AsyncIterator[Output] 。
# 模型组件
model = ChatOpenAI(model="gpt-4o-mini")
# 输出解析器组件
parser = StrOutputParser()
# 创建生成器
def split_into_list(input: Iterator[str]) -> Iterator[list[str]]:
buffer = ""
for chunk in input:
buffer += chunk
# 遇到句号,需要刷新buffer
while "。" in buffer:
# 找到句号位置
stop_index = buffer.index("。")
# yield 是python中的一个关键字,用于创建生成器函数
yield [buffer[:stop_index].strip()]
buffer = buffer[stop_index + 1 :]
# 处理buffer最后几个字
yield [buffer.strip()]
# 定义链 和 执行链
chain = model | parser | split_into_list
# stream返回的是一个迭代器，产生的是消息块
for chunk in chain.stream("写一段关于爱情的歌词，需要5句话,每句话用中文句号隔开"): # 实现流式传输
# chunk类型 就是一个AIMessageChunk
# print(chunk.content, end="|", flush=True) # 每个块的结果 使用 | 分割，再刷新
# 使用parser,结果就是str
print(chunk, end="|", flush=True)
# tmp_chunks = chunks[0] + chunks[1] + chunks[2] + chunks[3]
# print(tmp_chunks)深度探索流式传输
SSE协议介绍
HTTP 协议本⾝设计为⽆状态的请求-响应模式，严格来说，是⽆法做到服务器主动推送消息到客户 端，但通过 Server-Sent Events （服务器发送事件，简称 SSE）技术可实现流式传输，允许服 务器主动向浏览器推送数据流。
也就是说，服务器向客⼾端声明，接下来要发送的是流消息(streaming)，这时客⼾端不会关闭连接， 会⼀直等待服务器发送过来新的数据流。
SSE（Server-Sent Events）是⼀种基于 HTTP 的轻量级实时通信协议，浏览器可以通过内置的 EventSource API 接收并处理这些实时事件。
核心特点
• 基于 HTTP 协议
复⽤标准 HTTP/HTTPS 协议，⽆需额外端⼝或协议，兼容性好且易于部署。
• 单向通信机制
SSE 仅⽀持服务器向客⼾端的单向数据推送，客户端通过普通 HTTP 请求建立连接后，服务器可持续 发送数据流，但客户端无法通过同一连接向服务器发送数据。
• ⾃动重连机制
⽀持断线重连，连接中断时，浏览器会⾃动尝试重新连接（⽀持 retry 字段指定重连间隔）。
• ⾃定义消息类型
客⼾端发起请求后，服务器保持连接开放，响应头设置 Content-Type: text/event-stream ，标识为事件流格式，持续推送事件流。
websocket协议也能实现服务端相客户端推送信息，但是websocket建立的长连接需要服务端去管理连接，而SSE协议支持断开连接，无需管理，依旧是无状态协议。
数据格式
服务端向浏览器发送 SSE 数据，需要设置必要的 HTTP 头信息：
Content-Type: text/event-stream;charset=utf-8
Connection: keep-alive每⼀次发送的消息，由若⼲个 message 组成，每个 message 之间由 \n\n 分隔，每个 message 内 部由若干行组成，每一行都是如下格式：
[field]: value\nField 可以取值为:
• data[必需]：数据内容
• event[非必需]：表⽰⾃定义的事件类型，默认是message事件
• id[非必需]：数据标识符，相当于每⼀条数据的编号
• retry[非必需]：指定浏览器重新发起连接的时间间隔
1. langchain-openai 包通过集成 OpenAI Python SDK，提供了⼀个 HTTP 客⼾端。
2. 因此，⽀持 LangChain 向 OpenAI 的 API 发起调⽤请求。
3. 若希望发起流式传输请求，则需在请求中加⼊ stream=True ，向 OpenAI 说明以 SSE 协议进⾏流式返回。
4. LangChain 接收 OpenAI 的 SSE 格式的响应，并将其转换为 LangChain 自封装的消息格式，如 AIMessageChunk 消息。这样就可以以统⼀的⽅式处理来⾃不同模型提供商（OpenAI, Anthropic等）的流式响应。
使用 LangSmith 跟踪 LLM 应用
使用 LangChain 构建的许多应⽤程序，可能会包含多个步骤和多次的 LLM 调⽤。随着这些应⽤程序变 得越来越复杂，作为开发者，我们能够检查链或代理内部到底发⽣了什么变得⾄关重要。最好的⽅法 是使⽤ LangSmith。
LangSmith 与框架⽆关，它可以与 langchain 和 langgraph ⼀起使⽤，也可以不使⽤。LangSmith 是 ⼀个⽤于帮助我们构建⽣产级 LLM 应⽤程序的平台，它将密切监控和评估我们的应⽤。LangSmith官网
要想让 LangSmith 跟踪 LLM 应⽤，第⼀步申请 LangSmith API Key，点击 Settings，就会跳转 到"API Keys"设置⻚⾯，若没有跳转，可以在左侧 tab 栏中找到进⼊。
创建完成之后，需要配置环境变量
配置完之后，然后在代码中调用模型，在LangSmith中就能追踪到了
这是我之前就配好了的，所以内容就比较多。这里能实时监控到每次调用LLM的情况，对于调试代码和分析问题很方便
有“AI”的1024 = 2048，欢迎大家加入2048 AI社区
更多推荐
14
0- 0
已为社区贡献3条内容
所有评论(0)