当前位置：首页 > news >正文

AI Agent安全架构设计：基于最小权限原则的三层防御体系实践

news 2026/8/4 23:29:46

1. 从“全知全能”到“最小权限”：为什么AI Agent需要安全锁？

最近在折腾各种AI Agent框架，从AutoGPT到LangChain，再到一些自研的RAG系统，我发现一个越来越明显的趋势：大家热衷于给Agent赋予越来越多的工具和能力，恨不得让它能调用一切API、读写所有文件、执行任意命令。这就像给一个刚拿到驾照的新手，直接配了一辆可以飙到时速300公里、还能一键启动所有功能的超级跑车。听起来很酷，但第一次上路，大概率会撞得稀里哗啦。

我亲身经历过几次“车祸现场”。有一次，一个负责整理周报的Agent，因为权限过大，在遍历文件目录时，不小心把.git文件夹里的对象文件也给“整理”了，直接导致一个开发中的项目版本库损坏。还有一次，一个联网搜索的Agent，在获取信息后，被设计成可以自动将“重要发现”写入一个共享的Notion数据库。结果因为提示词的一点歧义，它把一些未经严格核实的、带有明显倾向性的网络论坛内容也当成了“事实”写了进去，差点引发团队内部的信息污染。这些事让我彻底明白，一个能力不受控的AI Agent，其破坏力可能远超它的创造力。

这引出了我们今天要深入探讨的核心概念：最小权限原则。这不是什么新词，在传统软件开发、尤其是安全领域，它是一条黄金法则。其核心思想是：一个程序、一个进程、或者一个用户，只应该拥有完成其特定任务所必需的最低限度的权限，不多也不少。把这个原则应用到AI Agent上，就是给这个“数字员工”上一把精准的“安全锁”。我们不再说“你可以访问整个服务器”，而是说“你只能读写/data/reports/目录下以.md结尾的文件”。我们不再说“你可以调用所有API”，而是说“你只能使用‘获取天气’和‘发送邮件’这两个接口”。

为什么这如此重要？除了防止上述的“意外破坏”，最小权限设计还能带来三个关键好处：

限制攻击面：即使Agent的提示词被恶意注入（Prompt Injection），或者其依赖的模型、工具链存在漏洞，由于其权限被严格限制，攻击者能做的事情也非常有限。他无法让Agent删除系统关键文件，也无法让它调用转账接口。
提升可预测性与可调试性：当一个Agent行为异常时，如果它的权限是模糊的、宽泛的，排查问题就像大海捞针。而如果权限是精确定义的，我们就可以快速排除大量不可能的操作路径，将问题定位在有限的几个授权动作内，大大降低了调试复杂度。
符合合规与审计要求：在许多涉及敏感数据（如用户隐私、商业数据）的场景下，数据访问必须有严格的、可审计的路径。为AI Agent配置最小权限，并记录其每一次权限使用，是满足这类合规性要求的技术基础。

所以，给AI Agent设计“安全锁”，绝不是限制其能力，而是为了让它能在安全、可控的边界内，更可靠、更持久地发挥价值。这就像给超级跑车装上完备的电子稳定系统、牵引力控制和防滚架——不是为了让它跑得慢，而是为了让顶尖的性能能够安全地释放。

2. 解构“OpenClaw 龙虾”：一个最小权限的具象化设计隐喻

“OpenClaw 龙虾”这个标题非常精妙，它不是一个具体的工具或框架，而是一个极具启发性的设计隐喻。我们来拆解一下这个隐喻，它几乎完美诠释了最小权限设计的精髓。

首先，是“龙虾”。龙虾有什么特点？它拥有一对非常强大的螯（爪子），这是它的核心工具，用于捕食、防御和挖掘。但同时，这对螯的力量是定向的、功能特定的。一只螯通常较大，用于粉碎；另一只较小，用于切割。龙虾不会用它的螯去梳理自己的触须，也不会用螯去完成精细的卵化工作——它有其他更合适的附肢来完成这些任务。这对应了AI Agent的工具集。一个设计良好的Agent，不应该拥有一把“瑞士军刀”式的万能工具，而应该像龙虾一样，拥有一系列功能明确、各司其职的专用工具。一个负责数据分析的Agent，它的“螯”可能是Pandas DataFrame操作和SQL查询；一个负责客服的Agent，它的“螯”可能是知识库检索和情感分析API。

