首页 / 文章 / 实用提示：你的AI智能体并不聪明。以下是打造高效AI智能体的方法

machine-learningai-agentopenaiagentsllmarchitecture

实用提示：你的AI智能体并不聪明。以下是打造高效AI智能体的方法

《实用笔记》操作指南：你的AI智能体并不聪明。以下是打造真正智能的AI智能体的方法——包括相关合同、检查清单，以及供采用该模式的团队直接使用的代码模板。

3392 词

可将此内容视为《你的AI智能体并不聪明，这样才能打造真正会思考的智能体》一文中理念面向操作人员的重构版本：清晰的阶段划分、有序的代码模块，以及能在交接过程中保留的恢复说明。

目录

将“目录”阶段视为可量化的界面使用效果最佳。在扩大范围之前，先记录一份完美的操作日志、一个故障案例以及回滚说明。 优先选择小型且可测试的单元，而非庞大的脚本。当某一步骤出错时，故障应指向单一责任点，而非复杂的流程链。 保持图结构的状态简洁且具有类型定义。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在流程中断后导致无法继续执行。

为何大多数AI智能体不过是复杂的提示词链

将大多数AI智能体所处的阶段视为可测量的界面，是使其效果最佳的方式。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 把这一阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声无息的半完成状态。 需为每轮对话和每次会话设定token预算。智能体工具往往会过度扩展上下文，设置上限可避免演示过程变成意外的费用账单。

“直接使用ReAct”方法的弊端

将“仅阶段测试”这一方法视为可度量的指标时效果最佳。在扩大范围之前，先记录一个成功的测试案例、一个失败案例以及回滚说明。在功能结果旁同时记录执行时间以及令牌或查询成本。提前了解成本情况，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图表状态简洁且类型明确，嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在出现中断后导致流程无法继续。

架构：四个节点，一个循环

将“架构四节点”模型视为可测量的结构来使用效果最佳。在扩大范围之前，先记录一个成功的案例、一个失败案例以及回滚说明。 将配置与应用程序代码分开。环境文件、密钥存储和功能标志应集中存放，这样操作人员无需查看整个结构就能进行审计。 保持图结构的扁平化与类型化。嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在中断后导致无法继续处理。

START --> Planner --> Executor <--> Replanner --> Reporter --> END

状态管理：一切的核心

在将状态管理“The Backbone”阶段视为可度量的对象时，其效果最佳。在扩大范围之前，需记录一份理想的操作日志、一个故障案例以及回滚说明。 同时文档化正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。 保持图结构的状态简洁且具有类型约束。嵌套的数据块会掩盖哪个节点修改了哪个字段的信息，还会在中断后导致流程无法继续。 在将状态管理“The Backbone”阶段视为可度量的对象时，其效果最佳。在扩大范围之前，需记录一份理想的操作日志、一个故障案例以及回滚说明。 应将此阶段视为输入与经过验证的输出之间的契约。为相关工件命名，明确成功判定标准，杜绝无声的半完成状态。

import operator
from typing import Annotated, TypedDict

from pydantic import BaseModel, Field

class StrategyState(TypedDict, total=False):
    """Global state that flows through the LangGraph nodes."""
    query: str
    plan: list[dict]
    scratchpad: Annotated[list[dict], operator.add]
    current_step: int
    final_report: str
    replan_count: int

结构化输出模式：PlanStep与Plan

在结构化输出模式的设计阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。应在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。对于会产生支出或修改生产数据的操作，必须经过人工审批。编译时的配置并不等同于业务功能的完整性。

AVAILABLE_TOOLS_TEXT = """
- get_metrics(ticker, metric?): Return stock metrics. 'metric' is optional
  (P/E, EPS, Revenue, Market Cap, Sector).
- search_news(ticker): Return recent news headlines for a ticker.
- compare_metrics(tickers: list, metric): Compare one metric across
  multiple tickers.
"""

class PlanStep(BaseModel):
    """A single executable step inside an analysis plan."""
    step_id: int = Field(description="Sequential step number")
    tool: str = Field(
        description=f"Tool to use. Must be one of:\n{AVAILABLE_TOOLS_TEXT}"
    )
    args: dict = Field(description="Arguments for the tool call")
    purpose: str = Field(description="Why this step is needed")

class Plan(BaseModel):
    """The full plan generated by the planner node."""
    goal: str = Field(description="The overall analysis goal")
    steps: list[PlanStep] = Field(
        description="Ordered list of steps to execute"
    )

