如何运行自定义函数
先决条件
本指南假设你熟悉以下概念
你可以使用任意函数作为 Runnable。这在格式化或需要其他 LangChain 组件未提供的功能时很有用，用作 Runnable 的自定义函数称为 RunnableLambdas。
请注意，这些函数的所有输入都需要是单个参数。如果你有一个接受多个参数的函数，你应该编写一个包装器，该包装器接受一个单个字典输入并将其解包为多个参数。
本指南将涵盖
- 如何使用
RunnableLambda构造函数从自定义函数显式创建 Runnable - 在链中使用自定义函数时的强制转换
- 如何接受并在自定义函数中使用运行元数据
- 如何通过使自定义函数返回生成器来使用流式传输
使用构造函数
下面，我们使用 RunnableLambda 方法显式包装我们的自定义逻辑
提示
参见 有关安装集成包的一般说明的此部分。
- npm
- yarn
- pnpm
npm i @langchain/openai @langchain/core
yarn add @langchain/openai @langchain/core
pnpm add @langchain/openai @langchain/core
import { StringOutputParser } from "@langchain/core/output_parsers";
import { ChatPromptTemplate } from "@langchain/core/prompts";
import { RunnableLambda } from "@langchain/core/runnables";
import { ChatOpenAI } from "@langchain/openai";
const lengthFunction = (input: { foo: string }): { length: string } => {
return {
length: input.foo.length.toString(),
};
};
const model = new ChatOpenAI({ model: "gpt-4o" });
const prompt = ChatPromptTemplate.fromTemplate("What is {length} squared?");
const chain = RunnableLambda.from(lengthFunction)
.pipe(prompt)
.pipe(model)
.pipe(new StringOutputParser());
await chain.invoke({ foo: "bar" });
"3 squared is \\(3^2\\), which means multiplying 3 by itself. \n" +
"\n" +
"\\[3^2 = 3 \\times 3 = 9\\]\n" +
"\n" +
"So, 3 squared"... 6 more characters
链中的自动强制转换
在使用自定义函数与 RunnableSequence.from 静态方法一起在链中时，你可以省略显式的 RunnableLambda 创建并依靠强制转换。
这是一个使用函数的简单示例，该函数获取模型的输出并返回其前五个字母
import { RunnableSequence } from "@langchain/core/runnables";
const storyPrompt = ChatPromptTemplate.fromTemplate(
"Tell me a short story about {topic}"
);
const storyModel = new ChatOpenAI({ model: "gpt-4o" });
const chainWithCoercedFunction = RunnableSequence.from([
storyPrompt,
storyModel,
(input) => input.content.slice(0, 5),
]);
await chainWithCoercedFunction.invoke({ topic: "bears" });
"Once "
请注意，我们不需要将自定义函数 (input) => input.content.slice(0, 5) 包装在 RunnableLambda 方法中。自定义函数被强制转换为 Runnable。有关更多信息，请参见 此部分。
传递运行元数据
Runnable lambda 可以选择接受 RunnableConfig 参数，它们可以使用该参数将回调、标签和其他配置信息传递给嵌套运行。
import { type RunnableConfig } from "@langchain/core/runnables";
const echo = (text: string, config: RunnableConfig) => {
const prompt = ChatPromptTemplate.fromTemplate(
"Reverse the following text: {text}"
);
const model = new ChatOpenAI({ model: "gpt-4o" });
const chain = prompt.pipe(model).pipe(new StringOutputParser());
return chain.invoke({ text }, config);
};
const output = await RunnableLambda.from(echo).invoke("foo", {
tags: ["my-tag"],
callbacks: [
{
handleLLMEnd: (output) => console.log(output),
},
],
});
{
generations: [
[
{
text: "oof",
message: AIMessage {
lc_serializable: true,
lc_kwargs: [Object],
lc_namespace: [Array],
content: "oof",
name: undefined,
additional_kwargs: [Object],
response_metadata: [Object],
tool_calls: [],
invalid_tool_calls: []
},
generationInfo: { finish_reason: "stop" }
}
]
],
llmOutput: {
tokenUsage: { completionTokens: 2, promptTokens: 13, totalTokens: 15 }
}
}
流式传输
你可以在链中使用生成器函数（即使用 yield 关键字且行为类似迭代器的函数）。
这些生成器的签名应该是 AsyncGenerator<Input> -> AsyncGenerator<Output>。
这些对于以下情况很有用： - 实现自定义输出解析器 - 修改先前步骤的输出，同时保留流式传输功能
以下是用逗号分隔的列表的自定义输出解析器示例。首先，我们创建一个生成此类列表作为文本的链
const streamingPrompt = ChatPromptTemplate.fromTemplate(
"Write a comma-separated list of 5 animals similar to: {animal}. Do not include numbers"
);
const strChain = streamingPrompt.pipe(model).pipe(new StringOutputParser());
const stream = await strChain.stream({ animal: "bear" });
for await (const chunk of stream) {
console.log(chunk);
}
Lion
,
wolf
,
tiger
,
cougar
,
leopard
接下来，我们定义一个自定义函数，该函数将聚合当前流式传输的输出并在模型生成列表中的下一个逗号时将其生成
// This is a custom parser that splits an iterator of llm tokens
// into a list of strings separated by commas
async function* splitIntoList(input) {
// hold partial input until we get a comma
let buffer = "";
for await (const chunk of input) {
// add current chunk to buffer
buffer += chunk;
// while there are commas in the buffer
while (buffer.includes(",")) {
// split buffer on comma
const commaIndex = buffer.indexOf(",");
// yield everything before the comma
yield [buffer.slice(0, commaIndex).trim()];
// save the rest for the next iteration
buffer = buffer.slice(commaIndex + 1);
}
}
// yield the last chunk
yield [buffer.trim()];
}
const listChain = strChain.pipe(splitIntoList);
const listChainStream = await listChain.stream({ animal: "bear" });
for await (const chunk of listChainStream) {
console.log(chunk);
}
[ "wolf" ]
[ "lion" ]
[ "tiger" ]
[ "cougar" ]
[ "cheetah" ]
调用它将提供完整的值数组
await listChain.invoke({ animal: "bear" });
[ "lion", "tiger", "wolf", "cougar", "jaguar" ]
下一步
现在，你已经了解了在你的链中使用自定义逻辑和实现流式传输的几种不同方法。
要了解更多信息，请参见本部分中有关 Runnable 的其他操作指南。