其次，是“Claw”（爪子/螯）。关键在于，龙虾的螯是“开放”的，但又是“受控”的。“开放”意味着这些工具的能力是强大的、可被调用的接口。“受控”则体现在两个方面：一是生理结构的天然限制，螯只能做特定类型的动作（夹、砸、剪），不能像手指一样灵活抓握所有物体；二是神经系统的调控，龙虾的大脑决定在什么时候、对什么目标、使用多大的力量去挥动它的螯。映射到AI Agent，这就是权限控制系统。它包含两个层面：

静态权限模型：在Agent初始化时，就通过配置声明它“可以做什么”。例如，在代码层面，只给它实例化FileReaderTool（权限：只读./input/目录）和WebSearchTool（权限：仅限GET请求，过滤成人内容），而不提供FileWriterTool或ShellCommandTool。这是“生理结构”的限制。
动态权限决策：在Agent运行过程中，根据当前任务上下文、用户指令和历史行为，动态决定是否批准某项工具调用。例如，即使用户说“把结果保存一下”，拥有FileWriterTool的Agent在尝试写入/etc/passwd系统文件时，权限中间件也应该拦截此次调用。这是“神经系统”的调控。

最后，是“Open”。这里的“开放”不是指开源（虽然开源实现很重要），而是指权限策略的透明化、可配置化和可观测性。一个好的最小权限系统，其策略规则应该是像配置文件一样清晰可见、易于修改的（YAML, JSON）。管理员能一目了然地看到：“Agent_A 可以在tenant_1的bucket_storage中上传文件，但单个文件不得大于10MB，且仅限.pdf, .txt格式”。同时，所有的权限检查、通过或拒绝的决策，都应该有清晰的日志记录，形成一个完整的审计追踪链条。

所以，“OpenClaw 龙虾”这个设计思想，就是倡导我们构建这样的AI Agent：它像龙虾一样，装备着为特定任务而生的、强大的专用工具（Claw）；这些工具的使用被一套明确的、静态与动态结合的权限规则所约束（最小权限）；而整个权限体系的策略和状态是透明、可配置、可观测的（Open）。接下来，我们就将这套隐喻，落地为实战中的具体设计模式与技术选型。

3. 实战架构：构建Agent权限控制的三层防御体系

纸上谈兵终觉浅，我们来搭建一个实实在在的、为AI Agent设计的最小权限实战架构。我将其称为“三层防御体系”，它从外到内，从粗到细，逐层收紧对Agent行为的管控。这个架构不依赖于某个特定框架，而是一种可以融入大多数Agent系统（如LangChain, LlamaIndex, AutoGen）的设计模式。

3.1 第一层：工具层的静态权限声明

这是最基础，也是最关键的一层。核心思想是：在工具（Tool）被创建和提供给Agent的那一刻，就将其权限“烙”在工具本身，而不是依赖Agent的“自觉”。

如何实现？我们不再创建通用的、无所不能的工具类。相反，我们创建的是经过“阉割”或“限定”的特化工具实例。举个例子，我们有一个通用的FileSystemTool。

错误做法（全权限工具）：

# 一个什么都能做的文件工具，极度危险 class DangerousFileTool(BaseTool): name = “file_tool” description = “Read, write, delete files. Use with extreme caution.” def _run(self, action: str, path: str, content: str = None): if action == “read”: with open(path, ‘r’) as f: return f.read() elif action == “write”: with open(path, ‘w’) as f: f.write(content) elif action == “delete”: os.remove(path) # ... 其他危险操作 # 然后把这个工具直接给Agent agent.tools.append(DangerousFileTool())

正确做法（最小权限工具实例）：

