在开发 LangChain 智能体时,我们常常面临这样的安全困境:如何将用户 ID、API 密钥等敏感信息传递给工具,同时防止 LLM 擅自修改这些关键参数?又该如何优雅处理工具调用过程中可能出现的各种异常?今天我们就来系统拆解运行时参数注入的核心技术与错误处理的最佳实践,帮助你构建更安全、更健壮的智能体应用。

一、运行时参数注入:安全与控制的平衡艺术

为什么需要运行时参数注入?

想象一个金融咨询智能体,当用户查询账户信息时,工具需要用户 ID 来定位数据,但这个 ID 绝不能由 LLM 生成 —— 否则可能导致越权访问。LangChain 的运行时参数注入机制就像一个安全闸门,让我们能够:

  • 隔离敏感参数:将用户 ID、API 密钥等敏感信息与 LLM 的控制范围隔离
  • 精准权限控制:由应用逻辑而非 LLM 决定哪些参数可以被修改
  • 防御安全风险:避免 LLM 生成恶意参数导致的数据泄露

核心实现:InjectedToolArg 注解的妙用

基础用法:标记运行时注入参数

python

运行

from typing import List
from langchain_core.tools import InjectedToolArg, tool
from typing_extensions import Annotated

user_to_pets = {}

@tool(parse_docstring=True)
def update_favorite_pets(
    pets: List[str], 
    user_id: Annotated[str, InjectedToolArg]
) -> None:
    """添加用户喜欢的宠物列表
    Args:
        pets: 要设置的宠物列表
        user_id: 用户唯一标识
    """
    user_to_pets[user_id] = pets

@tool(parse_docstring=True)
def list_favorite_pets(
    user_id: Annotated[str, InjectedToolArg]
) -> List[str]:
    """查询用户喜欢的宠物列表
    Args:
        user_id: 用户唯一标识
    """
    return user_to_pets.get(user_id, [])
参数模式对比:输入与调用的区别
  • 输入模式(供工具内部使用):包含所有参数

python

运行

print(update_favorite_pets.get_input_schema().model_json_schema())

json

{
  "properties": {
    "pets": {"type": "array"},
    "user_id": {"type": "string"}
  },
  "required": ["pets", "user_id"]
}

  • 调用模式(供 LLM 生成工具调用):自动移除注入参数

python

运行

print(update_favorite_pets.tool_call_schema.model_json_schema())

json

{
  "properties": {"pets": {"type": "array"}},
  "required": ["pets"]
}
运行时注入实战:构建安全的宠物管理链

python

运行

# 1. 定义用户ID
user_id = "123"

# 2. 模型生成工具调用(不包含user_id)
from langchain.chat_models import init_chat_model
llm = init_chat_model("gemini-2.0-flash")
tools = [update_favorite_pets, list_favorite_pets]
llm_with_tools = llm.bind_tools(tools)
ai_msg = llm_with_tools.invoke("我喜欢的动物是猫和鹦鹉")

# 3. 注入user_id并执行工具
from copy import deepcopy
from langchain_core.runnables import chain

@chain
def inject_user_id(ai_msg):
    tool_calls = []
    for call in ai_msg.tool_calls:
        call_copy = deepcopy(call)
        call_copy["args"]["user_id"] = user_id
        tool_calls.append(call_copy)
    return tool_calls

# 4. 构建完整执行链
tool_map = {tool.name: tool for tool in tools}
@chain
def tool_router(tool_call):
    return tool_map[tool_call["name"]]

full_chain = llm_with_tools | inject_user_id | tool_router.map()
full_chain.invoke("我喜欢的动物是猫和鹦鹉")

# 5. 验证结果
print(user_to_pets)  # 输出: {'123': ['猫', '鹦鹉']}

二、多种参数注解方式:灵活适配不同场景

Pydantic 模型方式:更规范的参数定义

python

运行

from pydantic import BaseModel, Field
from langchain_core.tools import tool

class PetSchema(BaseModel):
    pets: List[str] = Field(..., description="宠物列表")
    user_id: Annotated[str, InjectedToolArg] = Field(..., description="用户ID")

@tool(args_schema=PetSchema)
def update_favorite_pets(pets, user_id):
    user_to_pets[user_id] = pets

继承 BaseTool 方式:高度定制的参数控制

python

运行

from langchain_core.tools import BaseTool
from typing import List

class UpdatePetTool(BaseTool):
    name = "update_favorite_pets"
    description = "更新用户宠物列表"
    
    def _run(
        self, 
        pets: List[str], 
        user_id: Annotated[str, InjectedToolArg]
    ) -> None:
        user_to_pets[user_id] = pets
        
    async def _arun(
        self, 
        pets: List[str], 
        user_id: Annotated[str, InjectedToolArg]
    ) -> None:
        self._run(pets, user_id)

三、机密信息传递:安全与可追溯的平衡

RunnableConfig 传递机密:避免日志泄露

python

运行

