langchain agent的工具
本文介绍了使用@tool注解创建和定制LangChain工具的方法。主要内容包括:1)通过@tool注解快速创建工具函数;2)工具定制方法(修改名称、增加描述、规范化输入模式);3)工具访问上下文的实现方式(运行时上下文、长期记忆存储、实时反馈)。文中提供了创建电子印章工具和查询企业信息工具的具体代码示例,展示了如何通过BaseModel或JSON模式定义数据结构,以及如何使用ToolRuntim
1.创建工具
直接使用@tool注解一个标准的函数,就可以创建一个可以被agent使用的工具。如下代码是基于一个创建电子印章的简单示例方法创建的工具,具体如下:
import random
from langchain.tools import tool, ToolRuntime
@tool
def make_seal_tool(name: str, type: str)->str:
"""to make specific seal for someon"""
rnd = random.randint(1, 4)
match rnd:
case 1|2:
return f"{name}的{type} 制作成功, 印章图片的base64编码为"
case 3:
return f"{name}的{type} 制作失败,请求第三方数字证书系统失败."
case 4:
return f"{name}的{type} 制作失败, 系统内部错误"
case _:
return f"{name}的{type} 制作成功, 印章图片的base64编码为"
2.对工具做定制
2.1修改工具名称
使用@tool注解的工具,名称与函数名相同,可以增加注解属性修改工具名称,比如把make_seal_tool修改为makeSealTool,具体代码如下:
@tool(name='makeSealTool')
def make_seal_tool(name: str, type: str)->str:
"""to make specific seal for someone"""
……
2.2增加工具描述
在@tool注解中,使用description可以增加工具的描述,比如:
@tool(description='perform seal makeing,support four types, includeing ……')
def make_seal_tool(name: str, type: str)->str:
"""to make specific seal for someone"""
……
2.3规范化输入模式
很多情况下需要对数据结构重用,比如上面例子中的印章申请信息,封装到一个结构中。该结构既可以作为函数的输入,也可以在其他的地方使用。langchain agent支持两种类型的数据结构,基于BaseModel或者JSON格式。以下分别示例说明。
2.3.1BaseModel模式
参数结构继承BaseModel,具体如下:
class SealPySchema(BaseModel):
name: str = Field(description='enterprise name')
type: str = Field(description='seal type')
在注解中使用args_schema属性定义工具输入参数类型:
@tool(args_schema=SealPySchema)
def make_seal_tool(name: str, type: str)->str:
"""to make specific seal for someon"""
rnd = random.randint(1, 4)
match rnd:
case 1|2:
return f"{name}的{type} 制作成功, 印章图片的base64编码为"
case 3:
return f"{name}的{type} 制作失败,请求第三方数字证书系统失败."
case 4:
return f"{name}的{type} 制作失败, 系统内部错误"
case _:
return f"{name}的{type} 制作成功, 印章图片的base64编码为"
2.3.2JSON模式
JSON模式封装数据结构如下:
seal_json_schema = {
"type": "object",
"properties": {
"name": {"type": "string"},
"type": {"type": "string"},
},
"required": ["name", "type"]
}
在注解中使用args_schema属性定义工具输入参数类型:
@tool(args_schema=seal_schema)
def make_seal_tool(name: str, type: str)->str:
"""to make specific seal for someon"""
rnd = random.randint(1, 4)
match rnd:
case 1|2:
return f"{name}的{type} 制作成功, 印章图片的base64编码为"
case 3:
return f"{name}的{type} 制作失败,请求第三方数字证书系统失败."
case 4:
return f"{name}的{type} 制作失败, 系统内部错误"
case _:
return f"{name}的{type} 制作成功, 印章图片的base64编码为"
2.4访问上下文
工具内部可以通过ToolRuntime访问短期记忆、长期记忆、运行时上下文数据,从而可以根据运行的实时数据进行决策、个性化响应,实现跨对话数据共享,还可以实时反馈工具的执行情况。其中访问短期记忆直接参见langchain agent中的短期记忆的4.1,其他内容在下面说明
2.4.1运行时上下文
以下示例中定义了一个工具,该工具根据统一社会信用代码从数据库中查找对应的企业的法人姓名和公司名称。创建agent时指定上下文,调用agent时传入上下文数据,具体代码如下:
from dataclasses import dataclass
from langchain.agents import create_agent
@dataclass
class Context:
uniscid: str
ENTERPRISE_DATABASE = {
"51234569876543210": {
"name": "东成西就文化娱乐有限公司",
"legal": "欧阳锋",
"type": "内资企业"
},
"37654321123456789S": {
"name": "走南闯北旅游有限公司",
"legal": "徐霞客",
"type": "有限合伙"
}
}@tool
def get_enterprise_info(runtime: ToolRuntime[Context]) -> str:
"""Get the current enterprise's information."""
uniscid = runtime.context.uniscidif uniscid in ENTERPRISE_DATABASE:
enterprise = ENTERPRISE_DATABASE[uniscid]
return f"Legal name: {enterprise['legal']}\nType: {enterprise['type']}\nName: ${enterprise['name']}"
return f"enterprise {uniscid} not found"agent = create_agent(
model=llm,
tools=[get_enterprise_info],
context_schema=Context,
)result = agent.invoke(
{"messages": [{"role": "user", "content": "Who's enterprise's legal person and what's enterprise's name?"}]},
context=Context(uniscid="51234569876543210")
)
2.4.2长期记忆
与图一样,长期记忆使用Store,开发和测试基于内存模式,线上系统必须基于数据库,以下示例基于内存模式。
示例中第一次调用大模型把企业信息保存在长期记忆中,第二次调用从长期记忆检索获取企业数据。在创建agent时传入工具集合长期记忆。在工具中通过runtime访问长期记忆。具体代码如下:
from typing import Any
from langgraph.store.memory import InMemoryStore
from langchain.agents import create_agent
from langchain.tools import tool, ToolRuntime"""
以下是两个工具,分别用于保存数据和检索数据。
"""
@tool
def get_enterprise_info(uniscid: str, runtime: ToolRuntime) -> dict[str, Any]:
"""Look up enterprise info."""
store = runtime.store #通过runtime获取长期记忆
enterprise_info = store.get(("enterprises",), uniscid) #从长期记忆获取企业数据
return enterprise_info if enterprise_info else "Unknown enterprise"@tool
def save_enterprise_info(uniscid: str, enterprise_info: dict[str, Any], runtime: ToolRuntime) -> str:
"""Save enterprise info."""
store = runtime.store #通过 runtime获取长期记忆
store.put(("enterprises",), uniscid, enterprise_info) #把企业数据保存到长期记忆中
return "Successfully saved enterprise info."store = InMemoryStore()
agent = create_agent(
model=llm,
tools=[get_enterprise_info, save_enterprise_info],
store=store
)# 第一次调用,保存企业信息到长期记忆
agent.invoke({
"messages": [
{
"role": "user", "content": "Save the following enterprise: uniscid: 51234569876543210,"
"name: 东成西就文化娱乐有限公司, legal: 欧阳锋, type: 内资企业"
}
]
})#第二次调用从长期记忆检索数据
result = agent.invoke({
"messages": [{"role": "user", "content": "Get enterprise info for enterprse with id '51234569876543210'"}]
})
2.4.3运行情况实时反馈
通过上下文获取到stream_writer,调用stream_writer把工具运行的中间结果实时写入到流中。基于上面的获取企业信息工具,在其中输出中间结果,具体示例代码如下:
@tool
def get_enterprise_info(uniscid: str, runtime: ToolRuntime) -> dict[str, Any]:
"""Look up enterprise info."""
store = runtime.store
enterprise_info = store.get(("enterprises",), uniscid)
writer = runtime.stream_writer
writer(f"####{enterpriese_info['legal']=}####")
writer(f"####{enterpriese_info['name']=}####")
return enterprise_info if enterprise_info else "Unknown enterprise"
当使用stream_writer时,必须使用流模式调用agent并且stream_mode中必须有custom。
更多推荐



所有评论(0)