跳至主要內容

Agent 框架

SunnyFan大约 10 分钟约 3057 字

Agent 框架

简介

Agent 框架的价值，不在于把大模型包装成"会循环调用工具"的东西，而在于把模型推理、工具调用、任务拆解、执行控制、记忆、观测和安全边界组织成一套可运行的系统。相比单轮问答，Agent 更适合完成多步骤任务，但也更容易引入高成本、低稳定性和错误累积的问题。

Agent 系统的设计核心是"在可控的范围内赋予模型自主性"。一个设计良好的 Agent 应该做到：知道什么时候该自己决定、什么时候该调用工具、什么时候该请求人工确认、什么时候该停止并报告失败。过度的自主性会导致不可控的结果，而过度的限制则会让 Agent 退化为简单的 API 调用链。

特点

实现

最小 Agent：工具 + 推理循环

import json
from typing import Callable

class MiniAgent:
    def __init__(self, llm: Callable[[str], str], tools: dict[str, Callable[[str], str]], max_steps: int = 5):
        self.llm = llm
        self.tools = tools
        self.max_steps = max_steps

    def run(self, question: str) -> str:
        scratchpad = ""

        for step in range(self.max_steps):
            prompt = f"""
你是一个任务助手。

可用工具: {list(self.tools.keys())}

当前问题: {question}
{scratchpad}

请输出 JSON:
{{"thought":"你的思考","action":"工具名或finish","action_input":"输入"}}
"""
            raw = self.llm(prompt)
            data = json.loads(raw)

            thought = data["thought"]
            action = data["action"]
            action_input = data["action_input"]

            scratchpad += f"\nThought: {thought}"

            if action == "finish":
                return action_input

            if action not in self.tools:
                scratchpad += f"\nObservation: 未知工具 {action}"
                continue

            result = self.tools[action](action_input)
            scratchpad += f"\nAction: {action}"
            scratchpad += f"\nAction Input: {action_input}"
            scratchpad += f"\nObservation: {result}"

        return "达到最大步数限制，任务未完成"
# 模拟工具

def web_search(query: str) -> str:
    return f"搜索结果：关于【{query}】的 3 条摘要"


def calculator(expression: str) -> str:
    return str(eval(expression))


def fake_llm(prompt: str) -> str:
    if "销售额" in prompt and "Observation" not in prompt:
        return json.dumps({
            "thought": "我需要先查找销售数据",
            "action": "web_search",
            "action_input": "本月销售额 100, 上月销售额 80"
        }, ensure_ascii=False)

    return json.dumps({
        "thought": "我已经得到信息，可以回答了",
        "action": "finish",
        "action_input": "本月销售额比上月增长 25%"
    }, ensure_ascii=False)

agent = MiniAgent(fake_llm, {"web_search": web_search, "calculator": calculator})
print(agent.run("请分析销售额增长情况"))
最小 Agent 至少要解决：
- 工具怎么描述
- 输出怎么结构化
- 最多执行多少步
- 中间结果如何记录
- 失败时如何停下来

ReAct 与计划执行模式

from pydantic import BaseModel
from typing import List

class PlanStep(BaseModel):
    task: str
    tool: str
    tool_input: str

class TaskPlan(BaseModel):
    goal: str
    steps: List[PlanStep]

class PlanningAgent:
    def __init__(self, planner_llm, executor_llm, tools):
        self.planner_llm = planner_llm
        self.executor_llm = executor_llm
        self.tools = tools

    def plan(self, goal: str) -> TaskPlan:
        # 简化示意：真实项目中这里应输出稳定 JSON
        return TaskPlan(
            goal=goal,
            steps=[
                PlanStep(task="搜索竞品定价", tool="search", tool_input="竞品A 竞品B 定价"),
                PlanStep(task="整理搜索结果", tool="summarize", tool_input="竞品定价结果"),
                PlanStep(task="输出报告", tool="finish", tool_input="生成分析结论")
            ]
        )

    def execute(self, plan: TaskPlan):
        results = []
        for step in plan.steps:
            if step.tool == "finish":
                results.append(step.tool_input)
                break

            result = self.tools[step.tool](step.tool_input)
            results.append({"task": step.task, "result": result})
        return results
