AI/ML 详解：Agent 架构

就像写一个带 while 循环的脚本——每一轮 LLM 先"想一下要做什么"（Thought），再"调用一个工具"（Action），看到工具返回结果（Observation）后继续想下一步，直到能给出最终答案。

纯 LLM 只会"一次输入一次输出"——你问什么它直接答什么，没有访问外部工具的能力，也不能根据中间结果调整策略。但很多任务（"帮我查一下苹果今天的股价并算出和年初相比涨幅"）需要：先调 API、再做计算、可能还要再查更多数据。ReAct 把"推理"和"行动"交错起来，让 LLM 像一个能看到中间结果的程序员，边查边想边算。它是几乎所有现代 Agent 框架（LangChain Agent、LangGraph、OpenAI function calling）的鼻祖。

核心是一个 Thought → Action → Observation 的循环，全部通过 prompt 让 LLM 自己产出 Thought 和 Action：

每一轮 LLM 看到完整的历史（之前所有的 Thought / Action / Observation 都拼回 prompt），决定下一步。停止条件：LLM 自己说 "Final Answer: ..."，或者步数超过上限。

from langchain.agents import create_react_agent, AgentExecutor
from langchain.tools import tool
from langchain_openai import ChatOpenAI

# 1) 定义工具——和写一个普通 Python 函数没差别
@tool
def get_stock_price(symbol: str) -> float:
    """查询某只股票的当前价格"""
    return 188.42  # 简化：真实场景调 yfinance/Bloomberg

@tool
def calculator(expression: str) -> float:
    """计算数学表达式"""
    return eval(expression)

# 2) 组装 Agent——LangChain 会自动注入 ReAct prompt 模板
agent = create_react_agent(
    llm=ChatOpenAI(model="gpt-4o"),
    tools=[get_stock_price, calculator],
    prompt=hub.pull("hwchase17/react")  # 经典 ReAct prompt
)
executor = AgentExecutor(agent=agent, tools=[get_stock_price, calculator])

# 3) 运行——你只需要给目标，循环逻辑全部由框架接管
result = executor.invoke({"input": "苹果今天涨了多少？年初是 165"})

• 原论文 ReAct: Synergizing Reasoning and Acting — 配可视化 demo
• Lilian Weng 的 LLM Powered Autonomous Agents — Agent 全景图

就像项目经理 + 执行小弟的分工模式——先用一个 LLM 调用"规划出完整步骤列表"（像写一个 TODO list），再用另一个 LLM 调用"按列表逐条执行"。把"想清楚"和"动手做"拆成两个阶段，避免 ReAct 走两步就忘了大方向。

ReAct 的核心痛点是"短视"——它一次只想一步，所以遇到 10 步以上的复杂任务（"帮我策划下周的部门季度会"），中途容易迷路、做无用功、重复调用同一个工具。Plan-and-Execute 强制 LLM 先把全局任务分解成有序子任务，再逐条做，每条做完更新计划。这种"先做规划再动手"的范式来自经典 AI 规划（STRIPS、PDDL），结合 LLM 后变得通用且无需手写 domain knowledge。

关键差异 vs ReAct：ReAct 是"每一步独立决策"，Plan-and-Execute 是"有显式的 plan 状态在驱动"。Plan 是可以被持久化、审计、人工干预的——这对生产环境特别重要。

from langgraph.prebuilt import create_react_agent
from langgraph.graph import StateGraph
from langchain_openai import ChatOpenAI

planner = ChatOpenAI(model="gpt-4o")
executor_agent = create_react_agent(llm, tools=[search, calc, email])

def plan(state):
    # Planner 一次性产出完整步骤列表
    plan = planner.invoke(f"把目标拆成有序步骤：{state['goal']}")
    return {"plan": plan.steps, "done": []}

def execute(state):
    step = state["plan"][0]
    result = executor_agent.invoke({"input": step})
    return {"plan": state["plan"][1:],
            "done": state["done"] + [(step, result)]}

def replan(state):
    # 看历史结果决定是否修改剩余计划
    return planner.invoke(f"已完成 {state['done']}，剩余 {state['plan']}，要改吗？")

graph = StateGraph().add_node(plan).add_node(execute).add_node(replan)
graph.add_conditional_edges("execute", lambda s: "END" if not s["plan"] else "replan")

• LangGraph 的 Plan-and-Execute 教程 — 工程实现详解
• 论文 Plan-and-Solve Prompting — 早期范式的对照

