什么是 ‘Negative Prompts’？在 LangChain 中如何显式告诉 Agent “不要做什么”以减少误操作

大模型时代下的指令精度与挑战

随着大型语言模型（LLM）技术的飞速发展，我们正步入一个由人工智能驱动的全新时代。LLM强大的文本生成、理解和推理能力，使得构建智能助手、自动化工作流程、甚至是自主决策系统成为可能。在LangChain这样的框架中，通过将LLM与各种工具（Tools）、内存（Memory）和规划（Planning）机制相结合，我们能够创建出功能强大、能够执行复杂任务的Agent。这些Agent能够理解自然语言指令，自主选择并调用工具，从而完成从数据查询到代码生成等一系列操作。

然而，大模型的强大能力也伴随着一个核心挑战：控制。LLM本质上是概率模型，它们的“创造性”和“泛化能力”在带来惊喜的同时，也可能导致它们偏离预期，产生不准确、不安全、不相关甚至是有害的输出或行为。例如，一个旨在提供金融建议的Agent可能会在用户不经意间询问健康问题时，尝试给出医疗建议；一个负责处理敏感数据的Agent可能会在某种特定输入下，无意中泄露信息；或者一个代码生成Agent可能会生成不符合安全规范的代码。

为了确保Agent的行为符合我们的期望，并最大限度地减少误操作，我们不仅需要明确地告诉Agent“应该做什么”（即Positive Prompts），更需要清晰地界定“不应该做什么”（即Negative Prompts）。这不仅仅是提升用户体验的问题，更是构建负责任、安全、可信赖AI系统的关键。本文将深入探讨“Negative Prompts”的理念，并在LangChain Agent的语境下，详细阐述如何通过多种策略和代码实践，显式地向Agent传递“不要做什么”的指令，以提升其执行的精度和安全性。

什么是 ‘Negative Prompts’？

在LLM的交互语境中，’Negative Prompts’（负面提示或负向指令）指的是明确告知模型避免生成或执行特定内容、风格、行为或模式的指令。它们是与传统的“Positive Prompts”（正面提示）相对的概念，后者关注于指导模型生成什么。

举个简单的例子：

• Positive Prompt: "请写一首关于春天的诗歌，风格活泼。" (Focus on what to do)
• Negative Prompt: "在诗歌中不要提及任何悲伤的词语，不要使用深色意象。" (Focus on what not to do)

在图像生成领域，Negative Prompts的应用尤为普遍和直观。用户可以通过负面提示词，如“ugly, deformed, bad anatomy, disfigured”，来指导模型避免生成低质量、扭曲或解剖结构不正确的图像。

对于文本生成和Agent行为控制而言，Negative Prompts的重要性体现在以下几个方面：

1. 减少幻觉（Hallucinations）和不准确性：通过明确禁止模型“编造事实”、“提供未经证实的信息”，可以有效提升输出的真实性。
2. 确保安全性和合规性：禁止模型生成有害内容（如仇恨言论、非法建议）、泄露敏感信息或执行未经授权的操作，是构建安全AI系统的基石。
3. 遵循特定限制和规范：在企业级应用中，Agent往往需要在严格的业务规则和流程下运作。Negative Prompts可以确保Agent不越界，不违反既定协议。
4. 提高指令遵循度：有时，正面指令可能不足以完全覆盖所有潜在的误解。负面指令能够从另一个维度锁定模型的行为空间，使其更精确地遵循意图。
5. 优化用户体验：避免生成冗余、无关或令人反感的内容，从而提供更流畅、更专业的交互体验。

尽管Negative Prompts强大，但它们并非万能。模型对负面指令的遵循程度受多种因素影响，包括模型的规模、训练数据、指令的清晰度以及与正面指令的潜在冲突。因此，在实践中，通常需要结合多种策略，并进行迭代测试和优化。

LangChain Agent 架构回顾

在深入探讨如何实现Negative Prompts之前，我们先简要回顾LangChain Agent的核心架构。理解Agent的内部工作机制，有助于我们找到合适的切入点来注入和强制负面约束。

一个典型的LangChain Agent通常包含以下关键组件：

1. LLM (Large Language Model)：Agent的“大脑”，负责理解用户输入、进行推理、生成行动计划（Thought）以及最终的响应。
2. Tools (工具)：Agent的“手臂”，是Agent可以调用的外部功能或服务，例如搜索引擎、数据库查询、API调用、代码解释器等。每个工具都有一个描述，Agent会根据其描述和当前任务来决定是否以及如何使用它。
3. Agent Executor (执行器)：Agent的核心逻辑循环。它接收用户输入，将其传递给LLM。LLM生成一个“思想”和“行动”（Thought/Action），执行器根据“行动”调用相应的工具。工具返回“观察结果”（Observation），执行器再将这些结果反馈给LLM，形成一个循环，直到LLM认为任务完成并生成最终响应。
4. Prompt Template (提示模板)：定义了传递给LLM的指令结构。它通常包含Agent的系统级指令、可用的工具列表及其描述、历史对话信息（Memory），以及当前的用户输入。
5. Memory (内存)：用于存储Agent与用户之间的历史对话，以便Agent在后续交互中保持上下文。

Agent执行流程简化图：

用户输入 ->
    Prompt Template (注入系统指令、工具描述、历史) ->
        LLM (生成 Thought, Action) ->
            Agent Executor
                |
                V
            是否是最终响应？
            /        
           是         否
          /            
         V              V
      返回响应         调用工具 (Tool)
                         |
                         V
                    工具执行 (Observation)
                         |
                         V
                    LLM (循环)

从这个流程中我们可以看到，有几个关键点可以作为注入Negative Prompts的策略：

• Prompt Template: 直接在Agent的系统指令中添加负面约束。
• Tools: 通过限制工具的可用性或修改工具的行为来施加约束。
• Agent Executor 内部: 在Agent的思考和行动过程中进行监控和干预。
• 输出环节: 在Agent生成最终响应后进行校验。

接下来，我们将围绕这些切入点，详细探讨在LangChain中实现Negative Prompts的多种策略。

在 LangChain 中实现 ‘Negative Prompts’ 的策略与方法

本节将详细介绍在LangChain中显式告诉Agent“不要做什么”的六种主要策略，并提供具体的代码示例。

策略一：通过系统级 Prompt 指令进行约束

这是最直接也最常用的方法，即在Agent的初始提示（System Prompt）中明确地列出负面指令。这些指令将作为Agent在整个交互过程中的基本行为准则。

实现方式：
在构建Agent的PromptTemplate时，在系统消息中添加明确的负面约束。LangChain的Agent通常会使用一个包含system_message或类似概念的模板。

适用场景：

• 通用的行为准则，如“不要提及政治话题”、“不要提供医疗建议”。
• 对生成内容格式的简单限制，如“不要使用Markdown标题”。
• 禁止使用特定词汇或短语。

优点：

• 实现简单，直接有效。
• 适用于广泛的通用限制。

缺点：

• LLM可能会“遗忘”或忽略这些指令，尤其是在指令很长、冲突或复杂的场景下。
• 对于复杂的行为模式或安全关键型应用，单一的Prompt指令往往不够健壮。
• 容易被“Prompt Injection”攻击绕过。

代码示例：通用行为准则限制

我们将创建一个简单的Agent，并指示它不要提供医疗或法律建议。

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate

# 设置OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 定义一个简单的工具
@tool
def search_web(query: str) -> str:
    """Searches the web for information."""
    if "天气" in query:
        return f"查询到 '{query}' 的结果：今天天气晴朗，气温25度。"
    elif "股票" in query:
        return f"查询到 '{query}' 的结果：某公司股票上涨2%，建议谨慎投资。"
    elif "健康" in query or "疾病" in query or "法律" in query:
        return "很抱歉，我无法执行包含医疗或法律敏感信息的查询。"
    return f"查询到 '{query}' 的通用信息。"

tools = [search_web]

# 2. 定义包含负面指令的Prompt Template
# 注意：这里使用了create_react_agent内部的默认prompt结构，
# 我们通过system_message_template来注入自定义的系统指令。
# 为了更精细控制，我们可以构造一个Custom PromptTemplate。
# 但为了演示，我们先用这种方式。

