LangChain 0.2.x 多工具 Agent 实战：3步构建支持人工干预的自动化流程

LangChain 0.2.x 多工具 Agent 实战构建支持人工干预的自动化流程在当今快速发展的AI领域LangChain已成为构建智能代理(Agent)的首选框架之一。最新发布的0.2.x版本带来了更强大的多工具集成能力和灵活的控制机制特别是对人工干预(Human-in-the-loop)的支持使得构建生产级AI应用变得更加可靠和安全。本文将深入探讨如何利用这些新特性通过三个核心步骤构建一个既自动化又可控的AI代理系统。1. 理解LangChain Agent的核心架构LangChain Agent本质上是一个由大型语言模型(LLM)驱动的决策系统它能够根据任务需求自主选择和执行工具(Tools)。与传统的脚本化流程不同Agent具备动态决策能力这使得它能够处理更复杂的任务场景。关键组件解析工具(Tools)Agent可调用的功能模块如数据库查询API调用文件操作数学计算代理执行器(AgentExecutor)协调工具调用和LLM决策的中枢系统记忆(Memory)维护对话历史和上下文状态from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.tools import tool # 定义基础工具 tool def search_database(query: str) - str: 在内部数据库中搜索相关信息 # 实际数据库查询逻辑 return f找到3条关于{query}的记录 tool def send_notification(message: str, recipient: str) - str: 向指定收件人发送通知 return f已向{recipient}发送{message}版本0.2.x的重要改进特性0.1.x版本0.2.x版本工具调用基础绑定动态多工具选择人工干预无原生支持内置回调机制执行控制有限细粒度流程管理错误处理简单增强型恢复机制2. 构建多工具集成系统现代业务场景往往需要Agent能够灵活组合多种工具。LangChain 0.2.x通过改进的bind_tools机制使这一过程变得更加直观。2.1 定义工具集首先我们需要准备一组业务相关的工具。以下是一个电商客服Agent可能需要的工具示例from typing import Optional from langchain_core.tools import tool tool def check_order_status(order_id: str) - dict: 查询订单状态 # 模拟订单查询逻辑 return { order_id: order_id, status: 已发货, estimated_delivery: 2024-03-15 } tool def initiate_refund(order_id: str, reason: str) - str: 发起退款流程 return f订单{order_id}的退款已受理原因{reason} tool def escalate_to_human(reason: str) - str: 将问题转接给人工客服 return f已标记需要人工介入{reason}2.2 创建工具调用Agent将工具绑定到LLM并创建代理执行器from langchain_openai import ChatOpenAI from langchain import hub # 初始化LLM llm ChatOpenAI(modelgpt-3.5-turbo-0125, temperature0) # 从Hub获取预置prompt prompt hub.pull(hwchase17/openai-tools-agent) # 组合所有工具 tools [check_order_status, initiate_refund, escalate_to_human] # 创建Agent agent create_tool_calling_agent(llm, tools, prompt) # 创建执行器 agent_executor AgentExecutor( agentagent, toolstools, verboseTrue # 开启详细日志 )2.3 多工具协同示例现在Agent已经能够根据用户请求智能选择工具# 简单查询 response agent_executor.invoke( {input: 我的订单#12345状态如何} ) print(response[output]) # 复杂场景 complex_response agent_executor.invoke( {input: 订单#12345收到时已损坏我需要退款} ) print(complex_response[output])3. 实现人工干预机制在生产环境中完全自动化的Agent可能存在风险。LangChain 0.2.x原生支持人工审批环节关键场景下确保人类监督。3.1 审批回调函数创建一个自定义审批层在敏感操作前暂停执行from langchain_core.messages import AIMessage import json def human_approval(msg: AIMessage) - AIMessage: 人工审批工具调用的函数 if not msg.tool_calls: return msg # 无工具调用时直接通过 # 格式化工具调用信息 tool_strs \n\n.join( json.dumps(tool_call, indent2) for tool_call in msg.tool_calls ) # 构建审批提示 input_msg ( 以下工具调用需要审批\n\n f{tool_strs}\n\n 输入Y或Yes批准其他输入将拒绝。\n 您的决定 ) resp input(input_msg) if resp.lower() not in (yes, y): raise ValueError( f审批被拒绝\n\n{tool_strs} ) return msg3.2 集成审批流程将审批环节加入执行链from langchain_core.runnables import RunnablePassthrough # 构建带审批的执行链 secure_chain ( RunnablePassthrough.assign(inputlambda x: x[input]) | agent | human_approval | agent_executor ) # 测试敏感操作 try: result secure_chain.invoke({ input: 为订单#12345发起退款原因是商品质量问题 }) print(result[output]) except ValueError as e: print(f操作被阻止{e})3.3 选择性审批策略不是所有工具都需要审批我们可以实现更精细的控制def selective_approval(msg: AIMessage) - AIMessage: 只对高风险工具要求审批 if not msg.tool_calls: return msg high_risk_tools [initiate_refund, escalate_to_human] for tool_call in msg.tool_calls: if tool_call[name] in high_risk_tools: print(f高风险操作需要审批{tool_call}) return human_approval(msg) return msg # 低风险工具自动通过 # 更新执行链 selective_chain ( RunnablePassthrough.assign(inputlambda x: x[input]) | agent | selective_approval | agent_executor )4. 高级技巧与最佳实践4.1 工具调用优化当Agent需要同时调用多个工具时可以使用增强的调用策略from operator import itemgetter from typing import Dict, List def call_tools(msg: AIMessage) - List[Dict]: 增强型工具调用处理器 tool_map {tool.name: tool for tool in tools} results [] for tool_call in msg.tool_calls: try: output tool_map[tool_call[name]].invoke(tool_call[args]) results.append({ tool: tool_call[name], input: tool_call[args], output: output }) except Exception as e: results.append({ tool: tool_call[name], error: str(e) }) return results # 构建并行工具调用链 parallel_tool_chain ( agent | call_tools | (lambda x: \n.join( f工具 {res[tool]} 执行结果{res.get(output, res.get(error))} for res in x )) )4.2 状态管理与检查点对于长时间运行的任务状态管理至关重要from langchain_core.runnables import RunnableLambda class AgentState: def __init__(self): self.checkpoint None def save_checkpoint(self, data): self.checkpoint data return 状态已保存 def load_checkpoint(self): return self.checkpoint or 无可用检查点 # 初始化状态管理器 state_manager AgentState() # 添加状态操作工具 tool def save_state(note: str) - str: 保存当前任务状态 return state_manager.save_checkpoint(note) tool def load_state() - str: 加载上次保存的状态 return state_manager.load_checkpoint() # 添加到工具集 tools.extend([save_state, load_state])4.3 监控与日志完善的监控是生产系统的必备条件import logging from datetime import datetime logging.basicConfig( filenameagent_operations.log, levellogging.INFO, format%(asctime)s - %(levelname)s - %(message)s ) def log_operation(func): 工具调用日志装饰器 def wrapper(*args, **kwargs): start_time datetime.now() try: result func(*args, **kwargs) logging.info( f工具 {func.__name__} 执行成功 | f参数: {kwargs} | 耗时: {datetime.now()-start_time} ) return result except Exception as e: logging.error( f工具 {func.__name__} 执行失败 | f错误: {str(e)} ) raise return wrapper # 应用日志装饰器 check_order_status log_operation(check_order_status) initiate_refund log_operation(initiate_refund)5. 实战客户服务Agent完整示例让我们将这些概念整合到一个完整的客户服务Agent实现中from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_openai import ChatOpenAI from langchain import hub from typing import Dict, List import json # 1. 定义工具集 tool def get_order_details(order_id: str) - Dict: 获取订单详细信息 # 模拟数据库查询 return { order_id: order_id, customer: 张三, items: [商品A, 商品B], total: 299.99, status: 已发货 } tool def process_refund(order_id: str, amount: float) - str: 处理退款请求 return f订单 {order_id} 已退款 {amount}元 tool def contact_customer(email: str, message: str) - str: 联系客户 return f已发送消息到 {email}: {message} # 2. 审批机制 def manager_approval(msg: AIMessage) - AIMessage: 退款超过100元需要经理审批 for tool_call in msg.tool_calls or []: if tool_call[name] process_refund: amount tool_call[args].get(amount, 0) if float(amount) 100: print(大额退款需要经理审批:) print(json.dumps(tool_call[args], indent2)) resp input(批准? (Y/N): ) if resp.lower() ! y: raise ValueError(经理拒绝了该退款请求) return msg # 3. 创建Agent tools [get_order_details, process_refund, contact_customer] llm ChatOpenAI(modelgpt-3.5-turbo-0125) prompt hub.pull(hwchase17/openai-tools-agent) customer_service_agent create_tool_calling_agent(llm, tools, prompt) agent_executor AgentExecutor( agentcustomer_service_agent, toolstools, verboseTrue ) # 4. 带审批的执行链 secure_customer_chain ( RunnablePassthrough.assign(inputlambda x: x[input]) | customer_service_agent | manager_approval | agent_executor ) # 5. 测试场景 scenarios [ {input: 我想查询订单#1001的详情}, {input: 订单#1001的商品损坏了我需要退款300元}, {input: 请通知客户#1001退款已处理} ] for scenario in scenarios: print(f\n处理请求: {scenario[input]}) try: result secure_customer_chain.invoke(scenario) print(最终回复:, result[output]) except Exception as e: print(处理失败:, str(e))这个实现展示了如何将多工具集成、人工审批和业务逻辑结合到一个完整的解决方案中。在实际部署时您还需要考虑错误处理增加更健壮的异常捕获和恢复机制性能优化对耗时工具实现异步调用安全审计记录所有工具调用和审批决策用户体验设计更自然的审批交互界面6. 评估与改进构建Agent不是一次性的工作持续的评估和优化至关重要。关键指标监控指标说明目标值工具调用准确率Agent选择正确工具的比例90%人工干预率需要人工审批的操作比例10%平均响应时间从请求到完成的时间5秒任务完成率成功解决用户问题的比例85%改进策略工具优化为模糊工具添加更清晰的描述拆分多功能工具为单一职责工具Prompt工程明确Agent的角色和限制添加示例对话提高一致性审批流程实现分级审批策略添加自动审批规则减少人工干预# 评估示例工具调用准确率测试 test_cases [ (我的订单状态, get_order_details), (我要退款, process_refund), (联系客户, contact_customer) ] correct 0 for query, expected_tool in test_cases: response agent_executor.invoke({input: query}) used_tools [call[name] for call in response.get(tool_calls, [])] if expected_tool in used_tools: correct 1 accuracy correct / len(test_cases) * 100 print(f工具调用准确率: {accuracy:.1f}%)通过持续监控这些指标并迭代优化您的Agent将逐步达到生产级可靠性和效率要求。