# 首先，定义一个基础的文件操作工具，但内部逻辑包含权限检查点（第二层会用到） class BaseFileTool(BaseTool): name = “base_file_tool” # ... 基础实现 # 然后，通过工厂函数或配置，创建权限受限的具体工具实例 def create_restricted_file_tool(allowed_base_dir: str, allowed_operations: list, allowed_extensions: list): class RestrictedFileTool(BaseFileTool): name = “restricted_file_operator” description = f”Operate on files under {allowed_base_dir}. Allowed ops: {‘, ‘.join(allowed_operations)}. Allowed ext: {‘, ‘.join(allowed_extensions)}.” def _validate_and_run(self, action: str, relative_path: str, content: str = None): # 1. 路径规范化与越界检查（静态声明的一部分） full_path = os.path.join(allowed_base_dir, relative_path) if not os.path.commonpath([allowed_base_dir, full_path]) == allowed_base_dir: raise PermissionError(f”Access denied: Path {relative_path} is outside allowed directory.”) if allowed_extensions and not any(full_path.endswith(ext) for ext in allowed_extensions): raise PermissionError(f”Access denied: File extension not in {allowed_extensions}.”) if action not in allowed_operations: raise PermissionError(f”Operation {action} not permitted.”) # 2. 调用父类实际执行操作（父类可能包含更细的动态检查） return super()._run(action, full_path, content) def _run(self, *args, **kwargs): # 代理到验证方法 return self._validate_and_run(*args, **kwargs) return RestrictedFileTool() # 为“周报生成Agent”创建一个只能读取特定目录下.md文件的工具 weekly_report_agent_tool = create_restricted_file_tool( allowed_base_dir=“/data/weekly_input”, allowed_operations=[“read”], allowed_extensions=[“.md”, “.txt”] ) # 为“报告归档Agent”创建一个只能写入特定目录的.zip文件的工具 archive_agent_tool = create_restricted_file_tool( allowed_base_dir=“/data/archive”, allowed_operations=[“write”], allowed_extensions=[“.zip”] )

通过这种方式，每个Agent拿到的工具，从诞生起就戴上了“镣铐”。这是最有效、最根本的权限控制。

3.2 第二层：执行层的动态策略引擎

静态声明很好，但世界是动态的。用户可能临时要求Agent处理一个特殊任务，或者我们需要根据运行时状态（如当前时间、系统负载、敏感词触发）来调整权限。这就需要第二层：一个在Agent每次调用工具前进行拦截和裁决的动态策略引擎。

这个引擎通常以一个中间件的形式存在，插在Agent的决策逻辑和工具的执行逻辑之间。它的工作流程如下：

拦截：当Agent决定调用某个工具（Tool）并传入参数时，调用请求首先被策略引擎截获。
上下文收集：引擎收集当前执行的完整上下文，包括：调用的工具对象、传入的参数、当前的会话历史、用户身份、环境变量等。
策略评估：引擎将上下文信息送入一系列预定义的策略规则中进行评估。这些规则可以用简单的代码逻辑、类自然语言的策略语言（如Open Policy Agent的Rego）或专门的规则引擎来定义。
裁决：根据所有策略规则的评估结果，引擎做出裁决：允许、拒绝，或者需要附加条件（如需要二次确认、需要记录到特殊审计日志）。
执行或阻断：如果允许，则放行请求，工具正常执行；如果拒绝，则向Agent返回一个友好的错误信息（如“您没有权限执行此操作”），并记录审计日志。

一个简化的Python中间件示例：

class PolicyEnforcementMiddleware: def __init__(self, policy_rules): self.policy_rules = policy_rules # 一组策略函数 def before_tool_execute(self, agent_id, tool_name, tool_args, session_context): “”“在工具执行前调用”“” decision = “ALLOW” audit_log = {“agent_id”: agent_id, “tool”: tool_name, “args”: tool_args, “timestamp”: time.time()} for rule in self.policy_rules: result, reason = rule.evaluate(agent_id, tool_name, tool_args, session_context) if result == “DENY”: decision = “DENY” audit_log[“deny_reason”] = reason break # 一票否决 elif result == “CONDITIONAL”: # 处理附加条件，例如发送确认请求 if not self._request_human_approval(agent_id, tool_name, tool_args): decision = “DENY” audit_log[“deny_reason”] = “Human approval denied” break audit_log[“decision”] = decision self._write_audit_log(audit_log) if decision == “DENY”: raise PermissionError(f”Tool execution denied by policy. Reason: {audit_log.get(‘deny_reason’, ‘unknown’)}”) # 如果允许，则什么都不做，流程继续 # 示例策略规则：禁止在非工作时间写入数据库 def rule_no_db_write_after_hours(agent_id, tool_name, tool_args, context): if tool_name == “DatabaseWriteTool”: current_hour = datetime.now().hour if current_hour < 9 or current_hour > 18: return “DENY”, “Database writes are only allowed during business hours (9AM-6PM).” return “ALLOW”, None # 将中间件挂载到Agent的执行循环中 agent.execution_middleware.append(PolicyEnforcementMiddleware([rule_no_db_write_after_hours]))