# 假设我们需要一个Custom PromptTemplate来完全控制
# 我们可以基于LangChain的默认ReAct提示进行修改
# 这是一个简化的ReAct风格的提示，用于演示如何注入系统指令

# 原始 ReAct 提示通常是这样的，包含指令、工具、历史等
# 我们需要找到一个方式来注入我们的负面系统指令
# 对于 create_react_agent, 我们可以通过修改其底层使用的prompt
# 或者直接在prompt template中定义system message。

# 这里我们直接构造一个PromptTemplate，并将其传递给create_react_agent
# 假设create_react_agent可以接受一个包含system_message的PromptTemplate
# 实际上，create_react_agent 默认使用 `hub.pull("hwchase17/react")` 的 prompt，
# 但我们可以通过修改它来注入系统指令。

# 更直接的LangChain Agent Prompt 结构通常如下：
# from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder
# prompt = ChatPromptTemplate.from_messages(
#     [
#         ("system", "你是一个乐于助人的AI助手。请严格遵守以下规则：n1. 绝不能提供任何医疗建议。n2. 绝不能提供任何法律建议。n3. 不要讨论政治或宗教话题。n4. 如果用户请求医疗或法律内容，请明确拒绝并说明原因。"),
#         MessagesPlaceholder("chat_history"),
#         ("human", "{input}"),
#         MessagesPlaceholder("agent_scratchpad"),
#     ]
# )

# 为了演示 create_react_agent，我们通常需要一个 `agent_scratchpad`。
# 我们可以通过 `PromptTemplate.from_template` 来构造一个，但更推荐 `ChatPromptTemplate`。

# 重新构建一个适合 create_react_agent 的 PromptTemplate
# 这是 LangChain 默认 ReAct 提示的一个简化版本，用于演示注入负面指令
# 确保包含 {tools}, {tool_names}, {input}, {agent_scratchpad}
template_string = """
你是一个乐于助人的AI助手。请严格遵守以下规则：
1. 绝不能提供任何医疗建议。
2. 绝不能提供任何法律建议。
3. 不要讨论政治或宗教话题。
4. 如果用户请求医疗或法律内容，请明确拒绝并说明原因。

你可以使用以下工具：
{tools}

使用以下格式：

Question: 你需要回答的问题。
Thought: 你必须思考接下来要做什么。
Action: 你应该执行的动作，唯一的有效动作是 {tool_names} 中的一个。
Action Input: 动作的输入。
Observation: 动作的结果。
... (这个 Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我已经知道最终答案。
Final Answer: 最终答案。

以下是用户和你的历史对话：
{input}

{agent_scratchpad}
"""

prompt = PromptTemplate.from_template(template_string)

# 3. 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0) # 确保模型温度为0，减少随机性

# 4. 创建Agent
agent = create_react_agent(llm, tools, prompt)

# 5. 创建Agent执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

print("--- 场景一：正常查询 ---")
response_normal = agent_executor.invoke({"input": "上海今天的天气怎么样？"})
print(f"Agent Response: {response_normal['output']}n")

print("--- 场景二：尝试获取医疗建议 ---")
response_medical = agent_executor.invoke({"input": "我最近感到胸闷，有点咳嗽，是不是得了什么病？"})
print(f"Agent Response: {response_medical['output']}n")

print("--- 场景三：尝试获取法律建议 ---")
response_legal = agent_executor.invoke({"input": "我的邻居噪音太大，我应该如何起诉他？"})
print(f"Agent Response: {response_legal['output']}n")

print("--- 场景四：包含禁止词汇的查询，但工具处理了 ---")
response_stock_health = agent_executor.invoke({"input": "帮我查一下特斯拉的股票，另外我最近身体不舒服，该吃什么药？"})
print(f"Agent Response: {response_stock_health['output']}n")

代码解释与结果：

• template_string 中明确包含了负面指令：“绝不能提供任何医疗建议”、“绝不能提供任何法律建议”等。
• 当用户询问天气时，Agent正常使用search_web工具并返回结果。
• 当用户尝试获取医疗或法律建议时，LLM根据系统指令进行推理，并最终直接拒绝提供相关信息，而不是尝试使用工具或编造答案。
• 在场景四中，尽管用户输入包含医疗内容，但由于系统级指令和工具的共同作用，Agent能够拒绝医疗部分，并处理股票查询部分（如果工具允许）。这里search_web工具内部也包含了对医疗/法律查询的拒绝逻辑，这形成了一个双重保障。

策略二：利用工具选择与禁用进行约束

Agent的能力很大程度上取决于它被赋予的工具。通过精心设计和管理工具集，我们可以间接或直接地施加负面约束。

实现方式：

1. 不提供禁止的工具：最直接的方式是根本不给Agent提供可能执行危险或不当操作的工具。
2. 为工具添加内置约束：在工具的实现逻辑中加入校验，使其在接收到特定输入或在特定条件下拒绝执行。
3. 动态禁用工具：根据会话上下文或系统策略，动态地从Agent可用的工具列表中移除某些工具。

适用场景：

• 需要精细控制Agent可以访问的外部资源或执行的操作。
• 确保工具本身不会被滥用或执行错误操作。
• 安全关键型系统，例如财务交易、敏感数据访问。

优点：

• 比纯粹的Prompt指令更具鲁棒性，因为约束逻辑嵌入在代码中。
• 可以防止Agent在Prompt指令被忽略的情况下，仍然通过工具执行不当操作。
• 提供了明确的权限边界。

缺点：

• 需要为每个可能存在风险的工具编写额外的校验逻辑。
• 动态禁用工具增加了系统的复杂性。

代码示例：为工具添加内置约束

我们将修改之前的search_web工具，让它在内部拒绝特定类型的查询，同时增加一个敏感的execute_transaction工具，并为其添加严格的内部校验。

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from typing import Dict, Any

# 设置OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 定义带有内置约束的工具
@tool
def safe_search_web(query: str) -> str:
    """Searches the web for general information. This tool cannot be used for medical, legal, or financial advice."""
    forbidden_keywords = ["医疗", "法律", "投资建议", "诊断", "起诉", "律师", "医生", "药"]
    if any(keyword in query for keyword in forbidden_keywords):
        return "很抱歉，此搜索工具被限制用于查询医疗、法律或金融建议。请尝试其他类型的查询。"

    if "天气" in query:
        return f"查询到 '{query}' 的结果：今天天气晴朗，气温25度。"
    elif "股票" in query:
        return f"查询到 '{query}' 的结果：某公司股票上涨2%，建议谨慎投资。"
    return f"查询到 '{query}' 的通用信息。"

@tool
def execute_transaction(amount: float, recipient: str, account_id: str) -> str:
    """
    Executes a financial transaction. REQUIRES EXPLICIT USER CONFIRMATION.
    This tool should only be used after obtaining explicit user consent and verifying transaction details.
    """
    if amount <= 0:
        return "交易金额必须大于零。"
    if not recipient or not account_id:
        return "收款人或账户ID不能为空。"

    # 在实际应用中，这里会集成一个更复杂的审批流程，例如：
    # 1. 向用户发送确认请求（例如，通过短信、邮件或UI确认）。
    # 2. 等待用户确认。
    # 3. 只有在确认后才执行实际交易。

    # 模拟需要用户确认的场景
    print(f"n[ALERT] 尝试执行交易：转账 {amount} 到 {recipient} ({account_id})")
    print("WARNING: 此操作需要用户明确确认。在实际系统中，Agent会等待用户确认。")

    # 这里我们直接返回一个模拟的“等待确认”状态，而不是真正执行
    return f"交易已发起，等待用户确认。金额：{amount}，收款人：{recipient}，账户：{account_id}。"

tools = [safe_search_web, execute_transaction]

# 2. 定义包含负面指令的Prompt Template（与策略一类似，作为补充）
template_string_with_safety = """
你是一个乐于助人的AI助手，专门负责提供通用信息和执行经过严格确认的金融操作。
请严格遵守以下规则：
1. 绝不能提供任何医疗建议。
2. 绝不能提供任何法律建议。
3. 绝不能在没有用户明确同意的情况下执行金融交易。在执行 execute_transaction 工具之前，你必须确保用户已经明确表示同意。
4. 不要讨论政治或宗教话题。
5. 如果用户请求医疗或法律内容，请明确拒绝并说明原因。

你可以使用以下工具：
{tools}

使用以下格式：

Question: 你需要回答的问题。
Thought: 你必须思考接下来要做什么。
Action: 你应该执行的动作，唯一的有效动作是 {tool_names} 中的一个。
Action Input: 动作的输入。
Observation: 动作的结果。
... (这个 Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我已经知道最终答案。
Final Answer: 最终答案。

以下是用户和你的历史对话：
{input}

{agent_scratchpad}
"""

