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

<code class="python">Q: 2 + 3 * 4 = ?A:Step 1: 先算乘法 3*4 = 12Step 2: 再算加法 2 + 12 = 14所以答案是 14</code>

<code class="JSON">[  {    "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  }]</code>

<code class="python">User: 帮我写一篇 RAG 入门博客，最后给一个 Python+FAISS 的 Demo。LLM（Planner 输出）:1. 解释什么是 RAG 以及解决什么问题2. 画出一个典型 RAG 流程（构建知识库→检索→过滤→压缩→拼 prompt）3. 给出一个基于 pdfplumber + SentenceTransformer + FAISS 的代码示例4. 写一个总结和延伸阅读</code>

<code class="PYTHON">import jsondef 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</code>

<code class="PYTHON">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 = "</p><p>".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}")        print(f"[Step {step['step']}] 输出：{sub_answer}")        # 4️⃣ 把结果写入 Memory（如果有的话）        if memory is not None:            memory.write(                role="assistant",                content=f"[Step {step['step']}] {sub_answer}"            )</code>

<code class="python">from langchain_openai import ChatOpenAIfrom langchain_community.tools import DuckDuckGoSearchRunfrom 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"])</code>

<code class="python">from langchain_openai import ChatOpenAIfrom langchain.agents import AgentExecutor, create_react_agentfrom langchain import hubfrom langchain_community.tools import DuckDuckGoSearchRunfrom langchain_core.tools import tool# 1. LLMllm = ChatOpenAI(model="gpt-4o-mini", temperature=0)# 2. 定义工具search = DuckDuckGoSearchRun()@tooldef 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，负责 loopagent_executor = AgentExecutor(agent=agent, tools=tools, verbose=True)result = agent_executor.invoke({"input": "帮我查一下北京今天天气，再算一下 1+2*3"})print(result["output"])</code>

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

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