class ReplanDecision(BaseModel):
    """Output of the replanner node."""

reasoning: str = Field(
        description="Analysis of current progress and findings"
    )
    should_replan: bool = Field(
        description="Whether the plan needs modification"
    )
    updated_steps: list[PlanStep] = Field(
        default_factory=list,
        description="Remaining steps if replan is needed. Empty if no changes.",
    )

工具系统：三种工具，一个注册表

在工具系统第三阶段，应在修改代码之前明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储以及功能标志应集中存放于一个位置，以便操作人员无需查看整个系统结构即可进行审计。 在网关处进行身份验证，在数据层再次授权。仅凭承载令牌并不足以界定租户边界。

The ToolRegistry

在ToolRegistry阶段，修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程与异常恢复路径。重试机制、人工审核环节以及错误处理都是产品功能的一部分，而非后续需要补充的内容。 在网关处进行身份验证，在数据层再次授权。仅凭承载令牌并不足以界定租户边界。 在ToolRegistry阶段，修改代码之前需明确输入参数、该步骤的负责人以及结束标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关工件命名，定义成功检测标准，并拒绝默许的半完成状态。

from langchain_core.tools import BaseTool
from typing import Iterable, Mapping, Any

class ToolRegistry:
    """Namespace-aware container for LangChain tools."""
    def __init__(self) -> None:
        self._tools_by_toolset: dict[str, dict[str, BaseTool]] = {}
    def add_tools(self, toolset: str, tools: Iterable[BaseTool]) -> None:
        bucket = self._tools_by_toolset.setdefault(toolset, {})
        bucket.update({t.name: t for t in tools})
    def get_tools(self, toolset: str) -> tuple[BaseTool, ...]:
        return tuple(self._tools_by_toolset.get(toolset, {}).values())
    def invoke(
        self, toolset: str, tool_name: str, tool_args: Mapping[str, Any]
    ) -> Any:
        t = self._tools_by_toolset.get(toolset, {}).get(tool_name)
        if t is None:
            raise ValueError(
                f"Unknown tool '{tool_name}' in toolset '{toolset}'"
            )
        return t.invoke(dict(tool_args))

规划节点：行动前先思考

在处理“规划节点思考”阶段时，首先写下相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。

def planner_node(state: StrategyState) -> dict:
    """Create a step-by-step research plan using structured output."""
    planner = model.with_structured_output(Plan)

prompt = PLAN_PROMPT.format(
        available_tools=AVAILABLE_TOOLS_TEXT,
        ticker_choices=ticker_choices_text(),
        metric_choices=metric_choices_text(),
        query=state["query"],
    )
    plan: Plan = planner.invoke(prompt)
    steps = [s.model_dump() for s in plan.steps]
    return {"plan": steps, "current_step": 0}

PLAN_PROMPT：规则约束所在之处

在处理 PLANPROMPT 的“护栏”阶段时，首先写下相关契约：所需的输入参数、成功标志以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。 将配置信息与应用程序代码分开存放。环境文件、密钥存储以及功能开关应集中于一个位置，这样操作人员无需查看整个系统结构即可进行审计。 缓存系统中稳定的指令和工具架构。重复发送相同的开头信息是导致资源浪费的常见原因。

PLAN_PROMPT = """\
You are a financial research planner. Given a user's analysis request,
create a step-by-step research plan using the available tools.

Available tools:
{available_tools}
Rules:
- Use only the tools listed above.
- Every plan step must be executable with one of those tools.
- When a tool accepts 'ticker' or 'tickers', use only these exact values:
  {ticker_choices}
- When a tool accepts 'metric', use one of these exact values:
  {metric_choices}
- There are no other tools available. Final synthesis is handled separately.
Create an efficient plan. Group related lookups. Aim for 4-8 steps.
User request: {query}"""

执行节点：一步一个脚印

在处理“执行节点一”阶段时，首先需明确合同条款：所需的输入参数、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品功能的一部分，而非后续的优化工作。 在成本较高的操作之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次调用相同的大型语言模型。 在处理“执行节点一”阶段时，首先需明确合同条款：所需的输入参数、成功信号以及部分失败时的处理方式。这份清单能确保后续的代码修改保持一致性。 将此阶段视为输入与验证后输出之间的契约。为相关成果命名，定义成功判定标准，并拒绝默许部分完成的情况。

MAX_STEPS = 12  # Safety limit on total steps