# 计划-执行更适合复杂任务：
# 1. 信息收集
# 2. 多数据源整合
# 3. 子任务顺序明确
# 4. 需要中间校验和人工插入的任务

Agent 状态管理

from enum import Enum
from dataclasses import dataclass, field
from typing import Optional

class AgentState(Enum):
    IDLE = "idle"
    PLANNING = "planning"
    EXECUTING = "executing"
    WAITING_APPROVAL = "waiting_approval"
    FINISHED = "finished"
    FAILED = "failed"

@dataclass
class AgentContext:
    """Agent 执行上下文——记录完整的执行轨迹"""
    task_id: str
    goal: str
    state: AgentState = AgentState.IDLE
    plan: Optional[list] = None
    current_step: int = 0
    observations: list = field(default_factory=list)
    tool_calls: list = field(default_factory=list)
    errors: list = field(default_factory=list)
    final_answer: Optional[str] = None

    def to_trace(self) -> dict:
        """生成可读的执行轨迹"""
        return {
            "task_id": self.task_id,
            "goal": self.goal,
            "state": self.state.value,
            "steps_completed": self.current_step,
            "tool_calls_count": len(self.tool_calls),
            "errors_count": len(self.errors),
            "has_answer": self.final_answer is not None
        }

ctx = AgentContext(task_id="task-001", goal="分析竞品数据")
print(f"初始状态: {ctx.to_trace()}")

LangChain / 工具注册思路

from langchain.tools import tool

@tool
def search_database(query: str) -> str:
    """搜索业务数据库中的商品信息"""
    # 真实场景里这里应该是参数化查询，而不是字符串拼接
    return f"数据库搜索结果: {query}"

@tool
def calculate_metrics(data: str) -> str:
    """对逗号分隔的数值做基础统计"""
    values = [float(x.strip()) for x in data.split(',')]
    return f"均值={sum(values)/len(values):.2f}, 最大值={max(values)}, 最小值={min(values)}"
# 工具描述要具体，否则模型很难正确选择
# 好的描述应包含：
# 1. 这个工具解决什么问题
# 2. 输入格式是什么
# 3. 输出是什么
# 4. 什么时候该用 / 不该用

生产环境必须补的治理层

from dataclasses import dataclass
from time import time

@dataclass
class AgentRunLog:
    trace_id: str
    question: str
    step_count: int
    tools_called: list[str]
    start_time: float
    end_time: float | None = None
    final_answer: str | None = None

run_log = AgentRunLog(
    trace_id="agent-20260412-001",
    question="生成退款规则分析",
    step_count=0,
    tools_called=[],
    start_time=time()
)
# 高风险工具加白名单/审批
HIGH_RISK_TOOLS = {"delete_user", "refund_order", "shutdown_device"}

def can_execute_tool(tool_name: str) -> bool:
    return tool_name not in HIGH_RISK_TOOLS

print(can_execute_tool("refund_order"))  # False
# 超时与最大步数限制非常重要
MAX_STEPS = 5
MAX_RUNTIME_SECONDS = 20

Agent 评估

class AgentEvaluator:
    """Agent 系统评估器"""

    def __init__(self):
        self.test_cases = []

    def add_test_case(self, goal, expected_tools, expected_answer_contains=None):
        """添加测试用例"""
        self.test_cases.append({
            "goal": goal,
            "expected_tools": expected_tools,
            "expected_answer_contains": expected_answer_contains or []
        })

    def evaluate(self, agent):
        """运行评估"""
        results = []
        for case in self.test_cases:
            ctx = agent.run(case["goal"])
            # 检查工具调用是否正确
            tools_used = set(ctx.tool_calls)
            expected = set(case["expected_tools"])
            tool_accuracy = len(tools_used & expected) / len(expected) if expected else 1.0

            # 检查答案是否包含关键信息
            if ctx.final_answer and case["expected_answer_contains"]:
                answer_quality = all(
                    kw in ctx.final_answer
                    for kw in case["expected_answer_contains"]
                )
            else:
                answer_quality = ctx.final_answer is not None

            results.append({
                "goal": case["goal"],
                "tool_accuracy": tool_accuracy,
                "answer_quality": answer_quality,
                "steps": ctx.current_step,
                "errors": len(ctx.errors)
            })

        return results