from langchain_core.runnables import RunnableConfig
from langchain_core.tools import tool

@tool
def secure_tool(x: int, config: RunnableConfig) -> int:
    """安全工具:使用机密参数"""
    return x + config["configurable"]["__top_secret_int"]

# 调用时传递机密参数
result = secure_tool.invoke(
    {"x": 5},
    {"configurable": {"__top_secret_int": 2, "traced_key": "公开参数"}}
)
print(result)  # 输出: 7

安全特性说明:

  1. 前缀保护__前缀的参数不会被 LangSmith 追踪
  2. 分层存储:机密与普通参数分离存储
  3. 最小暴露:仅在工具执行时可见,不参与 LLM 推理

四、工具错误处理:构建健壮的智能体应用

try/except 基础错误处理:优雅返回错误信息

python

运行

from langchain_core.runnables import Runnable
from langchain_core.tools import tool

# 故意设计复杂工具以触发错误
@tool
def complex_tool(int_arg: int, float_arg: float, dict_arg: dict) -> int:
    """复杂工具:需要三个参数"""
    return int_arg * float_arg

# 构建错误处理链
def try_except_tool(tool_args: dict) -> str:
    try:
        return complex_tool.invoke(tool_args)
    except Exception as e:
        return f"调用工具时出错:{type(e).__name__}: {str(e)}"

chain = llm.bind_tools([complex_tool]) | (lambda msg: msg.tool_calls[0]["args"]) | try_except_tool

# 测试错误场景
print(chain.invoke("使用复杂工具,参数5, 2.1,空字典"))

模型回退策略:自动切换到更强大的模型

python

运行

from langchain_openai import ChatOpenAI

# 主链:使用基础模型
main_chain = llm.bind_tools([complex_tool]) | (lambda msg: msg.tool_calls[0]["args"]) | complex_tool

# 回退链:使用更强模型
fallback_chain = ChatOpenAI(model="gpt-4").bind_tools([complex_tool]) | \
                 (lambda msg: msg.tool_calls[0]["args"]) | complex_tool

# 组合链:主链失败时自动切换
robust_chain = main_chain.with_fallbacks([fallback_chain])

# 执行并验证
result = robust_chain.invoke("使用复杂工具,参数5, 2.1,空字典")
print(result)  # 输出: 10.5

异常重试机制:自动纠正工具调用错误

python

运行

from langchain_core.messages import HumanMessage, ToolMessage
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.runnables import chain

# 自定义异常类
class ToolCallException(Exception):
    def __init__(self, tool_call, error):
        self.tool_call = tool_call
        self.error = error

# 工具调用函数:抛出自定义异常
def tool_invoke(tool_args):
    try:
        return complex_tool.invoke(tool_args)
    except Exception as e:
        raise ToolCallException(tool_args, e)

# 异常转换函数:生成纠错提示
def exception_to_prompt(inputs):
    exception = inputs.pop("exception")
    messages = [
        HumanMessage(content="上次工具调用出错,请纠正参数后重试"),
        ToolMessage(
            tool_call_id="123",
            content=f"错误详情: {str(exception.error)}"
        )
    ]
    inputs["last_output"] = messages
    return inputs

# 构建自纠正链
prompt = ChatPromptTemplate.from_messages([
    ("human", "{input}"),
    ("placeholder", "{last_output}")
])

self_correcting_chain = prompt | llm.bind_tools([complex_tool]) | tool_invoke
self_correcting_chain = self_correcting_chain.with_fallbacks(
    [exception_to_prompt | self_correcting_chain], 
    exception_key="exception"
)

# 执行并验证
result = self_correcting_chain.invoke({
    "input": "使用复杂工具,参数5, 2.1,空字典"
})
print(result)  # 输出: 10.5

五、最佳实践与安全指南

参数注入最佳实践:

  1. 最小权限原则:仅注入必要的参数,避免过度暴露
  2. 类型明确化:使用类型注解确保参数类型正确
  3. 环境隔离:通过环境变量而非硬编码传递敏感信息

错误处理最佳实践:

  1. 分级处理:区分业务错误与系统错误,采取不同处理策略
  2. 错误上下文:在错误信息中包含足够的上下文,便于调试
  3. 有限重试:设置合理的重试次数,避免无限循环

结语:构建安全健壮的智能体系统

通过运行时参数注入与完善的错误处理机制,我们终于解决了智能体开发中的安全与稳定性难题。这些技术不仅能防止敏感信息泄露,还能大幅提升智能体的鲁棒性,使其在复杂业务场景中可靠运行。

如果本文对你有帮助,别忘了点赞收藏,关注我,一起探索更高效的开发方式~

Logo

脑启社区是一个专注类脑智能领域的开发者社区。欢迎加入社区,共建类脑智能生态。社区为开发者提供了丰富的开源类脑工具软件、类脑算法模型及数据集、类脑知识库、类脑技术培训课程以及类脑应用案例等资源。

更多推荐