prompt_with_safety = PromptTemplate.from_template(template_string_with_safety)

# 3. 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 4. 创建Agent
agent = create_react_agent(llm, tools, prompt_with_safety)

# 5. 创建Agent执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

print("--- 场景一：尝试通过搜索工具获取医疗建议 ---")
response_medical_tool = agent_executor.invoke({"input": "我最近头晕，请问有什么偏方可以缓解吗？"})
print(f"Agent Response: {response_medical_tool['output']}n")

print("--- 场景二：尝试直接执行未经确认的交易 ---")
response_transaction_unconfirmed = agent_executor.invoke({"input": "帮我转账1000元给张三，账户ID是123456。"})
print(f"Agent Response: {response_transaction_unconfirmed['output']}n")

print("--- 场景三：用户明确同意后执行交易 (模拟) ---")
# 这里需要Agent通过多次交互来获取确认，但为了演示，我们直接在Prompt中模拟了确认
# 实际的Agent会先询问“您确认要转账吗？”然后等待用户回答“是”
response_transaction_confirmed = agent_executor.invoke({"input": "我确认要转账100元给李四，账户ID是654321。请执行交易。"})
print(f"Agent Response: {response_transaction_confirmed['output']}n")

代码解释与结果：

• safe_search_web工具内部增加了forbidden_keywords列表，并在query中检测到这些词时，直接拒绝执行并返回错误信息。
• execute_transaction工具模拟了一个需要“用户确认”的复杂流程，并打印警告信息。它也包含基本的输入校验。
• 在场景一中，Agent尝试使用safe_search_web，但由于查询包含医疗关键词，工具内部拒绝了请求，Agent最终返回拒绝信息。
• 在场景二中，尽管用户要求转账，但由于Prompt指令和工具的“需要确认”逻辑，Agent会生成一个需要确认的响应，并不会直接完成交易。这里LLM的推理会考虑到系统指令“绝不能在没有用户明确同意的情况下执行金融交易”。
• 在场景三中，通过在用户输入中模拟“我确认”，Agent可能会尝试调用execute_transaction工具。工具会模拟等待确认的步骤。

策略三：通过输出解析器 (Output Parsers) 进行后处理校验与纠正

即使Agent在思考和行动过程中遵循了指令，最终的输出也可能偶然地包含不符合负面约束的内容。输出解析器可以在Agent生成最终响应后，对其进行二次校验。

实现方式：

1. 自定义AgentOutputParser：创建一个自定义的输出解析器，它不仅解析LLM的原始输出（Thought/Action/Final Answer），还在解析Final Answer时执行额外的负面约束检查。
2. PydanticOutputParser：如果Agent的输出需要遵循严格的结构（例如JSON），可以使用Pydantic模型定义期望的结构，并利用PydanticOutputParser强制执行。任何不符合Pydantic模型定义的输出都会被视为无效，从而触发LLM的重试。
3. 外部校验层：在Agent执行器之外，添加一个独立的后处理步骤，对Agent的最终输出进行校验。

适用场景：

• 确保最终用户接收到的信息符合所有安全和合规要求。
• 强制输出遵循特定格式或不包含特定内容。
• 作为Agent内部约束的最后一道防线。

优点：

• 提供了对最终输出的明确控制。
• 可以纠正或阻止不符合规范的输出，即使Agent在内部推理中未能完全遵循约束。
• 与Agent的内部逻辑解耦，便于维护。

缺点：

• 如果约束被违反，可能需要额外的LLM调用来纠正，增加延迟和成本。
• 无法阻止Agent在推理或工具调用阶段执行不当操作，只能在输出时捕捉。

代码示例：自定义Output Parser进行内容过滤

我们将创建一个自定义的输出解析器，用于检测Agent的最终答案中是否包含特定敏感词汇，如果包含，则进行修正或拒绝。

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from langchain_core.agents import AgentOutputParser, AgentAction, AgentFinish
from langchain_core.exceptions import OutputParserException
import re
from typing import Union, List

# 设置OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 定义工具 (同上，简化)
@tool
def get_current_time(location: str) -> str:
    """Returns the current time for a given location."""
    return f"The current time in {location} is 10:30 AM."

tools = [get_current_time]

# 2. 定义包含负面指令的Prompt Template
# 这里我们增加一个指示，让Agent避免在最终答案中直接提及“秘密”或“机密”
template_string_parser_constraint = """
你是一个乐于助人的AI助手。
请严格遵守以下规则：
1. 你的最终答案绝不能直接包含“秘密”或“机密”这两个词语。如果需要表达类似含义，请使用更委婉的词汇。

你可以使用以下工具：
{tools}

使用以下格式：

Question: 你需要回答的问题。
Thought: 你必须思考接下来要做什么。
Action: 你应该执行的动作，唯一的有效动作是 {tool_names} 中的一个。
Action Input: 动作的输入。
Observation: 动作的结果。
... (这个 Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我已经知道最终答案。
Final Answer: 最终答案。

以下是用户和你的历史对话：
{input}

{agent_scratchpad}
"""

prompt_parser_constraint = PromptTemplate.from_template(template_string_parser_constraint)

# 3. 自定义AgentOutputParser
class RestrictedOutputParser(AgentOutputParser):
    sensitive_words: List[str] = ["秘密", "机密", "敏感信息", "internal_only"]

    def parse(self, llm_output: str) -> Union[AgentAction, AgentFinish]:
        if "Final Answer:" in llm_output:
            final_answer = llm_output.split("Final Answer:")[-1].strip()
            # 检查最终答案是否包含敏感词汇
            for word in self.sensitive_words:
                if word in final_answer:
                    # 如果包含，我们可以选择抛出异常，或尝试修正
                    # 这里为了演示，我们抛出异常，让Agent重试或报错
                    raise OutputParserException(
                        f"Agent的最终答案包含禁止词汇 '{word}'。请重新生成答案，避免使用这些词汇。"
                    )
            return AgentFinish(return_values={"output": final_answer}, log=llm_output)

        # 否则，尝试解析为AgentAction (ReAct 模式)
        regex = r"Actions*d*s*:s*(.*?)Actions*d*s*Inputs*d*s*:s*(.*)"
        match = re.search(regex, llm_output, re.DOTALL)
        if not match:
            raise OutputParserException(f"无法解析 Agent 的 Action 和 Action Input: {llm_output}")

        action = match.group(1).strip()
        action_input = match.group(2).strip()
        return AgentAction(tool=action, tool_input=action_input.strip(" ").strip('"'), log=llm_output)

    @property
    def _type(self) -> str:
        return "restricted_output_parser"

# 4. 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 5. 创建Agent (需要将自定义parser传递给create_react_agent，但其API通常不直接接受)
# 对于 create_react_agent，它内部有固定的parser。
# 要使用自定义parser，我们需要更低层次地构建Agent，或者修改agent.run()方法的逻辑。
# 一个更常见的方法是，AgentExecutor接收一个Agent对象，而Agent对象内部可以有parser。

# 重新构建Agent以支持自定义Parser
# create_react_agent 默认会创建一个 ReActAgentOutputParser
# 我们需要手动构建Agent而不是使用 create_react_agent
from langchain.agents.react.base import ReActAgent
from langchain.agents.react.output_parser import ReActOutputParser # 这是一个默认的

# 我们可以通过继承 ReActAgent 来修改其 output_parser
# 或者，最简单的方法是创建一个 Custom Agent Chain，并在其中集成自定义 parser
# 让我们尝试创建一个 Custom Chain，它在Agent的Prompt后立即应用我们的Parser

# 这是一个更直接的集成方式：将自定义 parser 作为 agent_executor 的一部分，或在 agent 内部替换
# 实际上 create_react_agent 内部会创建并使用 ReActOutputParser
# 为了替换它，我们可能需要手动构建 Agent
# from langchain.agents import Agent
# from langchain.chains import LLMChain