就像写代码时先跑一遍单元测试——失败了不是马上换方案，而是先让自己写一段"反思笔记"（这次为什么错），把笔记塞进下一轮的 prompt，再重试。Reflexion = 失败 → 自评 → 学到东西 → 重来。

ReAct 和 Plan-and-Execute 都有个共性问题：错了就错了，不会从错误中学习。如果第一次解法失败，第二次还是大概率沿同一个错误路径。Reflexion 的洞察是：把"反思失败原因"显式写进 prompt 当成下一轮的额外 context，模型表现会显著变好——就像人类做题错了之后写错题本，下次遇到类似题就避坑。这种"自我反思作为短期记忆"是无需重新训练就提升 Agent 表现的低成本手段。

关键三件套：Actor（执行任务）、Evaluator（判断成败）、Reflector（写反思笔记）。反思不存进模型权重，只存在 prompt 里——所以是"短期记忆"，类似你给同事的 review 反馈，他改一遍提交，不是重新培训。

def reflexion_loop(task, max_trials=3):
    memory = []  # 反思笔记列表（短期记忆）

    for trial in range(max_trials):
        # 1) Actor 带反思笔记尝试任务
        prompt = f"""任务: {task}
过往反思笔记（避免重复同样的错误）:
{chr(10).join(memory)}"""
        output = actor_llm.invoke(prompt)

        # 2) Evaluator 判断成败（写代码题用单元测试；其他用 LLM judge）
        success, error_msg = evaluator(output)
        if success:
            return output

        # 3) Reflector 写反思笔记
        reflection = reflector_llm.invoke(f"""
你刚才完成了任务: {task}
你的输出: {output}
评估结果: {error_msg}
请用一段话反思: 这次为什么失败？下次应该怎么改？
""")
        memory.append(reflection)

    return output  # 达到上限仍未成功

• 原论文 Reflexion: Language Agents with Verbal Reinforcement Learning
• LangGraph 的 Reflection / Reflexion 教程

就像把 ReAct + 文件系统 + 记忆库 + 子任务队列 打包成"一个 main 函数无限循环"——你只丢一句话目标进去，Agent 自己拆任务、调工具、记中间结果、再继续，直到它觉得目标完成（或者把你的信用卡刷爆）。AutoGPT 是 2023 年点燃"自主 Agent"概念的开源项目，奠定了之后所有 autonomous agent 的设计模板。

之前的 Agent 都需要人在每一步介入或者只能跑短期任务。AutoGPT 试图回答："能不能给 LLM 一个长期目标（如『帮我搜罗本月所有 AI 创业公司的融资新闻并写成报告』），然后让它完全自主跑下去？"为此它把几个能力拼到一起：(a) 任务队列管理（todo / done）、(b) 文件读写（长期工作区）、(c) 互联网访问、(d) 向量记忆库（跨步骤记住关键信息）、(e) 子目标分解。它没发明新算法，价值在于"把 LLM 包装成一个具有持久状态的进程"这个工程范式。

注意它跟前三个架构的关系：内层每一步用 ReAct 决策；步骤之间通过任务队列做 Plan-and-Execute 风格的编排；失败时也可以加 Reflexion。所以 AutoGPT 不是替代品，是集大成的工程模板——一个面向"持续运行的 autonomous agent"的参考实现。

# 极简版 AutoGPT 主循环 —— 真实 AutoGPT 是它的工程化版本
def autogpt(goal, max_steps=50, budget_usd=5):
    task_queue = [goal]
    memory = VectorMemory()  # Chroma/FAISS 实例
    workspace = FileSystem("./workspace")
    cost = 0.0

    for step in range(max_steps):
        if not task_queue or cost > budget_usd:
            break

        task = task_queue.pop(0)
        context = memory.search(task, k=5)  # 召回历史相关

        # LLM 决定本步动作（用结构化输出）
        decision = llm.invoke(prompt=f"""
你是 {agent_persona}, 目标: {goal}
当前子任务: {task}
相关记忆: {context}
请输出 JSON: {{"action": "search|write_file|spawn_task|done",
              "args": ...}}
""")

        # 执行 + 写入记忆/任务队列/文件系统
        result = execute_action(decision, workspace)
        memory.add(f"step{step}: {task} → {result}")
        cost += estimate_cost(decision)

        if decision["action"] == "spawn_task":
            task_queue.extend(decision["args"]["subtasks"])
        elif decision["action"] == "done":
            break

• AutoGPT GitHub 项目 — 原始 README 看设计思路
• Anthropic 的 Building Effective Agents — 反思 autonomous agent 的边界