麻瓜教程

Appearance

计划与执行 ​

计划与执行是 LangChain 中的高级概念，它允许代理在执行任务之前先制定计划，然后按照计划执行任务。在 LangChain 1.2 中，可以使用 LCEL 语法更灵活地实现计划与执行工作流。

安装依赖 ​

在开始使用计划与执行之前，需要安装以下依赖：

bash
pip install langchain langchain-openai python-dotenv langchain-community google-search-results

什么是计划与执行？ ​

计划与执行是一种工作流程，其中代理首先分析任务，制定一个详细的计划，然后按照计划执行任务。这种方法可以帮助代理处理更复杂的任务，提高任务执行的效率和准确性。

计划与执行的工作流程通常包括：

  1. 分析任务，理解用户的需求
  2. 制定详细的计划，包括需要执行的步骤
  3. 按照计划执行任务，使用工具完成各个步骤
  4. 监控执行过程，根据需要调整计划
  5. 向用户提供最终结果

计划与执行的优势 ​

计划与执行方法相比直接执行任务有以下优势：

  1. 更好的任务分解：将复杂任务分解为简单的步骤
  2. 更高的执行效率：避免不必要的工具调用和重复工作
  3. 更好的错误处理：在执行前可以识别潜在的问题
  4. 更透明的执行过程：用户可以看到代理的思考过程
  5. 更好的结果质量：通过系统的规划和执行，提高结果的质量

实现计划与执行 ​

使用 Plan-and-Execute Agent ​

LangChain 提供了 PlanAndExecute 代理，它可以先制定计划，然后按照计划执行任务。

初始化 PlanAndExecute Agent ​

python
from langchain_openai import OpenAI
from langchain.agents import AgentType, initialize_agent, load_tools

# 初始化语言模型
llm = OpenAI(api_key="your-api-key")

# 加载工具
tools = load_tools(["serpapi", "llm-math"], llm=llm)

# 初始化 PlanAndExecute 代理
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.PLAN_AND_EXECUTE,
    verbose=True
)

使用 PlanAndExecute Agent ​

python
# 运行代理
response = agent.run("What is the current temperature in New York? What about Los Angeles? Compare the two temperatures.")
print(response)

自定义计划与执行 ​

在 LangChain 1.2 中，可以使用 LCEL 语法创建自定义的计划与执行系统。

示例：创建自定义计划与执行系统 ​

python
from langchain_openai import OpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain.tools import BaseTool
from typing import Optional, Type
from pydantic import BaseModel, Field

# 初始化语言模型
llm = OpenAI(api_key="your-api-key")
output_parser = StrOutputParser()

# 创建计划链
plan_template = """You are a helpful assistant. Given a task, create a detailed plan to complete it.

Task: {task}

Plan:"""
plan_prompt = PromptTemplate(template=plan_template, input_variables=["task"])
plan_chain = plan_prompt | llm | output_parser

# 创建执行链
execute_template = """You are a helpful assistant. Given a plan and the current state, execute the next step of the plan.

Plan: {plan}

Current state: {state}

Next step:"""
execute_prompt = PromptTemplate(template=execute_template, input_variables=["plan", "state"])
execute_chain = execute_prompt | llm | output_parser

# 定义工具输入模式
class SearchInput(BaseModel):
    query: str = Field(description="搜索查询")

# 创建搜索工具
class SearchTool(BaseTool):
    name = "search"
    description = "用于搜索网络信息"
    args_schema: Type[BaseModel] = SearchInput

    def _run(self, query: str) -> str:
        # 模拟搜索 API 调用
        return f"搜索结果: {query} 的相关信息"

# 初始化工具
search_tool = SearchTool()

# 执行计划与执行流程
def plan_and_execute(task):
    # 制定计划
    plan = plan_chain.invoke({"task": task})
    print("Plan:", plan)

    # 初始化状态
    state = "开始执行任务"

    # 执行计划
    steps = plan.split("\n")
    for i, step in enumerate(steps):
        if step.strip():
            print(f"\nExecuting step {i+1}: {step.strip()}")
            # 执行步骤
            next_step = execute_chain.invoke({"plan": plan, "state": state})
            print("Next step:", next_step)

            # 使用工具（如果需要）
            if "search" in next_step.lower():
                query = next_step.split("search for")[-1].strip()
                result = search_tool.run(query)
                print("Tool result:", result)
                state += f"\n执行步骤 {i+1}: {step.strip()}\n工具结果: {result}"
            else:
                state += f"\n执行步骤 {i+1}: {step.strip()}\n执行结果: {next_step}"

    # 生成最终答案
    final_template = """You are a helpful assistant. Given a task, a plan, and the execution state, provide a final answer.

Task: {task}

Plan: {plan}

Execution state: {state}

Final answer:"""
    final_prompt = PromptTemplate(template=final_template, input_variables=["task", "plan", "state"])
    final_chain = final_prompt | llm | output_parser

    final_answer = final_chain.invoke({"task": task, "plan": plan, "state": state})
    print("\nFinal answer:", final_answer)

    return final_answer