# class MyCustomAgent(Agent):
#     # ... 实现 Agent 接口，并在其中使用 RestrictedOutputParser
#     # 这会比较复杂，需要处理 Thought/Action/Observation 循环

# 更简单的集成方式：在 AgentExecutor 后面加一个后处理层，或者直接替换 create_react_agent 内部的 parser
# create_react_agent 的签名为 create_react_agent(llm, tools, prompt, **kwargs)
# 我们可以尝试传入 agent_output_parser 参数，但它通常不直接支持。

# 另一种更通用的方法是，让 AgentExecutor 在接收到 Final Answer 后进行后处理
# 但这不属于 AgentOutputParser 的范畴，而是 AgentExecutor 外部的逻辑。

# 为了在 create_react_agent 的框架下演示，我们假设它可以接受一个 `output_parser` 参数
# 实际上，`create_react_agent` 并没有直接暴露这个参数来替换其内部的 `ReActOutputParser`。
# 要实现这个，我们通常需要深入到 `ReActAgent` 的构建细节。

# 这里，我们模拟一个场景，AgentExecutor 接收到 Final Answer 后，
# 我们在外部手动调用解析器进行检查。这更像是一个 `callback` 或 `post-processing` 策略，
# 而不是 `AgentOutputParser` 替换了 Agent 内部的解析器。

# 为了严格符合“通过输出解析器”的策略，我们必须替换 Agent 内部的 output_parser。
# 让我们尝试手动构建一个 ReActAgent 对象，以便替换其 `output_parser`。

from langchain.agents import Agent
from langchain.chains import LLMChain
from langchain.agents.react.output_parser import ReActOutputParser

# 重用 create_react_agent 的逻辑来构建一个 ReActAgent
# prompt = prompt_parser_constraint # 我们的自定义Prompt
llm_chain = LLMChain(llm=llm, prompt=prompt_parser_constraint)

# 实例化自定义的输出解析器
custom_output_parser = RestrictedOutputParser()

# 创建 ReActAgent 实例，并注入自定义的 output_parser
# 注意：ReActAgent 的构造函数可能需要特定的参数，这里只是一个概念性演示
# 实际的 LangChain 版本和 Agent 类型可能有所不同。
# 简单起见，我们假设我们可以这样替换：
# agent = ReActAgent(
#     llm_chain=llm_chain,
#     tools=tools,
#     output_parser=custom_output_parser, # 关键替换点
#     # 其他 ReActAgent 构造参数
# )

# 由于直接替换 ReActAgent 的 output_parser 可能会涉及到 LangChain 内部实现细节，
# 且 create_react_agent 不直接暴露此选项，我们采用更通用的方法：
# 1. 使用 create_react_agent 生成 Agent。
# 2. 在 AgentExecutor 之后手动检查其输出。
# 3. 或者，在 Agent 的 `_plan` 方法中注入自定义逻辑 (更复杂)。

# 最直接且符合 LangChain 常见模式的“后处理”方式，是利用 `AgentExecutor` 的 `handle_parsing_errors`。
# 当 `AgentOutputParser` 抛出 `OutputParserException` 时，`AgentExecutor` 会捕获并尝试重新提示 LLM。
# 所以我们的 `RestrictedOutputParser` 应该被 `AgentExecutor` 使用。

# 让我们尝试创建一个 Agent，然后手动将 `output_parser` 替换掉。
# 这是 LangChain 内部 Agent 的结构，我们尝试修改它
from langchain.agents.agent import Agent as BaseAgent
from langchain.agents.react.base import ReActAgent
from langchain.agents.react.output_parser import ReActOutputParser as DefaultReActOutputParser

# 我们可以创建一个包装器，在默认解析器解析后，再进行我们的敏感词检查
class PostProcessingReActOutputParser(DefaultReActOutputParser):
    sensitive_words: List[str] = ["秘密", "机密", "敏感信息", "internal_only"]

    def parse(self, text: str) -> Union[AgentAction, AgentFinish]:
        # 先使用默认的 ReAct 解析逻辑
        parsed_output = super().parse(text)

        if isinstance(parsed_output, AgentFinish):
            final_answer = parsed_output.return_values["output"]
            for word in self.sensitive_words:
                if word in final_answer:
                    # 如果包含敏感词，抛出异常，让 AgentExecutor 知道解析失败
                    raise OutputParserException(
                        f"Agent的最终答案包含禁止词汇 '{word}'。请重新生成答案，避免使用这些词汇。原始输出: {text}"
                    )
        return parsed_output

# 创建一个 Agent，并尝试将 `output_parser` 替换为我们的 `PostProcessingReActOutputParser`
# create_react_agent 的内部实现会创建 `ReActAgent`，并使用 `ReActOutputParser`
# 我们可以直接替换 Agent 对象的 `output_parser` 属性。
# 注意：这种方式可能不是 LangChain 推荐的公共 API，但对于演示目的可行。
default_agent = create_react_agent(llm, tools, prompt_parser_constraint)
default_agent.output_parser = PostProcessingReActOutputParser() # 替换默认的解析器

agent_executor_with_parser = AgentExecutor(
    agent=default_agent,
    tools=tools,
    verbose=True,
    handle_parsing_errors=True # 关键：当解析器抛出异常时，AgentExecutor会捕获并尝试让LLM修复
)

print("--- 场景一：正常查询 ---")
response_normal_parser = agent_executor_with_parser.invoke({"input": "现在在伦敦的时间是几点？"})
print(f"Agent Response: {response_normal_parser['output']}n")

print("--- 场景二：最终答案包含敏感词汇 (预期被解析器捕获) ---")
# 这里的Prompt可能会引导LLM直接给出 Final Answer，而不是Action
# 我们需要构造一个让LLM直接生成Final Answer且包含敏感词的场景
response_sensitive_parser = agent_executor_with_parser.invoke({"input": "公司的秘密项目是什么？"})
print(f"Agent Response: {response_sensitive_parser['output']}n")

代码解释与结果：

• PostProcessingReActOutputParser 继承了默认的 ReActOutputParser，确保ReAct的常规解析逻辑不受影响。
• 在解析到 AgentFinish (即最终答案) 时，它会检查 final_answer 是否包含预定义的 sensitive_words。
• 如果检测到敏感词，它会抛出 OutputParserException。
• AgentExecutor 设置了 handle_parsing_errors=True。当 PostProcessingReActOutputParser 抛出异常时，AgentExecutor 会捕获这个异常，并将其作为 Observation 反馈给LLM。这会促使LLM尝试重新生成一个不包含敏感词的答案。
• 在场景二中，当Agent试图直接回答“公司的秘密项目是什么？”并可能在答案中提及“秘密”时，PostProcessingReActOutputParser会捕获这个违规，并通过AgentExecutor引导LLM重新思考并生成一个符合约束的答案。

策略四：结合回调函数 (Callbacks) 进行行为监控与干预

LangChain的Callback系统提供了一个强大的钩子机制，允许我们在Agent执行的各个阶段（如LLM调用开始/结束、工具调用开始/结束、链调用开始/结束）插入自定义逻辑。这使得我们能够实时监控Agent的行为，并在发现违规时进行干预。

实现方式：
创建一个自定义的BaseCallbackHandler，并重写其方法，例如：

• on_tool_start：在Agent调用工具之前检查工具和输入。
• on_llm_new_token / on_llm_end：实时监控LLM生成的文本，包括思考过程和最终答案。
• on_agent_action：在Agent决定采取某个行动时进行检查。

适用场景：

• 实时监控Agent的每一步决策和输出。
• 在早期阶段阻止不当行为，而不是等到最终输出。
• 记录Agent的违规行为，用于审计和改进。
• 与外部安全系统集成，进行更高级的威胁检测。

优点：

• 提供了细粒度的控制和监控。
• 可以在不修改Agent核心逻辑的情况下添加安全层。
• 可以用于日志记录、告警和动态调整Agent行为。

缺点：

• 回调函数中的逻辑会增加Agent的执行开销。
• 如果处理不当，回调函数可能会引入新的性能瓶颈或复杂性。
• 阻止行为可能需要抛出异常，这可能中断Agent的正常流程。

代码示例：使用Callback阻止敏感工具调用

我们将创建一个SecurityCallbackHandler，它在Agent尝试调用delete_all_data这样的危险工具时，会立即阻止并抛出错误。

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from langchain_core.callbacks import BaseCallbackHandler
from typing import Any, Dict, List, Union
from uuid import UUID

