每个用户使用的带LangChain封装中间件的登机口代理工具
一个计算器智能体,三种访问层级:LangChain的封装式中间件会根据运行时的用户类型切换工具列表,而非创建三个独立的智能体。
根据用户上下文,在运行时动态控制哪些智能体工具可见。
人工智能智能体通过各种工具与外部世界交互——这些工具是用于运算、搜索或其他操作的简单功能。默认情况下,与该智能体的每次对话都会看到相同的工具列表。但这往往不符合产品的实际需求。大楼的大厅对访客开放,办公室供员工使用,而服务器机房仅限管理员进入。同一栋建筑,不同的权限。
这一理念同样适用于智能体:一个智能体定义,但不同调用者可使用不同的工具集。
示例
一个计算器智能体提供了四种工具:加法、减法、乘法和除法。
共有三种访问层级:
- 内部——加法和减法
- 外部——仅加法
- 系统管理员——全部四种
显而易见(但糟糕的)方法
为每个层级创建三个独立的智能体。这种方式在系统提示、模型设置或工具实现未发生变化时有效。任何更改都必须重复三次,这种重复操作显然不是明智之举。
更好的方案:包装式中间件
只需保留一个智能体。在调用模型之前加入一个小型的检查点。当有请求到来时,检查点会识别请求方身份,并在模型运行前仅附加允许使用的工具。
这个检查点就是中间件。包装式意味着它围绕正常的模型调用流程:先检查请求,进行相应调整,然后再继续执行。
实现思路
导入内容
常用的组件包括:
create_agent—— 用于构建智能体@wrap_model_call—— 包装式中间件装饰器
ModelRequest / ModelResponse — 输入请求与输出响应的结构HumanMessage — 用户输入的文本@tool — 将函数转换为工具API密钥
openai_api_key = os.getenv("OPENAI_API_KEY")
if not openai_api_key:
raise ValueError(
"Please set the OPENAI_API_KEY environment variable."
)
运行时上下文
user_type用于指定访问级别:internal、external或sys_admin。调用时需传入该值,例如context={"user_type": "internal"}。
工具
将四个计算工具定义为经过装饰的普通函数。
中间件
@wrap_model_call
def dynamic_tools(
request: ModelRequest,
handler: Callable[[ModelRequest], ModelResponse],
) -> ModelResponse:
user_type = request.runtime.context["user_type"]
if user_type == "internal":
tools = [
addition,
subtraction,
]
elif user_type == "external":
tools = [
addition,
]
elif user_type == "sys_admin":
tools = [
addition,
subtraction,
multiplication,
division,
]
else:
raise ValueError(f"Unknown user type: {user_type}")
# Dynamically change the tools available to the model
modified_request = request.override(
tools=tools
)
return handler(modified_request)
包装器内的处理流程:
- 从
request.runtime.context中读取user_type - 选择该类型允许使用的工具列表
request.override(tools=tools) 用新列表克隆请求handler(),以便模型调用继续进行override() 会复制请求,仅替换你指定的字段——此处为 tools。
聊天模型
temperature=0 能让演示结果具有确定性。
创建智能体
在智能体上注册所有四个工具。中间件会决定模型每次请求实际看到哪些工具。系统提示应指导模型使用可用工具,并在所需操作不可用时说明,而非自行编造计算结果。
共享测试提示
所有层级都使用相同的 HumanMessage,仅上下文中的 user_type 会发生变化。
调用
## INTERNAL USER
print("\n" + "=" * 70)
print("USER TYPE: INTERNAL")
print("=" * 70)
result = agent.invoke(
{
"messages": [human_message]
},
context={
"user_type": "internal"
},
)
print(result["messages"][-1].content)
## EXTERNAL USER
print("\n" + "=" * 70)
print("USER TYPE: EXTERNAL")
print("=" * 70)
result = agent.invoke(
{
"messages": [human_message]
},
context={
"user_type": "external"
},
)
print(result["messages"][-1].content)
## SYS_ADMIN USER
print("\n" + "=" * 70)
print("USER TYPE: SYS_ADMIN")
print("=" * 70)
result = agent.invoke(
{
"messages": [human_message]
},
context={
"user_type": "sys_admin"
},
)
print(result["messages"][-1].content)
演示输出
在三种不同场景下使用相同消息,即可看出中间件正在正常发挥作用:
- 内部用户可以执行加法和减法,但不能进行乘法或除法运算
- 外部用户只能执行加法运算
- 系统管理员可以使用全部四种运算方式
核心模式
其原理可以用几句话概括:
@wrap_model_call
def dynamic_tools(request, handler):
user_type = request.runtime.context["user_type"]
tools = choose_tools(user_type)
modified_request = request.override(
tools=tools
)
return handler(modified_request)
中间件会封装模型调用过程,读取当前用户类型,选择相应的工具,对请求进行修改后再转发。在本示例中只有工具列表会发生变化——这已足以体现封装式控制的效果。
要点总结
封装式中间件能够让产品在模型接收到请求之前对其进行重新处理。此处通过user_type来限制可使用的工具,从而使一个代理能够服务于多个访问层级。最关键的代码行是:
modified_request = request.override(tools=tools)
这样每次请求都会生成一个ModelRequest,其工具界面会根据调用场景自动调整——无论是大厅、办公室还是服务器机房,而无需使用三个重复的智能体。
在正式环境中,应结合身份验证机制,根据经过核实的身份来设置user_type,而不能仅依赖客户端提供的字符串信息。记录每个请求所使用的工具也有助于在有人声称模型“本应”拥有某种特权工具时进行审计。请保持允许列表的规模较小,并在计算器(或任何其他)智能体中新增工具时及时审查这些列表。