这一层提供了极大的灵活性，可以实现诸如“单次操作额度限制”、“敏感操作二次确认”、“基于内容的风险拦截”等复杂控制。

3.3 第三层：环境层的沙箱与隔离

前两层主要控制“做什么”，第三层则控制“在哪里做”。即使Agent的权限被严格控制，我们仍然不希望它的任何操作直接影响真实的生产环境。因此，我们需要一个安全的沙箱环境。

文件系统隔离：使用容器（Docker）或虚拟机的只读卷、覆盖文件系统（OverlayFS），为Agent提供一个独立的、临时的文件空间。Agent以为自己是在修改/data/report.docx，实际上它操作的是沙箱内的一个副本。任务结束后，沙箱销毁，所有临时修改一并消失，只有通过特定“出口”（如经过审核的API）提交的结果会被持久化。
网络隔离：限制Agent容器的网络访问。只允许其访问白名单内的内部API端点或外部服务（如特定的天气API、搜索引擎API），阻断其对内部管理网络、数据库直连地址或其他敏感服务的访问。这可以通过容器网络的防火墙规则或服务网格（如Istio）的Sidecar代理来实现。
资源限制：通过Cgroups限制Agent进程的CPU、内存使用量，甚至运行时间。防止一个陷入死循环或提示词注入的Agent拖垮整个宿主系统。
敏感信息脱敏：在将环境变量、配置文件或用户输入传递给Agent之前，使用占位符或哈希值替换掉真实的API密钥、数据库连接字符串等敏感信息。Agent拿到的只是一个“令牌”，真正的凭证由沙箱外层的安全模块在需要时代理注入。

这三层防御体系是相辅相成的。工具层定义了能力的边界，策略层定义了行为的规则，环境层定义了活动的舞台。三者结合，才能为AI Agent打造一个既强大又安全的“数字牢笼”。

4. 核心挑战与避坑指南：权限设计中的“灰犀牛”

在实际项目中实施最小权限设计，你会遇到一些意料之外却又在情理之中的挑战。我把它们称为“灰犀牛”——那些明显、高概率，却常常被忽视的风险。下面是我踩过坑后总结出的核心挑战与应对策略。

4.1 挑战一：权限粒度的“过细”与“过粗”悖论

这是最常见的两难困境。权限设得太细（例如，每个文件、每个API端点都单独授权），管理复杂度会爆炸式增长，策略配置变得极其繁琐，且容易出错。权限设得太粗（例如，允许访问整个/data目录），又失去了最小权限的意义，一个路径遍历漏洞就可能造成大面积数据泄露。

避坑策略：寻找“恰如其分”的抽象层级不要纠结于对每一个具体资源授权。而是基于Agent的角色和任务类型来定义权限包。

