# LangChain Plan-and-Execute 模式深度解析与实战示例
> 掌握结构化任务处理的终极武器，让大语言模型像项目经理一样工作！
## 1. Plan-and-Execute 核心机制
### 1.1 两阶段工作流原理
```mermaid
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 技术架构组件
| 组件 | 功能描述 | 实现类 |
|-------------|------------------------------|--------------------------|
| **Planner** | 生成结构化执行计划 | LLM + 提示工程 |
| **Executor**| 按顺序执行计划步骤 | AgentExecutor |
| **检查点** | 保存/恢复执行状态 | langgraph-checkpoint |
| **工具集** | 具体功能实现单元 | 自定义或预构建Tools |
---
## 2. 完整实战示例：股票分析系统
### 2.1 场景描述
**任务**：分析苹果公司(AAPL)股票近期表现并生成投资建议报告
**输入**："请分析AAPL股票过去一个月的表现，给出投资建议"
**预期输出**：包含价格趋势、关键指标和投资建议的结构化报告
### 2.2 系统搭建步骤
#### 步骤1：环境配置
```python
# 安装必要库
!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：创建自定义工具
```python
# 股票数据获取工具
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提示模板
```python
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
```python
# 初始化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 执行过程解析
#### 生成的计划示例
```json
[
{
"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": "整合数据生成报告"}
}
]
```
#### 执行日志示例
```plaintext
> 进入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 结果优化技巧
```python
# 添加记忆缓存提升性能
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 电商价格监控系统
```mermaid
graph TB
A[监控任务] --> B[1. 爬取竞品价格]
A --> C[2. 提取特征数据]
A --> D[3. 价格对比分析]
A --> E[4. 生成调价建议]
```
**代码片段**：
```python
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 学术论文分析流水线
| 步骤 | 工具 | 输入 | 输出 |
|------|------|------|------|
| 1. PDF解析 | pdf_parser | 论文PDF | 结构化文本 |
| 2. 关键信息提取 | info_extractor | 解析文本 | 方法/结果数据 |
| 3. 相似论文检索 | semantic_search | 关键词 | 相关论文 |
| 4. 对比报告生成 | llm_writer | 所有数据 | 分析报告 |
---
## 4. 常见问题解决方案
### 4.1 错误处理模式
```mermaid
graph TD
A[步骤执行失败] --> B{错误类型}
B -->|数据错误| C[重试当前步骤]
B -->|逻辑错误| D[重新生成计划]
B -->|外部服务异常| E[启用备用工具]
```
### 4.2 性能优化策略
1. **并行执行**：对无依赖的步骤启用并行
```python
executor = load_agent_executor(..., parallel=True)
```
2. **计划缓存**：相似任务复用历史计划
3. **步骤压缩**：合并相似操作步骤
4. **资源监控**：自动跳过高负载工具
### 4.3 安全防护措施
```python
# 工具权限控制
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. 可复用的任务模板
>
```python
# 快速启动模板
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股票最近三个月表现")
```