langchain基础教程(3)---langchain一些高级用法
护栏机制(Guardrails)
Guardrails 是一种机制，可以让你验证 LLM（大语言模型）的输入和输出，确保其符合预期。
常见使用场景:
验证用户输入是否不在允许范围内
确保输入在调用 LLM 前满足特定条件（例如防御 提示注入攻击）
确保输出格式正确（例如是符合正确模式的 JSON 文档）
确保 LLM 输出符合业务规则和约束（例如，如果这是 X 公司的聊天机器人，回答中不能包含对竞争对手 Y 的引用）
检测幻觉（hallucinations）
使用时注意事项
理想情况下，Guardrail 的实现应遵循单一职责原则，即每个 Guardrail 类只验证一件事情。然后将多个 Guardrail 串联起来，以防护多个方面。Guardrail 链中的顺序很重要。第一个失败的 Guardrail 会触发整体失败。应确保最容易捕获错误的 Guardrail 排在链的前面，而那些仅在极少情况下才会失败的 Guardrail 放在链的后面。另外请记住，Guardrail 本身可以调用其他服务，甚至触发其他 LLM 交互。如果这些 Guardrail 执行有延迟或会带来额外的成本
护栏机制是用中间件实现的,langchain提供了内置护栏,如PII检测,人工干预
PII检测用于检测和处理对话中的个人身份信息 (PII)。此中间件可以检测常见的 PII 类型，如电子邮件、信用卡、IP 地址等。
人工干预: 用于在执行敏感操作之前要求人工批准。这是高风险决策最有效的防护栏之一。常用于金融交易和转账、删除或修改生产数据、向外部方发送通信以及任何具有重大业务影响的操作等代码案例:
from langchain.agents import create_agent
from langchain.agents.middleware import PIIMiddleware
agent = create_agent(model="gpt-4o",tools=[customer_service_tool, email_tool],middleware=[PIIMiddleware("email",strategy="redact",apply_to_input=True,)],
)result = agent.invoke({"messages": [{"role": "user", "content": "My email is john.doe@example.com and card is 4532-1234-5678-9010"}]
})
PIIMiddleware配置项:
pii_type: 要检测的 PII 类型。可以是内置类型（email、credit_card、ip、mac_address、url）或自定义类型名称
strategy: 如何处理检测项, 默认:"redact""block" - 检测到时抛出异常"redact" - 替换为 [REDACTED_TYPE]"mask" - 部分遮盖（例如，****-****-****-1234）"hash" - 替换为确定性哈希
detector: 自定义正则表达式或检测函数,默认使用PII内置检测器
apply_to_input: 默认True, 调用模型前检查,检查模型输入
apply_to_output: 默认False ,检查模型输出消息
apply_to_tool_results: 默认False,检查工具输出结果
Human-in-the-loop(人工干预)
在执行敏感操作时,需要人工干预
代码案例:
from langchain.agents import create_agent
from langchain.agents.middleware import HumanInTheLoopMiddleware
from langgraph.checkpoint.memory import InMemorySaver
from langgraph.types import Command
from langchain_openai import ChatOpenAIdeepseek_api_base = "https://api.deepseek.com"
deepseek_api_key = "sk-************************"
deepseek_model = "deepseek-chat"# 设置DeepSeek API Key和基础地址
llm = ChatOpenAI(model=deepseek_model,openai_api_key=deepseek_api_key,openai_api_base=deepseek_api_base, # 关键配置temperature=0.7,
)# 1. 定义示例敏感工具（模拟发送邮件、删除数据库）def send_email_func(content) -> str:"""向指定对象发送邮件，属于敏感操作需人工批准"""return f"邮件已发送：{content}"def get_weather(city: str) -> str:"""获取指定城市的天气"""return f"{city} 天气总是晴朗！"def get_company_info() -> str:"""获取用户公司信息，属于敏感操作需人工批准"""return f"公司信息：钱多事少离家近有限责任公司"def delete_database_func(table):"""删除数据库表，属于高危操作需人工批准"""return f"数据库表 {table} 已删除"def search_func(content) -> str:"""普通搜索操作，无需人工批准"""return f"搜索结果：{content}"# 2. 初始化 Agent，配置 HumanInTheLoop 中间件
agent = create_agent(model=llm,tools=[get_company_info, delete_database_func, get_weather],middleware=[HumanInTheLoopMiddleware(interrupt_on={"get_company_info": True, # 触发该工具时中断，等待人工批准"delete_database_func": True, # 触发该工具时中断"get_weather": False, # 普通工具不中断})],checkpointer=InMemorySaver(), # 持久化中断状态，确保恢复时上下文一致debug=True,
)# 配置会话ID（中断和恢复必须使用同一个 thread_id）
config = {"configurable": {"thread_id": "session_001"}}# 用户会话
user_request = {"messages": [{"role": "user","content": "查一下我公司信息",}]
}first_result = agent.invoke(user_request, config=config)
print("中断提示：", first_result) # 输出类似"需人工批准发送邮件操作，请确认"
user_confirm = input("是否允许执行？(y/n)：")
if user_confirm.lower() == "y":# 批准command = Command(resume={"decisions": [{"type": "approve"}]} # 关键：type指定"approve"表示允许执行)
else:# 驳回command = Command(resume={"decisions": [{"type": "reject", # 关键：type指定"reject"表示拒绝执行"reason": "拒绝操作", # 可选：添加驳回原因}]})final_result = agent.invoke(command, config=config)
print("执行结果：", final_result)运行时:
LangChain 的 create_agent 实际上是在 LangGraph 的运行时环境下运行的。
LangGraph 暴露了一个 Runtime 对象，其中包含以下信息：
Context (上下文): 静态信息，例如用户 ID、数据库连接，或代理调用所需的其他依赖项。
Store (存储): 一个 BaseStore 实例，用于长期记忆。
Stream writer (流写入器): 一个用于通过 "custom" 流模式进行信息流式传输的对象。
你可以在工具 (tools) 和中间件 (middleware) 中访问运行时信息。
多agent系统:
方案1: 工具调用, 主智能体将其他智能体当工具调用
方案2: 交接 (Handoffs) 当前智能体决定将控制权转移给另一个智能体。活动的智能体发生变化，用户可以继续直接与新的智能体交互。
选择方式:
如果子智能体不需要与用户对话,可以使用工具调用
如果需要子智能体能够与用户对话,则使用交接模式
工具调用示例:
subagent1 = create_agent(model="...", tools=[...])@tool("subagent1_name",description="subagent1_description"
)
def call_subagent1(query: str):result = subagent1.invoke({"messages": [{"role": "user", "content": query}]})return result["messages"][-1].contentagent = create_agent(model="...", tools=[call_subagent1])
在此模式中：
当主智能体决定任务与子智能体的描述匹配时，它会调用 call_subagent1。
子智能体独立运行并返回其结果。
主智能体接收结果并继续编排
交接 (Handoffs)模式: 在langchain中并没有支持,
可以使用LangGraph, AutoGen或者OpenAI的Agents
长期记忆
LangChain 智能体使用 LangGraph 持久化来实现长期记忆。
内存存储 (Memory storage): LangGraph 将长期记忆作为 JSON 文档存储在存储 (store) 中
然后在工具中进行读写;