多租户环境下LangChain Agent隔离性设计：基于上下文的工具权限控制

一、多租户环境下LangChain Agent的核心痛点

很多做SaaS服务的朋友应该都碰到过这么个问题：同一个后台给几百上千个客户用，每个客户都有自己的专属数据、专属工具，要是没做好隔离，A客户能看到B客户的隐私数据，那麻烦可就大了。比如做智能客服SaaS，甲公司的客服Agent能调用乙公司的订单接口，那订单数据乱传，客户肯定得跑路。

之前不少人做多租户隔离，都是简单给每个客户开个单独的数据库，或者给每个Agent绑个专属的API密钥，但这方法有俩大问题：一是成本高，客户多了之后，每个客户一套独立的Agent、一套独立的密钥，运维起来头大；二是权限粒度太粗，没法做到同一客户下，不同场景的Agent用不同的工具——比如某客户的财务Agent能调用财务工具，但不能调用营销工具，之前的方法做不到这么细。

LangChain Agent是现在很火的智能体框架，能让大模型调用各种工具完成任务，但原生的LangChain Agent没有专门针对多租户的隔离设计，直接用的话，很容易出现权限混乱的问题。

二、基于上下文的工具权限控制核心思路

我们要解决的核心问题，就是让每个Agent只能用自己能用到的工具，不能碰别人的东西，也不能碰自己不该碰的东西。那怎么实现呢？核心思路就是给每个Agent的“上下文”加个权限标签，这个标签就像Agent的“身份证”，里面包含了Agent所属的客户（租户）、所属的场景、能调用的工具范围。

举个例子，某客户有两个Agent：一个是财务Agent，一个是营销Agent。财务Agent的上下文标签里写着“租户ID：T1001，场景：财务，允许调用的工具：财务报表工具、发票查询工具”；营销Agent的上下文标签里写着“租户ID：T1001，场景：营销，允许调用的工具：客户画像工具、短信群发工具”。当Agent要调用工具的时候，先查自己的上下文标签，要是工具不在允许范围里，就不让调用。

这种方法的好处是，不用给每个客户单独部署一套Agent，所有客户的Agent可以共享一套基础代码，只需要给每个Agent分配不同的上下文标签就行，成本低、运维方便，而且权限粒度能做到很细。

三、具体实现步骤（含完整示例）

接下来我们用Python来做个完整的实现，先明确一下技术栈：Python 3.9+、LangChain 0.1.0+、FastAPI（用来做Agent的调用入口）、Redis（用来存上下文标签）。

3.1 准备工作：先搞懂LangChain的工具调用逻辑

LangChain的Agent调用工具的时候，会走这么个流程：大模型先判断需要调用什么工具，然后把工具名和参数传给Agent，Agent再去调用对应的工具，最后把工具返回的结果给大模型。我们要做的，就是在“Agent调用工具”这一步加个权限校验，只有符合上下文标签的工具才能被调用。

3.2 步骤1：定义上下文标签的存储结构

我们先定义一个上下文标签的结构，里面要包含租户ID、场景、允许调用的工具列表，还有过期时间（防止旧的标签一直有效）。用Redis存这个标签的好处是，读写速度快，还能设置过期时间，适合存这种临时的、需要频繁访问的信息。

先写个Redis的操作工具类，用来存和取上下文标签：

# 技术栈：Python 3.9+, LangChain 0.1.0+, FastAPI, Redis
import redis
from typing import List, Dict
import json

# 初始化Redis连接，这里假设Redis本地运行，端口6379
r = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

class ContextTag:
    def __init__(self, tenant_id: str, scene: str, allowed_tools: List[str], expire_seconds: int = 3600):
        self.tenant_id = tenant_id  # 租户ID，比如T1001、T1002
        self.scene = scene  # 场景，比如财务、营销
        self.allowed_tools = allowed_tools  # 允许调用的工具列表
        self.expire_seconds = expire_seconds  # 标签过期时间，默认1小时

    def save(self, agent_id: str) -> None:
        """把上下文标签存到Redis，键是agent_id，值是标签的JSON字符串"""
        tag_data = {
            "tenant_id": self.tenant_id,
            "scene": self.scene,
            "allowed_tools": self.allowed_tools
        }
        r.setex(
            name=f"context_tag:{agent_id}",
            time=self.expire_seconds,
            value=json.dumps(tag_data)
        )

    @staticmethod
    def get(agent_id: str) -> Dict | None:
        """从Redis获取指定agent_id的上下文标签"""
        tag_str = r.get(f"context_tag:{agent_id}")
        if tag_str:
            return json.loads(tag_str)
        return None

