Guardrails与Langchain AgentExecutor集成的最佳实践

概述

在构建基于大语言模型(LLM)的应用时，Guardrails和Langchain是两个非常强大的工具。Guardrails提供了内容验证和安全防护功能，而Langchain则简化了LLM应用的开发流程。本文将详细介绍如何正确地将Guardrails与Langchain的AgentExecutor进行集成，特别是针对流式处理场景。

常见集成问题分析

许多开发者在尝试将Guardrails与Langchain的AgentExecutor集成时会遇到以下典型问题：

1. 类型错误：当尝试在Langchain的RunnableSequence中间环节插入Guard时，会出现TypeError: RunnableSequence._transform() got an unexpected keyword argument 'tools'错误。

2. API缺失错误：在某些集成位置会出现ValueError: API must be provided错误。

3. 验证时机不当：Guard验证在Agent执行完成前就被触发，导致验证无效。

问题根源

这些问题的根本原因在于对Langchain执行流程的理解不足。Langchain的AgentExecutor工作流程包含多个阶段：

1. 工具绑定阶段：将工具定义转换为OpenAI兼容格式
2. 提示工程阶段：构建完整的提示信息
3. 模型调用阶段：实际调用LLM
4. 输出解析阶段：解析模型返回结果
5. 工具执行阶段：执行工具调用
6. 结果整合阶段：整合所有步骤结果

如果在不恰当的阶段插入Guard验证，就会导致上述错误。

正确集成方案

经过深入分析，正确的集成方式是将Guard放在整个AgentExecutor之后，而不是中间环节。这种架构有以下优势：

1. 确保所有处理流程已完成
2. 能够验证最终输出结果
3. 避免干扰Langchain内部执行流程

以下是正确的代码结构示例：

# 创建Agent组件链
agent = (
    RunnablePassthrough.assign(
        agent_scratchpad=lambda x: format_to_openai_tool_messages(
            x["intermediate_steps"]
        )
    )
    | prompt
    | llm_with_tools
    | OpenAIToolsAgentOutputParser()
)

# 创建执行器
agent_executor = AgentExecutor(agent=agent, tools=tools)

# 在最后添加Guard验证
chain = agent_executor | guard

实际应用示例

假设我们需要开发一个文档检索系统，要求输出中必须包含特定关键词"apricot"，我们可以这样实现：

# 定义文档检索工具
@tool
def get_retriever_docs(query: str) -> list[Document]:
    """返回检索到的文档列表"""
    return [Document(page_content="测试文档包含秘密代码'blue-green-apricot-brownie-cake-mousepad'")]

# 设置Guard验证规则
topic = "apricot"
guard = Guard().use(RegexMatch(topic, match_type="search", on_fail="filter"))

# 构建完整处理链
chain = agent_executor | guard

# 执行查询
query = "调用get_retriever_docs并告诉我文档中的秘密"
result = chain.invoke({"input":query})

性能考量

对于流式处理场景(streaming=True)，需要注意：

1. Guard验证会等待完整响应后才执行
2. 验证过程会增加少量延迟
3. 对于实时性要求高的场景，可以考虑异步验证

总结

将Guardrails与Langchain集成时，关键在于理解Langchain的执行流程和选择合适的集成点。通过在AgentExecutor之后添加Guard验证，可以确保：

1. 所有工具调用和结果整合已完成
2. 验证的是最终输出结果
3. 不干扰Langchain内部执行机制

这种架构既保持了Langchain的灵活性，又通过Guardrails增加了内容安全性和可靠性。