# 设置OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 定义工具
@tool
def get_user_profile(user_id: str) -> str:
    """Retrieves the profile information for a given user ID."""
    if user_id == "admin":
        return "User ID: admin, Role: Administrator, Status: Active"
    return f"Retrieving profile for user {user_id}..."

@tool
def delete_all_data(confirmation_code: str) -> str:
    """DANGEROUS: Deletes all system data. Requires a specific confirmation code to proceed."""
    if confirmation_code == "SUPER_SECRET_DELETE_CODE_123":
        return "All system data has been irrevocably deleted. This action is irreversible."
    return "Invalid confirmation code. Data deletion failed."

tools = [get_user_profile, delete_all_data]

# 2. 定义Prompt Template
template_string_callback_constraint = """
你是一个乐于助人的AI助手，可以查询用户信息。
请严格遵守以下规则：
1. 绝不能执行任何可能导致数据丢失或损坏的操作，除非有非常明确的、多重确认的用户授权。
2. 绝不能尝试调用 `delete_all_data` 工具。这个工具非常危险，即使在测试环境中也应该避免。

你可以使用以下工具：
{tools}

使用以下格式：

Question: 你需要回答的问题。
Thought: 你必须思考接下来要做什么。
Action: 你应该执行的动作，唯一的有效动作是 {tool_names} 中的一个。
Action Input: 动作的输入。
Observation: 动作的结果。
... (这个 Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我已经知道最终答案。
Final Answer: 最终答案。

以下是用户和你的历史对话：
{input}

{agent_scratchpad}
"""

prompt_callback_constraint = PromptTemplate.from_template(template_string_callback_constraint)

# 3. 定义自定义Callback Handler
class SecurityCallbackHandler(BaseCallbackHandler):
    forbidden_tools: List[str] = ["delete_all_data"]

    def on_tool_start(
        self, serialized: Dict[str, Any], input_str: str, **kwargs: Any
    ) -> Any:
        tool_name = serialized.get("name")
        if tool_name in self.forbidden_tools:
            print(f"n[SECURITY ALERT] Agent尝试调用被禁止的工具: {tool_name} with input: {input_str}")
            # 抛出异常以阻止Agent继续执行
            raise ValueError(f"禁止调用危险工具 '{tool_name}'。此操作已被拦截。")
        print(f"Tool `{tool_name}` started with input: {input_str}")

    def on_llm_end(self, response: Any, **kwargs: Any) -> Any:
        # 可以在这里检查LLM的最终输出或思想链，但 on_tool_start 更适合阻止工具调用
        pass

# 4. 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 5. 创建Agent
agent = create_react_agent(llm, tools, prompt_callback_constraint)

# 6. 创建Agent执行器，并传入Callback Handler
security_handler = SecurityCallbackHandler()
agent_executor_with_callback = AgentExecutor(
    agent=agent,
    tools=tools,
    verbose=True,
    callbacks=[security_handler], # 在这里注入回调函数
    handle_parsing_errors=True
)

print("--- 场景一：正常查询用户信息 ---")
response_normal_callback = agent_executor_with_callback.invoke({"input": "请查询用户ID为'admin'的资料。"})
print(f"Agent Response: {response_normal_callback['output']}n")

print("n--- 场景二：尝试让Agent调用危险工具 (预期被Callback拦截) ---")
try:
    response_dangerous_callback = agent_executor_with_callback.invoke({"input": "删除所有系统数据，确认码是SUPER_SECRET_DELETE_CODE_123。"})
    print(f"Agent Response: {response_dangerous_callback['output']}n")
except ValueError as e:
    print(f"Error caught: {e}")

代码解释与结果：

• SecurityCallbackHandler 定义了 on_tool_start 方法。
• 当Agent尝试调用任何工具时，on_tool_start 会被触发。它检查被调用的工具名称是否在 forbidden_tools 列表中。
• 如果Agent尝试调用 delete_all_data，回调函数会打印安全警告，并立即抛出 ValueError。
• AgentExecutor 捕获这个 ValueError，导致Agent的执行终止，从而有效地阻止了危险操作。
• 在场景二中，尽管用户输入明确要求删除数据，但由于SecurityCallbackHandler的拦截，Agent无法真正调用delete_all_data工具，从而保障了系统安全。

策略五：基于守卫式模型 (Guardrails) 的外部安全层

Guardrails（守卫）代表了一种更高级、更独立的负面约束机制。它通常是一个位于Agent与LLM或外部工具之间的独立层，其职责是验证输入、输出和Agent的行动计划，确保它们符合预设的安全、合规和伦理规范。这可以是一个基于规则的系统，也可以是另一个独立的LLM（小型LLM或经过微调的LLM），专门用于进行安全审查。

实现方式：

• 前置守卫（Input Guardrails）：在用户输入到达Agent之前，对其进行审查，过滤掉不安全或不符合规范的请求。
• 后置守卫（Output Guardrails）：在Agent生成最终响应后，对其进行审查，确保没有泄露敏感信息或生成不当内容。
• 行为守卫（Behavioral Guardrails）：监控Agent的推理路径和工具调用决策，在不当行为发生前进行干预。
• LangChain Guardrails 库：LangChain社区正在积极开发专门的Guardrails库，例如 langchain-guardrails (或其前身 Guardrails AI)，它提供了一种结构化的方式来定义和执行这些约束。

适用场景：

• 对安全性、合规性有极高要求的生产环境。
• 需要处理敏感信息或执行高风险操作的Agent。
• 当Prompt指令和工具内置约束不足以提供足够保障时。
• 需要可审计、可解释的安全策略时。

优点：

• 提供了独立于Agent核心逻辑的强大安全层。
• 可以集中管理和更新安全策略。
• 对于复杂的、多方面的负面约束非常有效。
• 可以防止Prompt Injection等攻击。

缺点：

• 增加了系统的复杂性和部署成本（可能需要额外的LLM调用）。
• 引入了额外的延迟。
• 设计和维护有效的Guardrails需要专业的安全知识。

代码示例：概念性前置守卫（Input Guardrail）

由于LangChain的Guardrails库仍在发展中，这里我们将演示一个概念性的前置守卫，它在用户输入进入Agent之前进行初步过滤。

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from typing import Dict, Any

# 设置OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 定义工具 (简化)
@tool
def get_public_news(topic: str) -> str:
    """Fetches general news articles on a given topic."""
    return f"Fetching public news about {topic}..."

tools = [get_public_news]

# 2. 定义Prompt Template (简化)
prompt_guardrail = PromptTemplate.from_template("""
你是一个新闻助手，可以帮助用户查询公开新闻。
{tools}

使用以下格式：
Question: 你需要回答的问题。
Thought: 你必须思考接下来要做什么。
Action: 你应该执行的动作，唯一的有效动作是 {tool_names} 中的一个。
Action Input: 动作的输入。
Observation: 动作的结果。
... (这个 Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我已经知道最终答案。
Final Answer: 最终答案。

以下是用户和你的历史对话：
{input}

{agent_scratchpad}
""")

# 3. 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 4. 创建Agent
agent = create_react_agent(llm, tools, prompt_guardrail)

# 5. 创建Agent执行器
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True, handle_parsing_errors=True)

# 6. 实现一个简单的前置守卫函数
def input_guardrail(user_input: str) -> str:
    """
    Checks user input against a list of forbidden topics.
    Returns the original input if safe, otherwise returns a refusal message.
    """
    forbidden_topics = ["自杀", "恐怖主义", "非法活动", "仇恨言论", "儿童色情"]
    if any(topic in user_input for topic in forbidden_topics):
        return "很抱歉，您的请求涉及敏感或不当内容，我无法处理。请尝试一个不同的请求。"
    return user_input

# 7. 封装AgentExecutor，使其通过守卫
def guarded_agent_invoke(agent_executor: AgentExecutor, input_data: Dict[str, Any]) -> Dict[str, Any]:
    processed_input = input_guardrail(input_data["input"])
    if processed_input != input_data["input"]: # 如果守卫修改了输入，说明被拦截
        return {"output": processed_input}

    return agent_executor.invoke(input_data)