evaluator = AgentEvaluator()
evaluator.add_test_case(
    goal="查询本月销售数据",
    expected_tools=["search_database"],
    expected_answer_contains=["销售额"]
)
print("Agent 评估框架已定义")

Agent 工具设计模式

def good_tool_description_example():
    """
    好的工具描述 vs 差的工具描述
    工具描述的质量直接影响 Agent 选择工具的准确性
    """
    good_examples = {
        "search_order": {
            "name": "search_order",
            "description": "根据订单号查询订单信息，包括状态、金额、收货地址和物流信息。输入为订单号（字符串），输出为订单详情（JSON）。",
            "when_to_use": "当用户询问订单状态、物流信息、退款进度时使用",
            "when_not_to_use": "当用户询问商品信息、价格或库存时不要使用"
        },
        "create_refund": {
            "name": "create_refund",
            "description": "为指定订单创建退款申请。输入为订单号和退款原因。只有在订单状态为'已签收'且签收时间不超过7天时才能创建。",
            "when_to_use": "当用户明确要求退款且订单符合退款条件时",
            "when_not_to_use": "订单未签收或超过退款期限时不要使用"
        }
    }

    for name, info in good_examples.items():
        print(f"\n工具: {name}")
        print(f"  描述: {info['description']}")
        print(f"  适用: {info['when_to_use']}")
        print(f"  不适用: {info['when_not_to_use']}")

good_tool_description_example()

Agent 成本控制

class AgentCostController:
    """
    Agent 成本控制器：
    - Token 预算限制
    - 步数限制
    - 工具调用次数限制
    """

    def __init__(self, max_steps=5, max_tokens=4000, max_tool_calls=10):
        self.max_steps = max_steps
        self.max_tokens = max_tokens
        self.max_tool_calls = max_tool_calls
        self.current_tokens = 0
        self.current_tool_calls = 0
        self.current_steps = 0

    def can_continue(self):
        """检查是否可以继续执行"""
        checks = {
            "steps": self.current_steps < self.max_steps,
            "tokens": self.current_tokens < self.max_tokens,
            "tool_calls": self.current_tool_calls < self.max_tool_calls,
        }
        all_ok = all(checks.values())
        if not all_ok:
            failed = [k for k, v in checks.items() if not v]
            return False, f"达到限制: {', '.join(failed)}"
        return True, "ok"

    def record_step(self, tokens_used=0, tool_called=False):
        """记录执行消耗"""
        self.current_steps += 1
        self.current_tokens += tokens_used
        if tool_called:
            self.current_tool_calls += 1

    def summary(self):
        return {
            "steps_used": self.current_steps,
            "steps_limit": self.max_steps,
            "tokens_used": self.current_tokens,
            "tokens_limit": self.max_tokens,
            "tool_calls_used": self.current_tool_calls,
            "tool_calls_limit": self.max_tool_calls,
        }

controller = AgentCostController(max_steps=5, max_tokens=8000, max_tool_calls=10)
can_go, reason = controller.can_continue()
print(f"可以继续: {can_go}, 原因: {reason}")
controller.record_step(tokens_used=500, tool_called=True)
print(controller.summary())

Agent 与人类协作