3.3 步骤2：定义带权限校验的工具

LangChain的工具是用Tool类定义的，我们可以写一个装饰器，用来给工具加权限校验逻辑。装饰器的作用是，当工具被调用的时候，先检查当前Agent的上下文标签，看看这个工具是不是在允许调用的列表里，如果不在，就返回“权限不足”的提示，不让工具执行。

先写权限校验的装饰器：

# 技术栈：Python 3.9+, LangChain 0.1.0+, FastAPI, Redis
from langchain.tools import Tool
from functools import wraps

def check_tool_permission(tool_name: str):
    """权限校验装饰器，检查当前Agent是否有权限调用该工具"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 从kwargs里获取agent_id（这个是我们调用工具的时候传的）
            agent_id = kwargs.get("agent_id")
            if not agent_id:
                return "错误：缺少Agent ID，无法校验权限"
            
            # 获取当前Agent的上下文标签
            tag = ContextTag.get(agent_id)
            if not tag:
                return "错误：上下文标签不存在或已过期"
            
            # 检查工具是否在允许列表里
            if tool_name not in tag["allowed_tools"]:
                return f"权限不足：您的Agent（{agent_id}）不允许调用工具{tool_name}，允许的工具列表：{tag['allowed_tools']}"
            
            # 权限校验通过，执行原函数
            return func(*args, **kwargs)
        return wrapper
    return decorator

然后我们定义几个测试用的工具，比如财务报表工具、发票查询工具、客户画像工具，给它们加上权限校验：

# 技术栈：Python 3.9+, LangChain 0.1.0+, FastAPI, Redis
# 定义财务报表工具
@check_tool_permission("财务报表工具")
def get_financial_report(agent_id: str, start_date: str, end_date: str) -> str:
    """获取指定时间段的财务报表"""
    # 模拟查询财务数据，实际项目中这里会调用真实的财务系统接口
    return f"财务报表（{start_date}至{end_date}）：营收100万，利润20万"

# 定义发票查询工具
@check_tool_permission("发票查询工具")
def query_invoice(agent_id: str, invoice_id: str) -> str:
    """查询指定ID的发票信息"""
    # 模拟查询发票数据
    return f"发票ID：{invoice_id}，金额：5000元，状态：已支付"

# 定义客户画像工具
@check_tool_permission("客户画像工具")
def get_customer_profile(agent_id: str, customer_id: str) -> str:
    """获取指定客户的画像信息"""
    # 模拟查询客户画像数据
    return f"客户ID：{customer_id}，等级：VIP，消费总额：10万"

# 把函数包装成LangChain的Tool
tools = [
    Tool(
        name="财务报表工具",
        func=get_financial_report,
        description="用于获取指定时间段的财务报表，参数需要包含agent_id、start_date（格式YYYY-MM-DD）、end_date（格式YYYY-MM-DD）"
    ),
    Tool(
        name="发票查询工具",
        func=query_invoice,
        description="用于查询指定ID的发票信息，参数需要包含agent_id、invoice_id"
    ),
    Tool(
        name="客户画像工具",
        func=get_customer_profile,
        description="用于获取指定客户的画像信息，参数需要包含agent_id、customer_id"
    )
]

3.4 步骤3：创建带上下文的LangChain Agent

接下来我们创建一个自定义的LangChain Agent，这个Agent会在调用工具的时候，自动把自己的agent_id传给工具，让工具能做权限校验。

# 技术栈：Python 3.9+, LangChain 0.1.0+, FastAPI, Redis
from langchain.agents import AgentExecutor, create_openai_tools_agent
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder

# 初始化大模型，这里用OpenAI的模型，实际项目中可以换成自己的大模型
llm = ChatOpenAI(model="gpt-3.5-turbo", temperature=0)

# 定义Agent的提示词模板，告诉Agent调用工具的时候要带上agent_id
prompt = ChatPromptTemplate.from_messages([
    ("system", "你是一个智能助手，调用工具时必须把当前Agent的ID（{agent_id}）作为参数传给工具。"),
    MessagesPlaceholder("chat_history", optional=True),
    ("human", "{input}"),
    MessagesPlaceholder("agent_scratchpad"),
])

# 创建Agent
agent = create_openai_tools_agent(llm, tools, prompt)

# 创建Agent执行器，用来执行Agent的任务
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

3.5 步骤4：用FastAPI做Agent的调用入口

最后我们用FastAPI做一个接口，用来接收用户的请求，给Agent分配agent_id和上下文标签，然后调用Agent完成任务。

# 技术栈：Python 3.9+, LangChain 0.1.0+, FastAPI, Redis
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel

app = FastAPI(title="多租户LangChain Agent服务")

# 定义请求体结构，用户调用接口时传的参数
class AgentRequest(BaseModel):
    tenant_id: str  # 租户ID，比如T1001
    scene: str  # 场景，比如财务、营销
    allowed_tools: List[str]  # 允许调用的工具列表
    input: str  # 用户的问题，比如“帮我查2024年1月的财务报表”

@app.post("/call-agent")
async def call_agent(request: AgentRequest):
    # 生成唯一的agent_id，这里用租户ID+场景+时间戳的方式，保证每个请求的agent_id唯一
    import time
    agent_id = f"{request.tenant_id}:{request.scene}:{int(time.time())}"
    
    # 保存上下文标签到Redis
    tag = ContextTag(
        tenant_id=request.tenant_id,
        scene=request.scene,
        allowed_tools=request.allowed_tools
    )
    tag.save(agent_id)
    
    try:
        # 调用Agent执行任务，把agent_id传给Agent
        result = agent_executor.invoke({
            "input": request.input,
            "agent_id": agent_id
        })
        return {"status": "success", "result": result["output"]}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Agent执行失败：{str(e)}")

四、应用场景分析

这种基于上下文的工具权限控制方法，适合很多SaaS场景，比如：

1. 企业级智能客服SaaS：不同企业的客服Agent只能调用自己企业的订单、物流工具，不能调用其他企业的；
2. 智能办公SaaS：不同部门的Agent只能调用自己部门的工具，比如财务部门的Agent不能调用研发部门的代码仓库工具；
3. 教育SaaS：不同学校的Agent只能调用自己学校的题库、成绩查询工具，不能调用其他学校的。

五、技术优缺点分析

5.1 优点

1. 成本低：所有租户的Agent共享一套基础代码，不用给每个租户单独部署，减少了服务器成本和运维成本；
2. 权限粒度细：可以做到租户级、场景级、工具级的权限控制，甚至可以做到同一租户下不同用户的Agent用不同的工具；
3. 扩展性强：如果要加新的工具，只需要定义新的Tool类，给它加上权限校验装饰器就行，不用改核心代码；
4. 安全性高：上下文标签存在Redis里，有过期时间，而且每个Agent的权限都是独立的，不会出现越权访问的问题。

5.2 缺点

1. 依赖Redis：如果Redis挂了，上下文标签就存不了，权限校验就做不了，所以需要做Redis的高可用；
2. 权限配置麻烦：如果租户很多、场景很多，手动配置每个Agent的允许工具列表会很麻烦，需要做一个可视化的权限配置后台；
3. 大模型的错误：如果大模型调用工具的时候，不小心传错了agent_id，还是可能出现权限问题，所以需要对大模型的输出做校验。

六、注意事项

1. 上下文标签的安全：上下文标签里包含了租户的权限信息，所以Redis的访问要做权限控制，不能让外部随便访问；
2. 过期时间的设置：上下文标签的过期时间不能太长，也不能太短，太长会有安全风险，太短会影响用户体验，一般设置1小时到24小时比较合适；
3. 工具参数的校验：除了校验工具的调用权限，还要校验工具的参数，比如财务报表工具的日期格式对不对，防止出现恶意请求；
4. 日志监控：要对Agent的调用日志做监控，记录每个Agent调用了什么工具、调用结果是什么，一旦出现权限问题，能快速定位。

七、文章总结

多租户环境下的LangChain Agent隔离性是很多SaaS项目都会碰到的问题，基于上下文的工具权限控制是一种低成本、高灵活性的解决方案。核心思路就是给每个Agent分配一个包含权限信息的上下文标签，在工具调用的时候做权限校验，只有符合要求的工具才能被调用。

通过上面的完整示例，我们可以看到，这种方法实现起来并不复杂，只需要在LangChain的工具调用流程中加一个权限校验的步骤就行。只要做好Redis的高可用、权限配置的可视化、日志监控等配套措施，就能满足大多数多租户场景的需求。