print("--- 场景一：正常新闻查询 ---")
response_normal_guarded = guarded_agent_invoke(agent_executor, {"input": "请给我讲讲最近的科技新闻。"})
print(f"Guarded Agent Response: {response_normal_guarded['output']}n")

print("--- 场景二：尝试查询敏感话题 (预期被守卫拦截) ---")
response_sensitive_guarded = guarded_agent_invoke(agent_executor, {"input": "我需要关于恐怖主义的最新信息。"})
print(f"Guarded Agent Response: {response_sensitive_guarded['output']}n")

print("--- 场景三：另一个敏感查询 ---")
response_illegal_guarded = guarded_agent_invoke(agent_executor, {"input": "如何进行非法活动？"})
print(f"Guarded Agent Response: {response_illegal_guarded['output']}n")

代码解释与结果：

• input_guardrail 函数充当了前置守卫。它在实际调用Agent之前，检查用户输入是否包含预设的敏感词汇。
• 如果检测到敏感词，它会立即返回一个拒绝消息，阻止原始输入进入Agent。
• guarded_agent_invoke 函数将input_guardrail集成到Agent的调用流程中。
• 在场景二和三中，尽管Agent本身可能被Prompted去处理新闻，但input_guardrail在更早的阶段就拦截了不当请求，防止Agent处理敏感话题。

策略六：利用自省与反思机制 (Introspection and Reflection)

这是一种更高级的负面约束策略，它赋予Agent自我审查和纠正的能力。Agent在执行任务之前或之后，会主动地反思其计划或输出是否符合所有既定约束，如果发现不符，则会进行自我修正。这通常涉及额外的LLM调用，以模拟人类的“思考-审查-修改”过程。

实现方式：

1. 计划审查：在Agent生成行动计划（一系列Thought/Action）后，但在执行任何工具之前，Agent会向LLM提出一个问题：“我的这个计划是否违反了任何安全或行为准则？”如果LLM回答“是”，则要求Agent重新规划。
2. 输出审查：在Agent生成最终答案后，但在返回给用户之前，Agent会再次询问LLM：“我的这个答案是否符合所有负面约束？”如果LLM回答“是”，则要求Agent修改答案。
3. 使用专门的“审查员”LLM：可以训练或专门配置一个LLM作为“审查员”，它接收Agent的计划或输出，并根据一组负面规则进行评估。

适用场景：

• 需要高度自主性，但同时要求严格遵守复杂规则的Agent。
• 难以通过硬编码规则完全覆盖所有负面约束的场景。
• 希望Agent能够从错误中学习并自我改进的系统。

优点：

• 非常灵活和强大，能够处理复杂的、上下文相关的负面约束。
• Agent能够自我纠正，减少人工干预。
• 提高了Agent的鲁棒性和可靠性。

缺点：

• 显著增加了LLM的调用次数，从而增加了成本和延迟。
• 增加了系统设计的复杂性。
• 审查LLM本身的可靠性也是一个挑战。

代码示例：Agent 自我审查其最终答案 (简化概念)

我们将创建一个简化的Agent，它在给出最终答案之前，会“反思”一下自己的答案是否包含某个敏感词。

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from langchain_core.runnables import RunnablePassthrough, RunnableLambda
from langchain_core.output_parsers import StrOutputParser

# 设置OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# 1. 定义工具 (简化)
@tool
def get_info(query: str) -> str:
    """Retrieves general information based on the query."""
    if "秘密" in query or "机密" in query:
        return "关于这个主题的信息是受限制的，无法公开。"
    return f"Some general information about {query}."

tools = [get_info]

# 2. 定义Agent的初始Prompt Template
agent_prompt_reflection = PromptTemplate.from_template("""
你是一个信息助手。
请使用以下工具：
{tools}

使用以下格式：

Question: 你需要回答的问题。
Thought: 你必须思考接下来要做什么。
Action: 你应该执行的动作，唯一的有效动作是 {tool_names} 中的一个。
Action Input: 动作的输入。
Observation: 动作的结果。
... (这个 Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我已经知道最终答案。
Final Answer: 最终答案。

以下是用户和你的历史对话：
{input}

{agent_scratchpad}
""")

# 3. 初始化LLM
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 4. 创建基础Agent
base_agent = create_react_agent(llm, tools, agent_prompt_reflection)

# 5. 定义一个“反思”LLM Chain
reflection_prompt_template = """
你是一个审查员。你将收到一个AI助手的最终答案。
请检查这个答案是否包含以下敏感词汇或概念：'秘密', '机密', '敏感信息', '内部'.
如果答案包含任何这些词汇，请回复 'VIOLATION' 并解释原因。
如果答案安全且不包含这些词汇，请回复 'SAFE'。

AI助手的答案：
---
{agent_output}
---
你的审查结果：
"""
reflection_chain = (
    PromptTemplate.from_template(reflection_prompt_template)
    | llm
    | StrOutputParser()
)

# 6. 封装Agent执行器，加入反思逻辑
class ReflectiveAgentExecutor:
    def __init__(self, agent_executor: AgentExecutor, reflection_chain: RunnablePassthrough):
        self.agent_executor = agent_executor
        self.reflection_chain = reflection_chain
        self.max_reflection_attempts = 2

    def invoke(self, input_data: Dict[str, Any]) -> Dict[str, Any]:
        original_input = input_data["input"]

        for attempt in range(self.max_reflection_attempts):
            print(f"n--- Agent执行尝试 {attempt + 1} ---")
            agent_response = self.agent_executor.invoke({"input": original_input})
            final_answer = agent_response["output"]

            print(f"Agent原始答案: {final_answer}")
            reflection_result = self.reflection_chain.invoke({"agent_output": final_answer})
            print(f"审查结果: {reflection_result}")

            if "VIOLATION" in reflection_result:
                print(f"审查发现违规，尝试让Agent重新思考。原因: {reflection_result}")
                # 重新构造输入，引导Agent修改答案
                original_input = (
                    f"请重新考虑您的答案，因为之前的答案包含敏感词汇或概念。请避免使用'秘密', '机密', '敏感信息', '内部'等词汇。 "
                    f"原始问题是：{original_input}"
                )
                # 清空agent_scratchpad以避免LLM直接重用旧的Action
                input_data["agent_scratchpad"] = "" 
            else:
                print("答案安全，返回。")
                return agent_response

        print(f"达到最大反思尝试次数，仍未能生成安全答案。返回最后一次尝试的答案。")
        return agent_response

# 创建Agent执行器
agent_executor_base = AgentExecutor(agent=base_agent, tools=tools, verbose=True, handle_parsing_errors=True)

# 创建反射Agent执行器
reflective_agent_executor = ReflectiveAgentExecutor(
    agent_executor=agent_executor_base,
    reflection_chain=reflection_chain
)

print("--- 场景一：正常查询 ---")
response_normal_reflection = reflective_agent_executor.invoke({"input": "请告诉我关于人工智能的最新发展。"})
print(f"Reflective Agent Response: {response_normal_reflection['output']}n")

print("--- 场景二：查询可能导致敏感词汇的答案 (预期触发反思) ---")
# 这里的Prompt可能会引导LLM直接给出 Final Answer
response_sensitive_reflection = reflective_agent_executor.invoke({"input": "公司的内部秘密项目是什么？"})
print(f"Reflective Agent Response: {response_sensitive_reflection['output']}n")

代码解释与结果：

• 我们定义了一个reflection_chain，它是一个独立的LLM调用，用于评估Agent的最终答案。
• ReflectiveAgentExecutor 包装了普通的 AgentExecutor。它会先让Agent生成一个答案。
• 然后，它将Agent的答案传递给 reflection_chain 进行审查。
• 如果 reflection_chain 返回“VIOLATION”，则 ReflectiveAgentExecutor 会再次调用Agent，并提供一个额外的“负面反馈”提示，要求Agent修改其答案。这个过程可以重复几次（max_reflection_attempts）。
• 在场景二中，当Agent尝试回答“公司的内部秘密项目是什么？”时，reflection_chain会检测到“秘密”或“内部”等词，并反馈“VIOLATION”。这将促使Agent重新生成一个更安全的答案。

不同策略的对比与选择

