LangChain Plan-and-Execute 模式深度解析与实战示例

LangChain Plan-and-Execute 模式深度解析与实战示例

掌握结构化任务处理的终极武器，让大语言模型像项目经理一样工作！

1. Plan-and-Execute 核心机制

1.1 两阶段工作流原理

graph LR
    A[用户输入] --> B[规划阶段 Planner]
    B --> C[生成JSON计划]
    C --> D[执行阶段 Executor]
    D --> E[步骤1 工具调用]
    D --> F[步骤2 工具调用]
    D --> G[步骤3 工具调用]
    E --> H[结果整合]
    F --> H
    G --> H

1.2 技术架构组件

2. 完整实战示例：股票分析系统

2.1 场景描述

任务：分析苹果公司(AAPL)股票近期表现并生成投资建议报告
输入："请分析AAPL股票过去一个月的表现，给出投资建议"
预期输出：包含价格趋势、关键指标和投资建议的结构化报告

2.2 系统搭建步骤

步骤1：环境配置

# 安装必要库
!pip install langchain langchain-experimental yfinance

# 导入核心模块
from langchain_experimental.plan_and_execute import PlanAndExecute, load_agent_executor
from langchain_community.tools import YahooFinanceNews, PythonREPLTool
from langchain_openai import ChatOpenAI
from langchain.chains import LLMChain
from langchain.prompts import PromptTemplate

步骤2：创建自定义工具

# 股票数据获取工具
def stock_data_tool(ticker: str, period: str = "1mo"):
    import yfinance as yf
    stock = yf.Ticker(ticker)
    hist = stock.history(period=period)
    return hist.to_json()

# 技术指标计算工具
def calculate_indicators(data_json):
    import pandas as pd
    from ta import add_all_ta_features
    df = pd.read_json(data_json)
    df = add_all_ta_features(df, open="Open", high="High", low="Low", close="Close", volume="Volume")
    return df[['Close', 'trend_sma_fast', 'momentum_rsi']].tail(10).to_dict()

步骤3：配置Planner提示模板

planner_prompt = PromptTemplate.from_template("""
你是一个专业股票分析师，请为以下任务创建执行计划：
任务：{input}

要求：
1. 使用JSON格式输出计划
2. 包含3-5个步骤
3. 每个步骤明确工具和参数

示例：
[
  {{"step":1, "tool":"stock_data_tool", "params":{{"ticker":"AAPL", "period":"1mo"}}}},
  {{"step":2, "tool":"calculate_indicators", "params":{{"data_json":"$1.output"}}}}
]

请生成计划：
""")

步骤4：构建完整Agent

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

# 工具集配置
tools = [
    Tool(name="stock_data", func=stock_data_tool, 
         description="获取股票历史数据，参数：ticker(股票代码), period(时间段)"),
    Tool(name="technical_analysis", func=calculate_indicators,
         description="计算技术指标，参数：data_json(股票数据)"),
    YahooFinanceNews(),
    PythonREPLTool()
]

# 创建Planner链
planner_chain = LLMChain(llm=llm, prompt=planner_prompt)

# 构建Executor
executor = load_agent_executor(llm, tools, verbose=True)

# 整合Plan-and-Execute Agent
agent = PlanAndExecute(
    planner=planner_chain,
    executor=executor,
    verbose=True
)

2.3 执行过程解析

生成的计划示例

[
  {
    "step": 1,
    "tool": "stock_data",
    "params": {"ticker": "AAPL", "period": "1mo"}
  },
  {
    "step": 2,
    "tool": "technical_analysis",
    "params": {"data_json": "$1.output"}
  },
  {
    "step": 3,
    "tool": "yahoo_finance_news",
    "params": {"query": "AAPL 近期新闻"}
  },
  {
    "step": 4,
    "tool": "python_repl",
    "params": {"query": "整合数据生成报告"}
  }
]

执行日志示例

> 进入Plan-and-Execute流程...
> 规划阶段：
  生成4步执行计划
  
> 执行阶段：
  [步骤1] 调用stock_data(AAPL, 1mo)
    输出：获取30天历史数据（日期,开盘价,最高价...）
  
  [步骤2] 调用technical_analysis(输入$1.output)
    输出：{
      "Close": [182.52, 183.86, ...],
      "RSI": [58.3, 61.2, ...]
    }
  
  [步骤3] 调用yahoo_finance_news("AAPL 近期新闻")
    输出：苹果发布新iPad... 分析师上调评级...
  
  [步骤4] 调用python_repl(数据整合)
    输入：整合$1-$3的输出
    输出：
      ## AAPL股票分析报告
      **价格趋势**：过去一月上涨5.2%
      **技术指标**：RSI=61.2（中性）
      **市场动态**：新产品发布推动股价
      **投资建议**：谨慎买入

2.4 结果优化技巧

# 添加记忆缓存提升性能
from langchain.cache import InMemoryCache
langchain.llm_cache = InMemoryCache()

# 设置超时防止卡死
agent = PlanAndExecute(
    ...,
    max_iterations=8,
    early_stopping_method="force"
)

# 添加人工审核节点
def human_approval(step_output):
    if "买入" in step_output and "建议" in step_output:
        return input("确认发送投资建议？(y/n): ")
    return "y"

3. 高级应用场景

3.1 电商价格监控系统

graph TB
    A[监控任务] --> B[1. 爬取竞品价格]
    A --> C[2. 提取特征数据]
    A --> D[3. 价格对比分析]
    A --> E[4. 生成调价建议]

代码片段：

plan = [
    {"step":1, "tool":"web_scraper", "params":{"urls":["amazon/ipad", ...]}},
    {"step":2, "tool":"data_extractor", "params":{"html":"$1.output"}},
    {"step":3, "tool":"pricing_analyzer", "params":{"our_price":599, "competitors":"$2.output"}},
    {"step":4, "tool":"report_generator", "params":{"analysis":"$3.output"}}
]

3.2 学术论文分析流水线

4. 常见问题解决方案

4.1 错误处理模式

graph TD
    A[步骤执行失败] --> B{错误类型}
    B -->|数据错误| C[重试当前步骤]
    B -->|逻辑错误| D[重新生成计划]
    B -->|外部服务异常| E[启用备用工具]

4.2 性能优化策略

1. 并行执行：对无依赖的步骤启用并行

   executor = load_agent_executor(..., parallel=True)

2. 计划缓存：相似任务复用历史计划
3. 步骤压缩：合并相似操作步骤
4. 资源监控：自动跳过高负载工具

4.3 安全防护措施

# 工具权限控制
restricted_tools = {
    "database_writer": ["admin"],
    "payment_api": ["finance_team"]
}

# 输出内容过滤
from langchain.output_parsers import CommaSeparatedListOutputParser
agent.output_parser = CommaSeparatedListOutputParser()

最佳实践：Plan-and-Execute 模式特别适合需要严格工作流的场景。通过将规划与执行分离，您获得了：

1. 更透明的决策过程
2. 更易调试的系统架构
3. 可复用的任务模板

# 快速启动模板
from langchain_experimental.plan_and_execute import PlanAndExecuteAgent

agent = PlanAndExecuteAgent.from_llm_and_tools(
    llm=ChatOpenAI(),
    tools=[stock_tool, analysis_tool],
    planner_prompt=planner_prompt
)
result = agent.run("分析TSLA股票最近三个月表现")