def executor_node(state: StrategyState) -> dict:
    """Execute the next pending step from the plan."""
    plan = state.get("plan", [])
    current_step = state.get("current_step", 0)
    if current_step >= len(plan):
        return {}
    if current_step >= MAX_STEPS:
        return {"current_step": len(plan)}
    step = plan[current_step]
    tool_name = step["tool"]
    tool_args = step["args"]
    try:
        result = str(
            TOOL_REGISTRY.invoke(
                AgentName.EXECUTOR.value, tool_name, tool_args
            )
        )
        status = "Error" if result.startswith("Error:") else "Success"
    except Exception as exc:
        result = f"Error: {exc}"
        status = "Error"
    entry = {
        "step": current_step + 1,
        "tool": tool_name,
        "args": tool_args,
        "result": result,
        "status": status,
    }
    return {
        "scratchpad": [entry],
        "current_step": current_step + 1,
    }

重规划节点：自我修正的发生地

当将阶段视为可测量的表面时，Replanner节点的效果最佳。在扩大范围之前，先记录一份理想的执行结果、一个失败案例以及回滚说明。在功能结果旁还需记录执行时间以及令牌或查询成本。提前了解这些成本信息，就能避免在从演示环境过渡到共享环境时出现意外费用。要保持图结构的层次简单且类型明确，嵌套的数据块会掩盖是哪个节点修改了哪个字段，还会在进程中断后导致无法继续执行。

MAX_REPLANS = 2  # Prevent infinite replanning

def replanner_node(state: StrategyState) -> dict:
    """Review progress and optionally modify the remaining plan."""
    plan = state.get("plan", [])
    current_step = state.get("current_step", 0)
    replan_count = state.get("replan_count", 0)
    scratchpad = state.get("scratchpad", [])
    remaining = plan[current_step:]
    if len(remaining) = MAX_REPLANS:
        return {}
    scratchpad_text = "\n".join(
        f"Step {e['step']}: {format_tool_call(e['tool'], e['args'])} "
        f"-> [{e['status']}] {e['result'][:150]}..."
        for e in scratchpad
    )
    remaining_text = "\n".join(
        f"Step {s['step_id']}: {format_tool_call(s['tool'], s['args'])} "
        f"- {s['purpose']}"
        for s in remaining
    )
    replanner = model.with_structured_output(ReplanDecision)
    prompt = REPLAN_PROMPT.format(
        goal=state["query"],
        scratchpad=scratchpad_text,
        remaining_steps=remaining_text,
    )
    decision: ReplanDecision = replanner.invoke(prompt)
    if decision.should_replan and decision.updated_steps:
        new_steps = plan[:current_step] + [
            s.model_dump() for s in decision.updated_steps
        ]
        return {"plan": new_steps, "replan_count": replan_count + 1}
    return {"replan_count": replan_count + 1}

REPLAN_PROMPT

将REPLANPROMPT阶段视为可度量的对象来使用效果最佳。在扩大范围之前，先记录一份理想的处理结果、一个失败案例以及回滚说明。 应将配置置于应用程序代码之外。环境文件、密钥存储和功能开关应集中存放，这样操作人员无需查看整个系统结构即可进行审计。 为每轮对话和每次会话设定令牌预算。智能工具往往会大量消耗上下文资源，设置上限可避免演示过程突然产生额外费用。

REPLAN_PROMPT = """\
You are a financial research planner reviewing progress on a research task.

Original goal: {goal}
Completed steps and findings so far:
{scratchpad}
Remaining steps in the plan:
{remaining_steps}
Based on the findings so far, should the remaining plan change?
If an expected tool failed or revealed something unexpected, add a step
to investigate.
If a step is now redundant, remove it.
Use only the available executable tools already shown in the plan.
Do not add recommendation, summary, or report-writing steps."""

构建图结构：LangGraph组装

将“Wiring the Graph LangGraph”阶段视为可度量的界面来处理时，其效果最佳。在扩大范围之前，需记录一个理想运行案例、一个故障案例以及回滚说明。 同时记录正常流程与恢复流程。重试机制、人工审核环节以及死信处理都是产品本身的一部分，而非后续需要补充的内容。 保持图结构扁平且类型明确。嵌套的数据块会掩盖哪个节点编写了哪个字段的信息，还会在中断后导致无法继续处理。 将“Wiring the Graph LangGraph”阶段视为输入与已验证输出之间的契约。为相关成果命名，明确成功标准，绝不允许出现无声的半完成状态。

from langgraph.graph import StateGraph, END
from enum import Enum