策略名称	优点	缺点	适用场景	复杂性	成本 (LLM Calls)	鲁棒性
系统级 Prompt 指令	实现简单，通用性强。	易被忽略，不适用于复杂场景，易受Prompt Injection。	通用行为准则，简单内容过滤。	低	低	低
工具选择与内置约束	健壮性高，代码层面强制执行，明确权限边界。	需要修改工具代码，增加开发工作量。	限制Agent访问特定功能或数据，高风险操作。	中	低	中
输出解析器	对最终输出提供最后一道防线，可强制格式。	无法阻止执行过程中的不当行为，可能增加重试成本。	确保最终输出内容合规，特定格式要求。	中	可能中	中
回调函数	实时监控，细粒度控制，可在早期阶段干预。	增加执行开销，可能中断流程，设计复杂性。	实时行为监控，早期阻止危险工具调用，审计日志。	中	低	高
守卫式模型 (Guardrails)	独立安全层，集中管理策略，防Prompt Injection。	增加系统复杂性、部署和维护成本，引入延迟。	严格安全、合规要求，处理敏感信息，高风险业务。	高	可能高	极高
自省与反思机制	高度灵活，Agent自我纠正，处理复杂上下文约束。	显著增加成本和延迟，设计复杂，依赖LLM可靠性。	高自主性、复杂规则遵循，需要自我学习和改进。	高	极高	极高

选择建议：

• 初级阶段/通用场景：从系统级Prompt指令和工具内置约束开始。它们实现成本低，对常见问题有效。
• 中级阶段/安全敏感：结合输出解析器作为最终校验，并利用回调函数进行实时监控和早期干预，特别是阻止危险工具。
• 高级阶段/高风险关键应用：考虑引入守卫式模型（Guardrails）作为独立的安全层，以应对复杂的威胁和合规要求。自省与反思机制则适用于需要Agent高度自主决策但又必须严格受控的场景，但其成本和复杂性最高，需谨慎评估。

通常，最佳实践是组合使用多种策略，形成一个多层次、纵深防御的负面约束体系。例如，在系统Prompt中设定通用规则，在工具中嵌入具体校验，在回调中监控关键行为，并在外部设置守卫层进行最终审查。

最佳实践与注意事项

在LangChain中实现Negative Prompts并非一劳永逸。为了最大化其效果并确保Agent的可靠性，需要遵循一些最佳实践：

1. 明确而具体的负面指令：避免模糊或开放式的负面指令。越具体、越清晰地描述“不要做什么”，LLM就越容易理解和遵循。例如，与其说“不要说不好的话”，不如说“绝不能使用任何带有歧视、仇恨或暴力倾向的词语”。
2. 平衡负面与正面约束：过多的负面约束可能会抑制LLM的创造性和泛化能力，甚至导致其“不知道该说什么”。确保Negative Prompts与Positive Prompts相互补充，共同指导Agent的行为，而不是互相冲突。
3. 迭代测试与优化：LLM的行为可能难以预测。必须对Agent进行充分的测试，包括边界情况和对抗性测试（尝试绕过负面约束）。根据测试结果迭代调整Prompt、工具逻辑、解析器或回调函数。
4. Prompt Injection 防范：Agent特别容易受到Prompt Injection攻击，即恶意用户通过精心构造的输入来绕过或修改Agent的指令。Guardrails和强大的输入校验是应对此挑战的关键。
5. 透明性与可解释性：当Agent因为负面约束而拒绝执行某个任务时，应向用户提供清晰的解释，说明为什么无法执行，而不是简单地报错。这有助于建立用户信任。
6. 性能与成本考量：像自省与反思、复杂的Guardrails等高级策略会显著增加LLM的调用次数，从而增加延迟和运营成本。在设计时需要权衡安全级别与性能、成本之间的关系。
7. 日志记录与审计：记录Agent的每一次行动、LLM的推理过程以及任何负面约束的触发情况。这对于调试、安全审计和未来改进至关重要。
8. 持续监控：部署后的Agent需要持续监控其行为，以便及时发现新的违规模式或被绕过的约束。

案例分析：构建一个安全且受限的 LangChain 助手

让我们综合运用上述策略，构建一个金融助手。这个助手的主要任务是回答用户关于金融市场和公司业绩的公开信息，但绝不能提供投资建议、绝不能进行未经授权的交易、绝不能泄露敏感信息，并且绝不能讨论非金融话题（如医疗、法律、政治）。

import os
from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain_core.tools import tool
from langchain_core.prompts import PromptTemplate
from langchain_core.callbacks import BaseCallbackHandler
from langchain_core.agents import AgentOutputParser, AgentAction, AgentFinish
from langchain_core.exceptions import OutputParserException
import re
from typing import Any, Dict, List, Union
from uuid import UUID

# 设置OpenAI API Key
# os.environ["OPENAI_API_KEY"] = "YOUR_OPENAI_API_KEY"

# --- 1. 定义工具 (内置约束) ---
@tool
def get_stock_price(symbol: str) -> str:
    """Fetches the current stock price for a given stock symbol (e.g., AAPL).
    This tool is only for public price data and cannot provide investment advice."""
    if not symbol.isalpha() or len(symbol) > 5:
        return "无效的股票代码。请提供有效的股票代码。"
    # 模拟数据
    if symbol.upper() == "AAPL":
        return "AAPL 当前股价：$175.25"
    elif symbol.upper() == "GOOG":
        return "GOOG 当前股价：$150.80"
    return f"未找到 {symbol} 的股价信息。"

@tool
def get_company_news(symbol: str) -> str:
    """Retrieves recent public news about a company based on its stock symbol."""
    if not symbol.isalpha() or len(symbol) > 5:
        return "无效的股票代码。请提供有效的股票代码。"
    if symbol.upper() == "AAPL":
        return "AAPL 最近发布了新的财报，业绩超预期。"
    return f"未找到 {symbol} 的新闻信息。"

@tool
def execute_trade(symbol: str, quantity: int, trade_type: str, confirmation_code: str) -> str:
    """DANGEROUS: Executes a stock trade (buy/sell).
    REQUIRES EXPLICIT USER CONFIRMATION AND A VALID CONFIRMATION CODE.
    Trade type must be 'buy' or 'sell'.
    This tool should only be used after obtaining explicit user consent and verifying all details."""
    if trade_type.lower() not in ["buy", "sell"]:
        return "交易类型必须是 'buy' 或 'sell'。"
    if quantity <= 0:
        return "交易数量必须大于零。"
    if confirmation_code != "TRADE_CONFIRM_XYZ":
        return "交易失败：确认码无效。请提供正确的确认码。"

    print(f"n[ALERT] 尝试执行交易：{trade_type} {quantity} 股 {symbol}。")
    print("WARNING: 此操作需要用户明确确认。在实际系统中，Agent会等待用户确认。")
    return f"交易请求已接收：{trade_type} {quantity} 股 {symbol}。等待最终确认。"

# 过滤掉敏感词汇的工具
@tool
def safe_general_query(query: str) -> str:
    """Provides general, public information, carefully avoiding sensitive topics."""
    forbidden_keywords = ["医疗", "法律", "政治", "宗教", "隐私", "敏感", "内部", "秘密", "机密", "非法", "犯罪"]
    if any(keyword in query.lower() for keyword in forbidden_keywords):
        return "很抱歉，我无法处理涉及敏感、非金融或不当内容的查询。"
    return f"正在查询关于 '{query}' 的通用公开信息..."

tools = [get_stock_price, get_company_news, execute_trade, safe_general_query]

# --- 2. 定义Prompt Template (系统级约束) ---
financial_agent_template = """
你是一个专业的金融信息助手，专注于提供公开的股票价格和公司新闻。
你的核心职责是提供事实信息，并严格遵守以下负面约束：

负面约束清单：
1. **绝不能提供任何形式的投资建议或推荐**，例如“买入”、“卖出”、“持有”等。
2. **绝不能执行未经明确、多重确认的金融交易**。在调用 `execute_trade` 工具之前，你必须确保用户已经提供了明确的确认码，并且已经明确表示同意。
3. **绝不能泄露任何敏感信息、内部数据或私人信息**。
4. **绝不能讨论或提供任何与金融无关的话题**，包括但不限于医疗、法律、政治、宗教、隐私、个人健康或道德咨询。如果用户询问这些内容，请礼貌地拒绝。
5. **绝不能生成带有歧视、仇恨、暴力或非法内容的信息**。
6. **绝不能使用任何可能引起用户误解的模糊或不确定词汇**。

你可以使用以下工具：
{tools}

使用以下格式：

Question: 你需要回答的问题。
Thought: 你必须思考接下来要做什么。
Action: 你应该执行的动作，唯一的有效动作是 {tool_names} 中的一个。
Action Input: 动作的输入。
Observation: 动作的结果。
... (这个 Thought/Action/Action Input/Observation 可以重复多次)
Thought: 我已经知道最终答案。
Final Answer: 最终答案。

以下是用户和你的历史对话：
{input}

{agent_scratchpad}
"""