# 运行计划与执行流程
result = plan_and_execute("What is the capital of France? What is the population of that city?")
print(result)

使用 LCEL 构建计划与执行链 ​

在 LangChain 1.2 中，可以使用 LCEL 语法更优雅地构建计划与执行链。

示例：使用 LCEL 构建计划与执行 ​

python
from langchain_openai import OpenAI
from langchain_core.prompts import PromptTemplate
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnableLambda, RunnablePassthrough

# 初始化语言模型
llm = OpenAI(api_key="your-api-key")
output_parser = StrOutputParser()

# 创建计划链
plan_prompt = PromptTemplate(
    template="""You are a helpful assistant. Given a task, create a detailed plan to complete it.

Task: {task}

Plan:""",
    input_variables=["task"]
)
plan_chain = plan_prompt | llm | output_parser

# 创建执行链
execute_prompt = PromptTemplate(
    template="""You are a helpful assistant. Given a plan and the current state, execute the next step of the plan.

Plan: {plan}

Current state: {state}

Next step:""",
    input_variables=["plan", "state"]
)
execute_chain = execute_prompt | llm | output_parser

# 创建最终答案链
final_prompt = PromptTemplate(
    template="""You are a helpful assistant. Given a task, a plan, and the execution state, provide a final answer.

Task: {task}

Plan: {plan}

Execution state: {state}

Final answer:""",
    input_variables=["task", "plan", "state"]
)
final_chain = final_prompt | llm | output_parser

# 定义执行步骤的函数
def execute_steps(input_data):
    task = input_data["task"]
    plan = input_data["plan"]

    state = "开始执行任务"
    steps = plan.split("\n")

    for i, step in enumerate(steps):
        if step.strip():
            next_step = execute_chain.invoke({"plan": plan, "state": state})
            state += f"\n执行步骤 {i+1}: {step.strip()}\n执行结果: {next_step}"

    return {"task": task, "plan": plan, "state": state}

# 创建完整的计划与执行链
plan_and_execute_chain = (
    {"task": RunnablePassthrough()}
    | RunnableLambda(lambda x: {"task": x, "plan": plan_chain.invoke(x)})
    | RunnableLambda(execute_steps)
    | final_chain
)

# 使用链
result = plan_and_execute_chain.invoke("What is the capital of France?")
print(result)

计划与执行的最佳实践 ​

1. 明确任务目标 ​

在使用计划与执行方法之前，确保任务目标明确，便于代理制定合理的计划。

2. 选择合适的工具 ​

根据任务的性质，选择合适的工具，确保代理能够完成各个步骤。

3. 监控执行过程 ​

在执行过程中，监控代理的执行情况，确保计划的执行符合预期。

4. 调整计划 ​

根据执行过程中的情况，及时调整计划，确保任务能够顺利完成。

5. 评估结果 ​

任务完成后，评估结果的质量，总结经验教训，提高未来任务的执行效率。

示例：使用计划与执行处理复杂任务 ​

示例：研究一个主题 ​

python
from langchain_openai import OpenAI
from langchain.agents import AgentType, initialize_agent, load_tools

# 初始化语言模型
llm = OpenAI(api_key="your-api-key")

# 加载工具
tools = load_tools(["serpapi", "llm-math"], llm=llm)

# 初始化 PlanAndExecute 代理
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.PLAN_AND_EXECUTE,
    verbose=True
)

# 运行代理
response = agent.run("研究 LangChain 的最新发展，包括新特性、应用案例和未来趋势。")
print(response)

示例：解决一个问题 ​

python
from langchain_openai import OpenAI
from langchain.agents import AgentType, initialize_agent, load_tools

# 初始化语言模型
llm = OpenAI(api_key="your-api-key")

# 加载工具
tools = load_tools(["serpapi", "llm-math", "python_repl"], llm=llm)

# 初始化 PlanAndExecute 代理
agent = initialize_agent(
    tools=tools,
    llm=llm,
    agent=AgentType.PLAN_AND_EXECUTE,
    verbose=True
)

# 运行代理
response = agent.run("分析过去 5 年全球温度变化趋势，并预测未来 5 年的变化。")
print(response)

总结 ​

计划与执行是 LangChain 中的高级概念，它允许代理在执行任务之前先制定计划，然后按照计划执行任务。在 LangChain 1.2 中，可以使用 LCEL 语法更灵活地实现计划与执行工作流。

通过本文的介绍，您应该已经了解了如何使用内置的 PlanAndExecute 代理，以及如何使用 LCEL 语法创建自定义的计划与执行系统。在实际应用中，您可以根据具体需求选择合适的方法，构建功能强大的 LLM 应用。

Copyright © 2025-2026 京公网安备 11010102000001号 京ICP备2024078128号-1