角色维度：定义如“数据分析师”、“内容审核员”、“客服代表”等角色模板。
任务维度：定义如“读取用户反馈”、“生成季度图表”、“归档历史日志”等任务模板。
资源抽象：将资源分组。例如，不是授权给/data/feedback/2024/03/*.csv，而是定义一个名为“FEEDBACK_CSV_READONLY”的资源标签，它映射到一组具体的文件路径模式。当新的反馈文件存入2024/04目录时，只需更新资源标签的定义，所有关联该标签的权限自动生效。

这样，权限配置就从“资源->Agent”的直接映射，变成了“角色/任务 -> 资源标签 -> 具体资源”的间接管理，在安全性和可管理性之间取得平衡。

4.2 挑战二：工具组合与权限的“特权升级”

单个工具是安全的，但Agent可能会通过组合多个低权限工具，实现一个高权限操作。例如，Agent没有被授权直接“删除数据库记录”，但它可能被授权“执行SQL查询”和“写入日志文件”。一个恶意的提示词可能诱导它先执行“SELECT * FROM users;”将数据查询出来，再通过写入日志工具将数据输出到某个可访问的文件中，从而实现数据泄露。这就是一种“权限拼接”攻击。

避坑策略：实施上下文感知的关联策略动态策略引擎（第二层）必须能够分析工具调用的序列和上下文，而不仅仅是单个调用。

会话级风险评分：为每个Agent会话维护一个风险分数。当检测到可疑的工具调用序列（如短时间内连续调用数据读取和网络发送工具）时，提高风险分数。当分数超过阈值时，触发更严格的审查（如要求人工确认）或直接终止会话。
数据流跟踪：对敏感数据（如数据库查询结果中的身份证号、手机号）进行标记。当这些被标记的数据试图通过未被授权输出敏感信息的工具（如一个只允许输出摘要的日志工具）流出时，策略引擎应能识别并拦截。这需要与数据脱敏、水印技术结合。

4.3 挑战三：异常处理与权限边界的模糊地带

当权限检查失败时，如何反馈给Agent和用户？直接抛出一个冰冷的“Permission Denied”错误，可能会让Agent陷入困惑，甚至在其思维链中引发连锁错误。但反馈太多信息（如“你不能访问/etc/shadow，但你可以访问/home/user/docs”），又可能向潜在的攻击者泄露系统内部结构，帮助其进行侦察。

避坑策略：设计分层的、友好的拒绝反馈

对Agent：返回结构化、可操作的错误信息。例如，{“error”: “permission_denied”, “suggestion”: “You do not have write access to that location. Consider using the ‘save_to_temp’ tool instead.”}。这样，Agent的LLM核心可以理解错误类型，并根据建议调整后续行动。
对最终用户（如果是人机交互）：展示友好的、业务层面的提示。例如，“系统无法将文件保存到指定位置，可能是路径不存在或权限不足。请尝试另一个位置或联系管理员。” 避免暴露内部路径、工具名等细节。
对管理员/审计员：在审计日志中记录完整的详细信息，包括被拒绝的请求、触发的策略规则、会话ID、用户身份等，以便进行安全分析和事件溯源。

4.4 挑战四：权限策略的持续维护与测试

权限策略不是一次性配置。随着业务变化、新工具引入、新威胁出现，策略需要不断更新。如何保证新策略不会阻断正常的业务流程？如何测试权限系统的有效性？

避坑策略：建立策略的CI/CD管道像对待应用程序代码一样对待安全策略。

版本控制：将所有权限策略文件（YAML/JSON/Rego）纳入Git管理。
单元测试：为每一条策略规则编写测试用例，验证其在各种预设场景下（正常请求、越权请求、边界情况）能否做出正确裁决。
集成测试：在测试环境中，用真实的Agent工作流（包括各种边缘用例和历史上出现过的故障场景）进行自动化测试，确保策略更新后，核心业务功能依然畅通，而安全漏洞被有效堵住。
金丝雀发布：将新的策略先应用到一小部分低风险、非核心的Agent或环境中，观察一段时间，确认无误后再全量推广。

5. 从设计到实现：一个基于LangChain的“龙虾”Agent构建示例

理论说了这么多，我们用一个具体的、简化的例子，把“OpenClaw 龙虾”的设计理念串起来。我们将使用流行的LangChain框架，构建一个具有最小权限的“周报数据提取Agent”。

场景：我们需要一个Agent，它能从指定的共享目录（只读）读取每周的销售数据CSV文件，计算一些基本统计量（如总额、平均值），然后将摘要写入另一个专门的“报告输出”目录（只写），最后通过邮件工具（仅能发送到内部邮件组）发送通知。它不能读取其他无关文件，不能向其他目录写入，也不能随意发邮件。

5.1 第一步：定义权限边界与创建受限工具

首先，我们创建三个遵循最小权限原则的工具实例。

import os import pandas as pd from langchain.tools import BaseTool from typing import Type from pydantic import BaseModel, Field import smtplib from email.mime.text import MIMEText # —————— 1. 只读CSV文件工具 —————— class ReadCSVInput(BaseModel): “”“工具输入模型，用于参数验证。”“” filename: str = Field(description=”The name of the CSV file (e.g., ‘sales_week_15.csv’) within the allowed directory.”) class RestrictedCSVReaderTool(BaseTool): name = “read_sales_csv” description = “Read a sales data CSV file from the weekly input directory. Only files under ‘/data/weekly_sales_input/’ with .csv extension can be read.” args_schema: Type[BaseModel] = ReadCSVInput allowed_base_dir: str = “/data/weekly_sales_input” # 静态权限声明 def _run(self, filename: str) -> str: “”“核心执行逻辑，包含静态权限检查。”“” # 构造完整路径 full_path = os.path.join(self.allowed_base_dir, filename) # 1. 路径越界检查 if not os.path.commonpath([self.allowed_base_dir, os.path.abspath(full_path)]) == self.allowed_base_dir: raise PermissionError(f”Security violation: Attempted to access path outside of {self.allowed_base_dir}”) # 2. 文件类型检查 if not filename.endswith(‘.csv’): raise PermissionError(“Only .csv files are allowed.”) # 3. 文件存在性检查（业务逻辑） if not os.path.exists(full_path): return f”Error: File {filename} does not exist in the input directory.” # 4. 执行读取 try: df = pd.read_csv(full_path) # 返回一个简化的描述，避免在提示词中传递过大数据 return f”Successfully read {filename}. Shape: {df.shape}. Columns: {list(df.columns)}. Head (3 rows):\n{df.head(3).to_string()}” except Exception as e: return f”Error reading file: {str(e)}” def _arun(self, filename: str): raise NotImplementedError(“Async not supported”) # —————— 2. 只写摘要文件工具 —————— class WriteSummaryInput(BaseModel): summary_content: str = Field(description=”The text content of the weekly summary to be saved.”) class RestrictedSummaryWriterTool(BaseTool): name = “write_weekly_summary” description = “Write the weekly summary text to a file in the output directory. Files will be named automatically with a timestamp.” args_schema: Type[BaseModel] = WriteSummaryInput allowed_base_dir: str = “/data/weekly_summaries_output” # 静态权限声明 def _run(self, summary_content: str) -> str: full_path = os.path.join(self.allowed_base_dir, f”summary_{pd.Timestamp.now().strftime(‘%Y%m%d_%H%M%S’)}.txt”) # 路径越界检查（虽然通过构造函数基本保证，但双重验证更安全） if not os.path.commonpath([self.allowed_base_dir, os.path.abspath(full_path)]) == self.allowed_base_dir: raise PermissionError(f”Security violation: Attempted to write outside of {self.allowed_base_dir}”) try: os.makedirs(self.allowed_base_dir, exist_ok=True) with open(full_path, ‘w’, encoding=‘utf-8’) as f: f.write(summary_content) return f”Summary successfully written to: {os.path.basename(full_path)}” except Exception as e: return f”Error writing summary: {str(e)}” def _arun(self, summary_content: str): raise NotImplementedError(“Async not supported”) # —————— 3. 受限邮件发送工具 —————— class SendNotificationInput(BaseModel): subject: str = Field(description=”Subject of the notification email.”) body: str = Field(description=”Body content of the email.”) class RestrictedEmailTool(BaseTool): name = “send_internal_notification” description = “Send an email notification to the predefined internal reports mailing list. Cannot send to arbitrary addresses.” args_schema: Type[BaseModel] = SendNotificationInput allowed_recipient: str = “reports@internal.company.com” # 静态权限声明 def _run(self, subject: str, body: str) -> str: # 注意：真实环境中，邮箱密码等应来自环境变量或安全密钥库，不应硬编码。 sender = “agent_system@company.com” # 强制使用预设的收件人，忽略任何可能通过提示词注入的收件人参数 recipients = [self.allowed_recipient] msg = MIMEText(body) msg[‘Subject’] = subject msg[‘From’] = sender msg[‘To’] = “, “.join(recipients) try: # 简化示例，真实情况使用SMTP库配置 # with smtplib.SMTP(‘smtp.internal.company.com’, 587) as server: # server.login(sender, os.getenv(‘SMTP_PASSWORD’)) # server.sendmail(sender, recipients, msg.as_string()) print(f”[SIMULATED EMAIL] To: {recipients}, Subject: {subject}, Body: {body[:50]}…”) return f”Notification email sent successfully to internal reports list.” except Exception as e: return f”Failed to send email: {str(e)}” def _arun(self, subject: str, body: str): raise NotImplementedError(“Async not supported”)

5.2 第二步：集成动态策略中间件

我们为这个Agent增加一个简单的动态策略：禁止在非工作时间（假设为UTC时间22:00至06:00）写入摘要文件，以防自动化任务在系统维护时段产生意外干扰。

from datetime import datetime, time from langchain.agents import AgentExecutor from langchain.agents import Tool class TimeBasedPolicyMiddleware: “”“一个简单的时间策略中间件示例。”“” def __init__(self, start_restricted_hour=22, end_restricted_hour=6): self.restricted_start = start_restricted_hour self.restricted_end = end_restricted_hour def check_tool_call(self, tool_name: str, tool_args: dict) -> (bool, str): “”“检查工具调用是否被允许。返回 (是否允许, 原因)。”“” current_hour = datetime.utcnow().hour is_restricted_hours = self.restricted_start <= current_hour or current_hour < self.restricted_end if tool_name == “write_weekly_summary” and is_restricted_hours: return False, f”Tool ‘{tool_name}’ is not allowed during restricted hours (UTC {self.restricted_start}:00-{self.restricted_end}:00).” return True, “” # 包装工具，在调用前插入策略检查 def create_tool_with_policy(tool_instance, policy_middleware): original_run = tool_instance._run def wrapped_run(*args, **kwargs): # 这里简化了参数传递，实际需根据工具args_schema解析 is_allowed, reason = policy_middleware.check_tool_call(tool_instance.name, kwargs) if not is_allowed: return f”Action blocked by policy: {reason}” return original_run(*args, **kwargs) tool_instance._run = wrapped_run return tool_instance # 创建策略中间件实例 time_policy = TimeBasedPolicyMiddleware(start_restricted_hour=22, end_restricted_hour=6) # 创建工具实例并应用策略包装 csv_reader = RestrictedCSVReaderTool() summary_writer = create_tool_with_policy(RestrictedSummaryWriterTool(), time_policy) email_sender = RestrictedEmailTool() tools = [csv_reader, summary_writer, email_sender] # 注意：这里只是概念演示。在完整LangChain Agent执行器中，需要更精细地集成中间件，通常通过自定义Agent或Executor类实现。

5.3 第三步：配置Agent与提示词工程

最后，我们配置Agent，并通过提示词进一步约束其行为范围。

from langchain.agents import initialize_agent, AgentType from langchain.chat_models import ChatOpenAI # 示例，可用其他LLM llm = ChatOpenAI(temperature=0, model=“gpt-4”) # 使用低temperature以获得更确定性的输出 # 构建系统提示词，明确Agent的角色和权限边界 system_message = “”” You are a Weekly Sales Data Assistant. Your task is to help generate a weekly sales summary. You have access to the following tools, each with strict limitations: 1. ‘read_sales_csv’: Can ONLY read CSV files from the ‘/data/weekly_sales_input/’ directory. You must provide a filename like ‘sales_week_15.csv’. 2. ‘write_weekly_summary’: Can ONLY write text summary files to the ‘/data/weekly_summaries_output/’ directory. It will generate the filename automatically. 3. ‘send_internal_notification’: Can ONLY send emails to the internal reports mailing list (reports@internal.company.com). You need to provide a subject and body. Your workflow should be: 1. Use ‘read_sales_csv’ to get the data for the requested week. 2. Analyze the data (calculate total sales, average per day, etc.) based on the columns you see. 3. Use ‘write_weekly_summary’ to save your analysis as a text summary. 4. Use ‘send_internal_notification’ to notify that the summary is ready. IMPORTANT RULES: - You MUST NOT attempt to read or write files outside the specified directories. - You MUST NOT try to send emails to any address other than the internal list. - If a tool returns a permission error or says ‘blocked by policy’, stop and report the issue. Do not try to work around it. - If you are unsure about a filename, ask the user for clarification. “”” # 初始化Agent（这里使用ZERO_SHOT_REACT_DESCRIPTION，简单示例） agent = initialize_agent( tools, llm, agent=AgentType.ZERO_SHOT_REACT_DESCRIPTION, verbose=True, agent_kwargs={ “prefix”: system_message # 将系统提示词作为前缀注入 } ) # 运行Agent try: result = agent.run(“Read the sales data for week 15, generate a summary, save it, and send a notification with the subject ‘Weekly Sales Summary W15 Ready’.”) print(result) except Exception as e: print(f”Agent execution failed: {e}”)

通过这个例子，你可以看到“OpenClaw 龙虾”理念是如何落地的：