class AgentName(Enum):
    PLANNER = "planner"
    EXECUTOR = "executor"
    REPLANNER = "replanner"
    REPORT = "report"

def build_graph():
    """Build and compile the LangGraph planning-agent workflow."""
    workflow = StateGraph(StrategyState)
    workflow.add_node(AgentName.PLANNER.value, planner_node)
    workflow.add_node(AgentName.EXECUTOR.value, executor_node)
    workflow.add_node(AgentName.REPLANNER.value, replanner_node)
    workflow.add_node(AgentName.REPORT.value, report_node)
    workflow.set_entry_point(AgentName.PLANNER.value)
    workflow.add_edge(AgentName.PLANNER.value, AgentName.EXECUTOR.value)
    workflow.add_edge(AgentName.EXECUTOR.value, AgentName.REPLANNER.value)
    workflow.add_conditional_edges(
        AgentName.REPLANNER.value,
        should_continue_execution,
        {
            AgentName.EXECUTOR.value: AgentName.EXECUTOR.value,
            AgentName.REPORT.value: AgentName.REPORT.value,
        },
    )
    workflow.add_edge(AgentName.REPORT.value, END)
    return workflow.compile()

def should_continue_execution(state: StrategyState) -> str:
    """Return the next node name after re-planning."""
    if state.get("current_step", 0) >= len(state.get("plan", [])):
        return AgentName.REPORT.value
    return AgentName.EXECUTOR.value

实际执行示例：跟踪单个查询

在接下来的实际执行示例中，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。除了功能结果外，还需记录执行时间以及令牌或查询成本。提前了解这些成本可以避免在从演示环境过渡到共享环境时出现意外费用。对于那些会耗费资金或修改生产数据的操作，必须经过人工审批。仅靠编译时的配置并不足以确保业务的完整性。

{
  "metric": "P/E",
  "values": {
    "NVDA": 58.3,
    "AMD": 102.5
  }
}

后续步骤

在“下一步去向”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 配置信息应置于应用程序代码之外。环境文件、密钥存储和功能标志应集中存放于一个位置，以便操作人员无需查看整个流程即可进行审计。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务流程的完整性。

总结

在“最终思考”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 需同时记录正常流程和异常恢复流程。重试机制、人工审核环节以及错误处理都是产品不可或缺的部分，而非后续才需要补充的内容。 对于涉及资金支出或修改生产数据的操作，必须经过人工审批。编译时的连接方式并不等同于业务功能的完整性。 在“最终思考”阶段，应在修改代码之前明确输入参数、该步骤的负责人以及终止标准。操作人员应能够从已知的检查点重新运行该步骤，而无需猜测隐藏状态。 应将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功判定标准，绝不允许出现无声无息的半完成状态。

让我们一起持续学习

在完成“让我们继续学习”阶段时，首先写下相关约定：所需的输入参数、成功标志以及部分失败时的处理方式。这样的清单能确保后续的代码修改保持一致性。 在功能结果旁记录执行时间以及令牌或查询成本。提前了解成本情况，可避免在从演示环境过渡到共享环境时出现意外费用。 在耗时较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次收取相同的LLM调用费用。

来自创始人的留言

在处理阶段中的A消息时，首先需列出相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。

运营检查清单

在完成运营检查清单阶段的工作时，同样要首先列出相关规范：所需输入、成功信号以及部分失败时的处理方式。这样的检查清单能确保后续的代码修改保持一致性。

应优先选择小型且可测试的单元，而非庞大的脚本。当某个步骤失败时，故障应能指向具体的责任模块，而非整个复杂的流程。

在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次计费相同的LLM调用。

锁定依赖版本，并记录用于运行演示的镜像摘要。可重复性比经验知识更重要。

将此阶段视为输入与经过验证的输出之间的契约。为相关成果命名，明确成功标准，拒绝默许的半完成状态。

在成本较高的步骤之后设置检查点。当操作员重新尝试后续节点时，恢复流程不应再次计费相同的LLM调用。

在升级整个系统之前，先冻结版本，为关键路径保存标准记录，并确认回滚步骤。共享环境需要设置速率限制、租户检查机制，以及明确的密钥轮换负责人。与其追求花哨的一次性演示，不如注重扎实的可靠性。

关于 fea74fe7fb83 的批处理说明：不要将提供者密钥放入代码仓库，为每个会话设置令牌上限，并将转录内容存储在评估测试用例的旁边，以便后续更换模型时仍能保持可比性。

全部文章发送简报