如何构建一个不会失控的AI智能体：智能体循环、规划、工具调用、记忆与护栏

# 绝对不要将此代码直接部署到生产环境。缺乏退出机制、工具验证、预算控制与错误处理。
def naive_agent(task, tools):
    messages = [{"role": "user", "content": task}]
    while True:                                   # (1) 存在无限循环风险
        resp = llm(messages, tools=tools)
        if resp.tool_calls:
            for call in resp.tool_calls:
                result = tools[call.name](**call.args)   # (2) 工具幻觉会导致KeyError
                messages.append(tool_result(call, result))  # (4) 工具抛错时直接崩溃
        else:
            return resp.content                  # (3) 缺少Token与步骤预算限制

def run_agent(task: str, tools: dict, max_steps: int = 8) -> str:
    messages = [system_prompt(tools), {"role": "user", "content": task}]
    for step in range(max_steps):                      # 硬上限 -> 防止无限循环
        resp = llm(messages, tools=tool_schemas(tools))
        if not resp.tool_calls:
            return resp.content                        # 模型判定任务完成
        for call in resp.tool_calls:
            if call.name not in tools:                 # 拒绝幻觉工具
                result = f"error: unknown tool '{call.name}'"
            else:
                result = safe_invoke(tools[call.name], call.args)  # 内部包含try/except
            messages.append(tool_result(call, result)) # 观察结果 -> 供下一步推理使用
    return "stopped: step budget exhausted"            # 安全失败，永不无限运行

@tool
def get_campaign_kpis(campaign_id: str, date_from: str, date_to: str) -> dict:
    """返回指定日期范围内某个活动的KPI（展示量、点击量、转化率、花费）。
    日期格式为ISO-8601 (YYYY-MM-DD)。使用此工具回答性能相关问题。"""
    validate_iso_date(date_from); validate_iso_date(date_to)   # 永远不要信任模型的参数
    require_tenant_scope(campaign_id)                          # 权限控制：智能体不能跨租户访问
    return kpi_service.query(campaign_id, date_from, date_to)

# LangGraph通过检查点（checkpointer）提供持久化的工作记忆与恢复能力。
agent = graph.compile(checkpointer=PostgresSaver(conn))   # 状态按thread_id持久化
context = memory_store.search(query=task, tenant_id=tenant, k=5)  # 长期记忆 = 检索

def guarded_step(state, call, tool):
    if state.steps > MAX_STEPS:      raise StopAgent("step budget")     # 限制循环步数
    if state.tokens > TOKEN_BUDGET:  raise StopAgent("token budget")    # 限制Token成本
    if tool.name not in ALLOWLIST:   return "error: tool not permitted" # 最小权限原则
    if tool.is_destructive and not human_approved(call):               # 确认写操作
        return "paused: awaiting human approval"
    result = safe_invoke(tool, call.args)                              # 确保循环永不崩溃
    return validate_output(result)                                     # 模式、PII与护栏检查

# LangGraph：智能体就是一个图。显式、可检查、可恢复。
graph = StateGraph(AgentState)
graph.add_node("reason", call_model)
graph.add_node(
    "act", run_tools
)
graph.add_conditional_edges("reason", lambda s: "act" if s.tool_calls else END)
graph.add_edge("act", "reason")                    # 观察 -> 再次推理（循环）
agent = graph.compile(checkpointer=memory)         # 持久化状态 + 恢复 + HITL暂停