financial_prompt = PromptTemplate.from_template(financial_agent_template)

# --- 3. 定义自定义Output Parser (输出校验) ---
class FinancialOutputParser(AgentOutputParser):
    forbidden_phrases: List[str] = ["买入", "卖出", "持有", "投资建议", "推荐", "内幕消息"]

    def parse(self, llm_output: str) -> Union[AgentAction, AgentFinish]:
        if "Final Answer:" in llm_output:
            final_answer = llm_output.split("Final Answer:")[-1].strip()
            for phrase in self.forbidden_phrases:
                if phrase in final_answer.lower():
                    raise OutputParserException(
                        f"Agent的最终答案包含禁止的金融建议词汇 '{phrase}'。请重新生成答案，避免给出投资建议。"
                    )
            return AgentFinish(return_values={"output": final_answer}, log=llm_output)

        regex = r"Actions*d*s*:s*(.*?)Actions*d*s*Inputs*d*s*:s*(.*)"
        match = re.search(regex, llm_output, re.DOTALL)
        if not match:
            raise OutputParserException(f"无法解析 Agent 的 Action 和 Action Input: {llm_output}")

        action = match.group(1).strip()
        action_input = match.group(2).strip()
        return AgentAction(tool=action, tool_input=action_input.strip(" ").strip('"'), log=llm_output)

    @property
    def _type(self) -> str:
        return "financial_output_parser"

# --- 4. 定义自定义Callback Handler (行为监控) ---
class FinancialSecurityCallbackHandler(BaseCallbackHandler):
    forbidden_tools_pattern: List[str] = ["execute_trade"] # 监测高风险工具
    sensitive_keywords_llm_output: List[str] = ["秘密", "机密", "内部", "私人"]

    def on_tool_start(self, serialized: Dict[str, Any], input_str: str, **kwargs: Any) -> Any:
        tool_name = serialized.get("name")
        if tool_name in self.forbidden_tools_pattern:
            print(f"n[SECURITY ALERT] Agent尝试调用高风险工具: {tool_name} with input: {input_str}")
            # 这里不直接抛异常，而是让 execute_trade 工具内部逻辑处理，或者记录警告
            # 如果需要绝对阻止，可以抛出异常
            # raise ValueError(f"禁止直接调用危险工具 '{tool_name}'。此操作已被拦截。")
        print(f"Tool `{tool_name}` started with input: {input_str}")

    def on_llm_end(self, response: Any, **kwargs: Any) -> Any:
        # 检查LLM在Thought阶段的输出，看是否有敏感词汇
        llm_output_text = response.generations[0][0].text
        for keyword in self.sensitive_keywords_llm_output:
            if keyword in llm_output_text.lower():
                print(f"n[SECURITY WARNING] LLM在思考过程中可能提及敏感词汇: '{keyword}'。原始输出: {llm_output_text[:100]}...")

# --- 5. 初始化LLM ---
llm = ChatOpenAI(model="gpt-4o", temperature=0)

# --- 6. 创建Agent 并注入自定义Parser ---
# 替换默认的ReActOutputParser为我们的FinancialOutputParser
# (与策略三相同，我们假设可以通过替换 agent.output_parser 来实现)
base_financial_agent = create_react_agent(llm, tools, financial_prompt)
base_financial_agent.output_parser = FinancialOutputParser()

# --- 7. 创建Agent执行器 并注入Callback Handler ---
financial_security_handler = FinancialSecurityCallbackHandler()
financial_agent_executor = AgentExecutor(
    agent=base_financial_agent,
    tools=tools,
    verbose=True,
    callbacks=[financial_security_handler],
    handle_parsing_errors=True
)

# --- 8. 实现一个前置守卫 (Input Guardrail) ---
def financial_input_guardrail(user_input: str) -> str:
    """
    Checks user input for non-financial or sensitive topics before passing to Agent.
    """
    non_financial_topics = ["医疗", "法律", "政治", "宗教", "健康", "私人", "个人信息"]
    if any(topic in user_input.lower() for topic in non_financial_topics):
        return "很抱歉，我是一个金融信息助手，无法处理与金融无关的话题。请专注于金融市场或公司信息。"
    return user_input

def guarded_financial_agent_invoke(agent_executor: AgentExecutor, input_data: Dict[str, Any]) -> Dict[str, Any]:
    processed_input = financial_input_guardrail(input_data["input"])
    if processed_input != input_data["input"]:
        return {"output": processed_input}

    return agent_executor.invoke(input_data)

print("--- 场景一：正常查询股票价格 ---")
response1 = guarded_financial_agent_invoke(financial_agent_executor, {"input": "苹果公司 (AAPL) 的股票价格是多少？"})
print(f"Agent Response: {response1['output']}n")

print("--- 场景二：尝试获取投资建议 (预期被Prompt和Output Parser拒绝) ---")
response2 = guarded_financial_agent_invoke(financial_agent_executor, {"input": "现在是买入苹果股票的好时机吗？给我一个投资建议。"})
print(f"Agent Response: {response2['output']}n")

print("--- 场景三：尝试执行未经确认的交易 (预期被工具和Callback处理) ---")
response3 = guarded_financial_agent_invoke(financial_agent_executor, {"input": "帮我买100股GOOG，确认码是XYZ。"})
print(f"Agent Response: {response3['output']}n")

print("--- 场景四：尝试查询非金融话题 (预期被Input Guardrail拦截) ---")
response4 = guarded_financial_agent_invoke(financial_agent_executor, {"input": "我最近身体不舒服，请问有什么治疗方法？"})
print(f"Agent Response: {response4['output']}n")

print("--- 场景五：尝试通过工具获取敏感信息 (预期被工具内置约束拒绝) ---")
response5 = guarded_financial_agent_invoke(financial_agent_executor, {"input": "请帮我查询公司的内部秘密项目。"})
print(f"Agent Response: {response5['output']}n")

案例分析结果：

• 场景一（正常查询）：Agent正常调用get_stock_price工具，返回正确的股价信息。
• 场景二（投资建议）：用户请求投资建议。LLM在生成Final Answer时，可能包含“买入”等词汇。FinancialOutputParser会捕获这些禁止词汇，抛出异常。由于handle_parsing_errors=True，AgentExecutor会尝试让LLM重新生成答案，最终Agent会明确拒绝提供投资建议，并解释原因。
• 场景三（未经确认交易）：用户请求交易。FinancialSecurityCallbackHandler会记录Agent尝试调用高风险工具execute_trade。execute_trade工具内部会检测确认码。由于用户提供的确认码“XYZ”不是"TRADE_CONFIRM_XYZ"，工具会返回交易失败，Agent最终会向用户解释交易失败的原因。
• 场景四（非金融话题）：用户询问医疗问题。financial_input_guardrail会在Agent收到请求之前就将其拦截，并返回一个明确的拒绝消息，防止Agent处理不相关话题。
• 场景五（敏感信息）：用户尝试查询“内部秘密项目”。safe_general_query工具内部会检测到“秘密”等敏感词，并拒绝处理，返回无法处理的提示。

这个案例展示了如何通过多层次的负面约束（系统Prompt、工具内置约束、Output Parser、Callback Handler、Input Guardrail）来共同确保Agent的安全性和合规性。每层约束都在不同阶段发挥作用，形成一个健壮的防御体系。

展望未来：更智能、更安全的 Agent

随着大模型技术和LangChain等框架的不断演进，对Agent行为的精确控制和负面约束将成为AI应用落地的核心竞争力。未来，我们可能会看到以下趋势：

• 更智能的Prompt Engineering工具：自动化地生成和优化Negative Prompts，减少人工干预。
• 标准化的Guardrails库：更成熟、易于集成的Guardrails解决方案，提供开箱即用的安全策略。
• 可解释的约束违反检测：当Agent违反约束时，系统