如何将 Runnables 转换为 Tools
在这里,我们将演示如何将 LangChain Runnable 转换为可供 agents、chains 或 chat models 使用的 tool。
依赖
注意: 本指南需要 langchain-core >= 0.2.13。我们还将使用 OpenAI 进行 embeddings,但任何 LangChain embeddings 都应该足够。我们将使用一个简单的 LangGraph agent 作为演示。
%%capture --no-stderr
%pip install -U langchain-core langchain-openai langgraph
LangChain 的 tools 是代理、链或聊天模型可以用来与世界进行交互的接口。有关工具调用、内置工具、自定义工具等更多信息,请参阅 此处 的操作指南。
LangChain tools-- BaseTool 的实例-- 是具有附加约束的 Runnables,这些约束使它们能够被语言模型有效地调用:
- 它们的输入被限制为可序列化的,特别是字符串和 Python
dict对象; - 它们包含名称和描述,指明如何以及何时使用它们;
- 它们可能包含一个详细的 args_schema 来描述它们的参数。也就是说,虽然一个工具(作为
Runnable)可能接受单个dict输入,但填充dict所需的具体键和类型信息应在args_schema中指定。
接受字符串或 dict 输入的 Runnables 可以使用 as_tool 方法转换为工具,这允许指定名称、描述和额外的参数 schema 信息。
基本用法
使用类型化的 dict 输入:
from typing import List
from langchain_core.runnables import RunnableLambda
from typing_extensions import TypedDict
class Args(TypedDict):
a: int
b: List[int]
def f(x: Args) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(f)
as_tool = runnable.as_tool(
name="My tool",
description="Explanation of when to use tool.",
)
print(as_tool.description)
as_tool.args_schema.model_json_schema()
Explanation of when to use tool.
{'properties': {'a': {'title': 'A', 'type': 'integer'},
'b': {'items': {'type': 'integer'}, 'title': 'B', 'type': 'array'}},
'required': ['a', 'b'],
'title': 'My tool',
'type': 'object'}
as_tool.invoke({"a": 3, "b": [1, 2]})
'6'
在不键入信息的情况下,可以通过 arg_types 指定参数类型:
from typing import Any, Dict
def g(x: Dict[str, Any]) -> str:
return str(x["a"] * max(x["b"]))
runnable = RunnableLambda(g)
as_tool = runnable.as_tool(
name="My tool",
description="Explanation of when to use tool.",
arg_types={"a": int, "b": List[int]},
)
或者,可以通过直接传递工具所需的 args_schema 来完全指定 schema:
from pydantic import BaseModel, Field
class GSchema(BaseModel):
"""Apply a function to an integer and list of integers."""
a: int = Field(..., description="Integer")
b: List[int] = Field(..., description="List of ints")
runnable = RunnableLambda(g)
as_tool = runnable.as_tool(GSchema)
字符串输入也受支持:
def f(x: str) -> str:
return x + "a"
def g(x: str) -> str:
return x + "z"
runnable = RunnableLambda(f) | g
as_tool = runnable.as_tool()
as_tool.invoke("b")
'baz'
在 Agents 中
下面我们将 LangChain Runnables 作为工具集成到 agent 应用中。我们将通过以下方式进行演示:
- 文档 retriever;
- 一个简单的 RAG 链,允许 agent 将相关查询委托给它。
我们首先实例化一个支持 tool calling 的聊天模型:
pip install -qU "langchain[google-genai]"
import getpass
import os
if not os.environ.get("GOOGLE_API_KEY"):
os.environ["GOOGLE_API_KEY"] = getpass.getpass("Enter API key for Google Gemini: ")
from langchain.chat_models import init_chat_model
llm = init_chat_model("gemini-2.0-flash", model_provider="google_genai")
遵循 RAG 教程,我们首先构建一个检索器:
from langchain_core.documents import Document
from langchain_core.vectorstores import InMemoryVectorStore
from langchain_openai import OpenAIEmbeddings
documents = [
Document(
page_content="Dogs are great companions, known for their loyalty and friendliness.",
),
Document(
page_content="Cats are independent pets that often enjoy their own space.",
),
]
vectorstore = InMemoryVectorStore.from_documents(
documents, embedding=OpenAIEmbeddings()
)
retriever = vectorstore.as_retriever(
search_type="similarity",
search_kwargs={"k": 1},
)
我们接下来创建一个简单的预构建的 LangGraph 代理,并为其提供工具:
from langgraph.prebuilt import create_react_agent
tools = [
retriever.as_tool(
name="pet_info_retriever",
description="Get information about pets.",
)
]
agent = create_react_agent(llm, tools)
for chunk in agent.stream({"messages": [("human", "What are dogs known for?")]}):
print(chunk)
print("----")
{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_W8cnfOjwqEn4cFcg19LN9mYD', 'function': {'arguments': '{"__arg1":"dogs"}', 'name': 'pet_info_retriever'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 19, 'prompt_tokens': 60, 'total_tokens': 79}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-d7f81de9-1fb7-4caf-81ed-16dcdb0b2ab4-0', tool_calls=[{'name': 'pet_info_retriever', 'args': {'__arg1': 'dogs'}, 'id': 'call_W8cnfOjwqEn4cFcg19LN9mYD'}], usage_metadata={'input_tokens': 60, 'output_tokens': 19, 'total_tokens': 79})]}}
----
{'tools': {'messages': [ToolMessage(content="[Document(id='86f835fe-4bbe-4ec6-aeb4-489a8b541707', page_content='Dogs are great companions, known for their loyalty and friendliness.')]", name='pet_info_retriever', tool_call_id='call_W8cnfOjwqEn4cFcg19LN9mYD')]}}
----
{'agent': {'messages': [AIMessage(content='Dogs are known for being great companions, known for their loyalty and friendliness.', response_metadata={'token_usage': {'completion_tokens': 18, 'prompt_tokens': 134, 'total_tokens': 152}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-9ca5847a-a5eb-44c0-a774-84cc2c5bbc5b-0', usage_metadata={'input_tokens': 134, 'output_tokens': 18, 'total_tokens': 152})]}}
----
请参阅LangSmith trace 以了解上述运行。
更进一步,我们可以创建一个简单的 RAG 链,它接受一个附加参数——在这里是答案的“风格”。
from operator import itemgetter
from langchain_core.output_parsers import StrOutputParser
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import RunnablePassthrough
system_prompt = """
You are an assistant for question-answering tasks.
Use the below context to answer the question. If
you don't know the answer, say you don't know.
Use three sentences maximum and keep the answer
concise.
Answer in the style of {answer_style}.
Question: {question}
Context: {context}
"""
prompt = ChatPromptTemplate.from_messages([("system", system_prompt)])
rag_chain = (
{
"context": itemgetter("question") | retriever,
"question": itemgetter("question"),
"answer_style": itemgetter("answer_style"),
}
| prompt
| llm
| StrOutputParser()
)
请注意,我们 chain 的输入 schema 包含了必需的参数,因此在没有进一步指定的情况下,它会被转换为一个工具:
rag_chain.input_schema.model_json_schema()
{'properties': {'question': {'title': 'Question'},
'answer_style': {'title': 'Answer Style'}},
'required': ['question', 'answer_style'],
'title': 'RunnableParallel<context,question,answer_style>Input',
'type': 'object'}
rag_tool = rag_chain.as_tool(
name="pet_expert",
description="Get information about pets.",
)
下面我们再次调用代理。请注意,代理在其 tool_calls 中填充了必需的参数:
agent = create_react_agent(llm, [rag_tool])
for chunk in agent.stream(
{"messages": [("human", "What would a pirate say dogs are known for?")]}
):
print(chunk)
print("----")
{'agent': {'messages': [AIMessage(content='', additional_kwargs={'tool_calls': [{'id': 'call_17iLPWvOD23zqwd1QVQ00Y63', 'function': {'arguments': '{"question":"What are dogs known for according to pirates?","answer_style":"quote"}', 'name': 'pet_expert'}, 'type': 'function'}]}, response_metadata={'token_usage': {'completion_tokens': 28, 'prompt_tokens': 59, 'total_tokens': 87}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': None, 'finish_reason': 'tool_calls', 'logprobs': None}, id='run-7fef44f3-7bba-4e63-8c51-2ad9c5e65e2e-0', tool_calls=[{'name': 'pet_expert', 'args': {'question': 'What are dogs known for according to pirates?', 'answer_style': 'quote'}, 'id': 'call_17iLPWvOD23zqwd1QVQ00Y63'}], usage_metadata={'input_tokens': 59, 'output_tokens': 28, 'total_tokens': 87})]}}
----
{'tools': {'messages': [ToolMessage(content='"Dogs are known for their loyalty and friendliness, making them great companions for pirates on long sea voyages."', name='pet_expert', tool_call_id='call_17iLPWvOD23zqwd1QVQ00Y63')]}}
----
{'agent': {'messages': [AIMessage(content='According to pirates, dogs are known for their loyalty and friendliness, making them great companions for pirates on long sea voyages.', response_metadata={'token_usage': {'completion_tokens': 27, 'prompt_tokens': 119, 'total_tokens': 146}, 'model_name': 'gpt-4o-mini', 'system_fingerprint': None, 'finish_reason': 'stop', 'logprobs': None}, id='run-5a30edc3-7be0-4743-b980-ca2f8cad9b8d-0', usage_metadata={'input_tokens': 119, 'output_tokens': 27, 'total_tokens': 146})]}}
----
请参阅以上运行的 LangSmith 跟踪。