掌握工业级智能体开发核心技术，错误率降低70%，复杂任务处理能力提升5倍。
核心选择标准：
工具调用次数 ≤ 3 → ReAct
存在子任务依赖 → Plan-and-Execute
需要并行执行 → Plan-and-Execute
from langchain.agents import AgentExecutor, Tool, initialize_agent
from langchain_core.prompts import PromptTemplate
# 1. 定义工具集
tools = [
Tool(
name="Search",
func=search_api,
description="通用搜索引擎"
),
Tool(
name="Calculator",
func=math_calculator,
description="数学计算器"
)
]
# 2. 创建ReAct提示模板
react_prompt = PromptTemplate.from_template("""
你是一个智能助手，请严格按以下格式响应：
Thought: 思考当前问题和可用工具
Action: 工具名称（必须是{tool_names}之一）
Action Input: 工具输入参数
Observation: 工具返回结果
...（重复直到得出结论）
Final Answer: 最终答案
问题：{input}
""")
# 3. 初始化ReAct智能体
agent = initialize_agent(
tools,
llm,
agent="react-docstore",
prompt=react_prompt,
max_iterations=5 # 防止无限循环
)
# 4. 执行任务
response = agent.invoke({
"input": "特斯拉当前股价的3倍是多少？"
})
print(response["output"])执行过程可视化：
Thought: 需要先获取特斯拉当前股价 Action: Search Action Input: 特斯拉当前股价 Observation: $245.78 Thought: 需要计算3倍股价 Action: Calculator Action Input: 245.78 * 3 Observation: 737.34 Final Answer: 特斯拉当前股价的3倍是$737.34
2.2 企业级优化技巧
def safe_tool_call(tool_name, input):
try:
return tools[tool_name](input)
except Exception as e:
return f"工具调用失败：{str(e)}"
# 在Agent中重写工具调用逻辑
agent.tool_call = safe_tool_call超时控制
import signal
class TimeoutException(Exception):
pass
def handler(signum, frame):
raise TimeoutException("执行超时")
signal.signal(signal.SIGALRM, handler)
signal.alarm(10) # 10秒超时
try:
agent.invoke(input)
except TimeoutException:
return "请求处理超时，请简化您的问题"
from langchain_experimental.plan_and_execute import ( PlanAndExecute, load_agent_executor, load_chat_planner ) from langchain.chains import LLMChain # 1. 创建任务规划器 planner = load_chat_planner( llm, system_prompt="你是一个高级任务规划专家，请将复杂任务分解为可执行的子任务步骤" ) # 2. 创建任务执行器 executor = load_agent_executor( llm, tools, verbose=True ) # 3. 构建Plan-and-Execute智能体 agent = PlanAndExecute( planner=planner, executor=executor, max_iterations=3 # 最大规划迭代次数 ) # 4. 执行复杂任务 response = agent.run( "规划一个上海3日游：需包含外滩、迪士尼、豫园，预算5000元，生成详细行程表" )
任务分解示例：
{
"plan": [
{"id": 1, "task": "查询上海迪士尼门票价格", "dependencies": []},
{"id": 2, "task": "查询外滩附近酒店价格", "dependencies": []},
{"id": 3, "task": "计算交通费用", "dependencies": [1,2]},
{"id": 4, "task": "生成每日行程安排", "dependencies": [1,2,3]},
{"id": 5, "task": "制作预算表", "dependencies": [4]}
]
}3.3 高级特性实现
并行执行引擎
from concurrent.futures import ThreadPoolExecutor
def execute_parallel(tasks):
with ThreadPoolExecutor(max_workers=5) as executor:
future_to_task = {executor.submit(run_task, task): task for task in tasks}
results = {}
for future in concurrent.futures.as_completed(future_to_task):
task = future_to_task[future]
results[task["id"]] = future.result()
return results动态重规划机制
def dynamic_replan(current_plan, failed_task):
# 1. 移除依赖失败任务的所有后续任务
new_plan = [t for t in current_plan if failed_task["id"] not in t.get("dependencies", [])]
# 2. 添加替代方案
new_task = create_alternative_task(failed_task)
new_plan.append(new_task)
# 3. 重新计算依赖关系
return resolve_dependencies(new_plan)
pie title 测试用例分布 “简单查询” ： 45 “多步骤操作” ： 30 “异常处理” ： 25
测试环境：
1000个真实客服对话样本
Azure D13 v2虚拟机（8核/56GB内存）
GPT-4 Turbo模型
典型场景对比：
用户请求： “我的订单#12345想退货，但已经超过7天了怎么办？” ReAct处理： 1. 查询订单状态 → 已签收10天 2. 直接回复：“超过退货期限无法处理” Plan-and-Execute处理： 1. 查询订单详情 2. 检查退货政策例外条款 3. 查找客户历史订单记录 4. 生成替代方案： “虽然超过7天，但您是VIP客户，可申请特殊退货， 我们将安排快递上门取件，是否接受？”
class HybridAgent: def __init__(self): self.react_agent = create_react_agent() self.plan_execute_agent = create_plan_execute_agent() self.classifier = load_task_classifier() # 任务分类模型 def route(self, query): complexity_score = self.classifier.predict(query) if complexity_score < 0.7: return self.react_agent else: return self.plan_execute_agent def run(self, query): agent = self.route(query) return agent.invoke(query)
ReAct循环失控
症状：智能体陷入无限循环
解决方案：
# 添加循环检测
MAX_ITERATIONS = 10
iteration_count = 0
while not done:
iteration_count += 1
if iteration_count > MAX_ITERATIONS:
raise AgentLoopError("超过最大迭代次数")Plan-and-Execute规划偏差
症状：生成不可执行的计划
解决方案：
# 添加计划验证层
def validate_plan(plan):
for task in plan:
if not has_tool(task["action"]):
raise InvalidTaskError(f"无可用工具: {task['action']}")
if missing_params(task):
raise MissingParamsError(f"缺少参数: {task['action']}")工具依赖死锁
症状：任务A依赖B，B又依赖A
解决方案：
# 依赖环检测算法 def detect_cycle(dependencies): graph = build_graph(dependencies) try: topological_sort(graph) # 若无环可排序 except CycleError: return True return False
初级阶段：
掌握ReAct基础实现
构建带3个工具的客服机器人
进阶阶段：
实现Plan-and-Execute行程规划器
添加并行执行优化
企业实战：
80%的任务用ReAct解决，20%复杂任务用Plan-and-Execute
始终添加防护机制防止失控
复杂计划需人工审核后再执行
遵循此技术路线，您将能设计出符合企业需求的智能体系统，建议去聚客AI学院了解系统学习的技术框架和更多AI大模型应用开发学习视频资料。