class HumanInLoopAgent:
    """
    人在环路的 Agent：
    在关键决策点暂停，等待人工确认
    """

    def __init__(self, llm, tools):
        self.llm = llm
        self.tools = tools
        self.pending_actions = []
        self.approved_actions = []

    def run_with_approval(self, question):
        """带人工审批的执行流程"""
        steps = self._plan_steps(question)

        for step in steps:
            # 检查是否需要人工确认
            if step["risk_level"] == "high":
                approval = self._request_human_approval(step)
                if not approval:
                    self.pending_actions.append({"step": step, "status": "rejected"})
                    continue

            result = self.tools[step["tool"]](step["input"])
            self.approved_actions.append({"step": step, "status": "approved", "result": result})

        return self._synthesize_results()

    def _plan_steps(self, question):
        """简化：返回执行计划"""
        return [
            {"tool": "search", "input": question, "risk_level": "low"},
            {"tool": "database_write", "input": "update data", "risk_level": "high"},
        ]

    def _request_human_approval(self, step):
        """请求人工审批"""
        print(f"[需要审批] 工具: {step['tool']}, 输入: {step['input']}")
        return True  # 实际系统中应等待真实响应

    def _synthesize_results(self):
        return "任务完成"

print("人在环路 Agent 已定义")
Agent 不是"自动执行越多越强"，而是：
- 任务拆解合理
- 工具能力清晰
- 执行轨迹可回放
- 高风险动作可拦截
- 失败可以及时停止

优点

缺点

总结

Agent 框架真正解决的，不是"让模型变聪明"，而是"让模型在一套受控执行框架中做事"。如果没有工具边界、结构化输出、超时限制和审计机制，Agent 很容易变成一个看起来会思考、但实际不可控的复杂黑盒。

关键知识点

  • ReAct 是循环式推理，不等于简单链式 API 调用。
  • 工具设计质量往往比模型本身更决定 Agent 效果。
  • 计划模式适合复杂任务，但不代表所有任务都值得先规划。
  • Agent 成功率会随着步骤数增加而下降，需要中间校验。
  • 状态管理和执行轨迹回放是生产环境的关键能力。

项目落地视角

  • 客服助手：查知识库、查订单、生成建议回复。
  • 内部运营助手：聚合数据、查询报表、生成分析摘要。
  • 开发助手：检索代码、运行测试、整理修改建议。
  • 企业知识问答：Agent 往往要和 RAG、权限系统、日志系统一起工作。

常见误区

  • 把所有复杂逻辑都交给模型自己决定。
  • 不记录中间步骤，出了问题无法回放。
  • 工具描述太模糊，导致模型频繁误调用。
  • 没有设置最大步数和超时，结果任务无限循环。

进阶路线

  • 学习 LangChain、LangGraph、AutoGen、CrewAI 等不同编排思路。
  • 研究多 Agent 协作、DAG 编排、Memory 与状态机。
  • 建立 Agent 评估数据集，而不是只看单次演示效果。
  • 将 Agent 与 RAG、函数调用、安全审核和人工确认结合起来。

适用场景

  • 多步骤分析任务。
  • 业务助手 / Copilot。
  • 数据检索 + 处理 + 输出一体化任务。
  • 需要调用多个工具完成任务的智能工作流。

落地建议

  • 从最小可控 Agent 开始，不要一上来堆十几个工具。
  • 所有工具都要有清晰描述、输入校验和异常处理。
  • 高风险操作一定要加人工确认和权限控制。
  • 先建评估集和失败样例库，再谈持续优化。

排错清单

  • 工具调用乱：先看工具描述是否重叠或含糊。
  • 结果不稳定：先看输出格式是否约束足够强。
  • 成本过高：先看步骤数、上下文长度和是否真的需要规划阶段。
  • 线上不可控：先补审计日志、超时控制、人工确认和禁用高风险工具。

复盘问题

  • 这个任务真的需要 Agent，还是简单函数调用 / RAG 就够了？
  • 当前 Agent 的主要瓶颈在模型、工具、规划，还是治理？
  • 哪些步骤最容易失败？是否应该人工插入校验点？
  • 你的 Agent 是"看起来很智能"，还是"真的能稳定完成任务"？

延伸阅读

Copyright © 2026 SunnyFan