【Agentic专题】 Planning专题学习与面经

Q: 2 + 3 * 4 = ?
A:
Step 1: 先算乘法 3*4 = 12
Step 2: 再算加法 2 + 12 = 14
所以答案是 14

[
  {
    "step": 1,
    "description": "收集 RAG 基础定义与优缺点",
    "need_retrieval": true
  },
  {
    "step": 2,
    "description": "整理 RAG 全流程说明和流程图",
    "need_retrieval": false
  },
  {
    "step": 3,
    "description": "实现一个基于 PDF + FAISS 的 RAG 示例代码",
    "need_retrieval": false
  },
  {
    "step": 4,
    "description": "撰写前言、总结，并补充延伸阅读",
    "need_retrieval": true
  }
]

User: 帮我写一篇 RAG 入门博客，最后给一个 Python+FAISS 的 Demo。

LLM（Planner 输出）:
1. 解释什么是 RAG 以及解决什么问题
2. 画出一个典型 RAG 流程（构建知识库→检索→过滤→压缩→拼 prompt）
3. 给出一个基于 pdfplumber + SentenceTransformer + FAISS 的代码示例
4. 写一个总结和延伸阅读

import json

def plan_task(user_goal: str, llm) -> list:
    """
    调用 LLM，把复杂目标拆成 3~7 步的结构化计划。
    """
    prompt = f"""
                你是一个任务规划助手。现在有一个复杂目标，请你把它拆分成 3~7 个有序子任务。

                要求：
                1. 每个子任务尽量原子化，可以独立完成。
                2. 标明该步是否需要调用外部知识库（RAG）（例如需要查 PDF / 文档 / 网页）。
                3. 只输出 JSON 数组，不要多余解释。

                JSON 元素字段：
                - "step": 整数，从 1 开始
                - "description": 这一步要做什么
                - "need_retrieval": true 或 false

                用户目标：{user_goal}
            """
    # 假设 llm(prompt) 直接返回一个字符串形式的 JSON
    resp = llm(prompt)
    plan = json.loads(resp)
    return plan

def execute_plan(plan: list, llm, retriever=None, memory=None):
    """
    逐步执行规划好的子任务。
    retriever: 检索函数，例如 RAG 的 retrieve_docs(desc) -> [doc1, doc2, ...]
    memory: 一个简单的 Memory 对象，支持 get_recent() / write(...)
    """
    for step in plan:
        desc = step["description"]
        need_ret = step.get("need_retrieval", False)

        # 1️⃣ 按需要触发 RAG 检索
        context = ""
        if need_ret and retriever is not None:
            docs = retriever(desc)
            # 简单拼一下，也可以在这里先做摘要
            context = "\n\n".join(docs)

        # 2️⃣ 从 Memory 里拿最近若干条历史记录（可选）
        history = ""
        if memory is not None:
            history = memory.get_recent()

        # 3️⃣ 让 LLM 专注完成“当前这一步”
        sub_prompt = f"""
                        你正在执行一个多步任务中的第 {step["step"]} 步。

                        当前子任务描述：
                        {desc}

                        可用资料（可能为空）：
                        {context or "【无外部资料】"}

                        历史记录（可能为空）：
                        {history or "【无历史记录】"}

                        请只完成当前这一步，不要提前执行后续步骤。
                       """
        sub_answer = llm(sub_prompt)
        print(f"[Step {step['step']}] 子任务说明：{desc}\n")
        print(f"[Step {step['step']}] 输出：{sub_answer}\n")

        # 4️⃣ 把结果写入 Memory（如果有的话）
        if memory is not None:
            memory.write(
                role="assistant",
                content=f"[Step {step['step']}] {sub_answer}"
            )

from langchain_openai import ChatOpenAI
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_experimental.plan_and_execute import (
    PlanAndExecute,
    load_chat_planner,
    load_agent_executor,
)

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
tools = [DuckDuckGoSearchRun()]

planner = load_chat_planner(llm) 
executor = load_agent_executor(llm, tools, verbose=True)

agent = PlanAndExecute(planner=planner, executor=executor, verbose=True)

res = agent.invoke({"input": "帮我调研三篇关于 RAG+RL 的论文并输出一个中文总结"})
print(res["output"])

from langchain_openai import ChatOpenAI
from langchain.agents import AgentExecutor, create_react_agent
from langchain import hub
from langchain_community.tools import DuckDuckGoSearchRun
from langchain_core.tools import tool

# 1. LLM
llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)

# 2. 定义工具
search = DuckDuckGoSearchRun()

@tool
def calculator(expr: str) -> str:
    """计算一个数学表达式，例如 '1+2*3'。"""
    import math
    try:
        return str(eval(expr))
    except Exception as e:
        return f"error: {e}"

tools = [search, calculator]

# 3. 用官方 hub 上的 ReAct prompt 模板
prompt = hub.pull("hwchase17/react") 

# 4. 创建 ReAct Agent（底层是一个 Runnable 序列）
agent = create_react_agent(llm, tools, prompt)

# 5. 外面再包一层 AgentExecutor，负责 loop
agent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)

result = agent_executor.invoke({"input": "帮我查一下北京今天天气，再算一下 1+2*3"})
print(result["output"])

def validate_plan(plan):
    for step in plan:
        if step["tool"] not in ALLOWED_TOOLS:
            return False, f"非法工具: {step['tool']}"
    return True, ""

每一步需要做到：
- 单步是可执行的：一个子 Agent 可以在若干次工具调用内完成。
- 避免过于抽象的表达，如“完成所有实验并写完论文”。