AI Agent架构设计：让大模型从"答问者"蜕变为"执行者"的实验手记

一、当大模型只会"说"不会"做"，痛点何在？

大模型能写诗、能解题、能侃侃而谈，但它始终被困在文本的牢笼里。你让它订机票，它给你写一段订机票的步骤说明；你让它查数据库，它给你一段SQL示例。模型知道"怎么做"，却无法"真的做"。这就是AI Agent要解决的核心问题——赋予大模型行动能力。Agent不是简单的API封装，而是一个包含感知、规划、执行、反馈的闭环系统。设计一个可靠的Agent，远比写一个好Prompt复杂得多。本文记录了我在Agent系统设计中的实验与思考，从架构选型到多模态交互，一步步拆解Agent的工程实现。

二、Agent的底层架构：感知-规划-执行-反馈闭环

2.1 ReAct范式：推理与行动的交织

ReAct（Reasoning + Acting）是当前Agent设计的主流范式。模型在每一步先"思考"（Thought），再选择"行动"（Action），然后观察"结果"（Observation），循环往复直到任务完成。这种范式的优势在于：推理过程可追溯，行动结果可验证，错误可回溯。

graph LR
    A[用户任务输入] --> B[Thought: 推理当前状态]
    B --> C[Action: 选择并执行工具]
    C --> D[Observation: 观察执行结果]
    D --> E{任务是否完成?}
    E -->|否| B
    E -->|是| F[输出最终结果]

    style A fill:#fff3e0
    style F fill:#c8e6c9
    style E fill:#e1f5fe

2.2 工具调用的函数映射机制

Agent的核心能力之一是工具调用。大模型输出结构化的工具调用请求，系统将其映射到具体的函数执行。这个映射过程需要解决三个问题：工具描述如何让模型准确理解、参数如何安全传递、执行结果如何高效反馈。

2.3 记忆系统：短期上下文与长期知识

Agent需要记忆来维持状态。短期记忆依赖上下文窗口，存储当前对话和中间结果；长期记忆则需要外部存储，如向量数据库，用于跨会话的知识检索。记忆系统的设计直接影响Agent的连贯性和知识积累能力。

三、生产级Agent框架实现

3.1 核心Agent类设计

import json
import re
from typing import Any, Callable, Optional
from dataclasses import dataclass, field

@dataclass
class Tool:
    """工具定义：将函数封装为模型可理解的描述"""
    name: str
    description: str
    parameters: dict  # JSON Schema格式的参数描述
    func: Callable
    required_params: list = field(default_factory=list)

    def execute(self, **kwargs) -> str:
        """安全执行工具函数，捕获异常避免Agent崩溃"""
        try:
            result = self.func(**kwargs)
            # 将结果转为字符串，控制长度避免上下文溢出
            result_str = str(result)
            if len(result_str) > 2000:
                result_str = result_str[:2000] + "\n...[结果已截断]"
            return result_str
        except Exception as e:
            # 返回错误信息而非抛出异常，让Agent有机会自我修正
            return f"工具执行失败：{type(e).__name__}: {str(e)}"


class Agent:
    """基于ReAct范式的AI Agent框架"""

    def __init__(self, llm_client: Any, system_prompt: str = "",
                 max_iterations: int = 10):
        self.llm_client = llm_client
        self.system_prompt = system_prompt
        self.max_iterations = max_iterations  # 防止无限循环
        self.tools: dict[str, Tool] = {}
        self.conversation_history = []

    def register_tool(self, tool: Tool):
        """注册工具，Agent可动态扩展能力"""
        self.tools[tool.name] = tool

    def _build_tool_descriptions(self) -> str:
        """生成工具描述文本，供模型理解可用工具"""
        descriptions = []
        for tool in self.tools.values():
            param_desc = json.dumps(tool.parameters, ensure_ascii=False)
            descriptions.append(
                f"工具名：{tool.name}\n"
                f"描述：{tool.description}\n"
                f"参数：{param_desc}\n"
                f"必填参数：{tool.required_params}"
            )
        return "\n\n".join(descriptions)

    def _parse_action(self, response: str) -> Optional[dict]:
        """从模型输出中解析工具调用请求"""
        # 尝试匹配 Action: xxx\nAction Input: {...} 格式
        action_pattern = r"Action:\s*(\w+)\s*\n\s*Action Input:\s*(\{.*?\})"
        match = re.search(action_pattern, response, re.DOTALL)

        if match:
            tool_name = match.group(1).strip()
            try:
                params = json.loads(match.group(2).strip())
            except json.JSONDecodeError:
                # 参数解析失败时返回空参数，让工具自行处理
                params = {}
            return {"tool": tool_name, "params": params}

        return None

    def run(self, task: str) -> str:
        """执行Agent任务，ReAct循环"""
        # 构建初始Prompt
        tool_desc = self._build_tool_descriptions()
        agent_prompt = (
            f"{self.system_prompt}\n\n"
            f"你可以使用以下工具：\n{tool_desc}\n\n"
            f"请按以下格式思考和行动：\n"
            f"Thought: 分析当前状态，思考下一步\n"
            f"Action: 选择工具名称\n"
            f"Action Input: 工具参数（JSON格式）\n\n"
            f"当你认为任务已完成时，请输出：\n"
            f"Final Answer: 最终答案\n\n"
            f"用户任务：{task}"
        )

        self.conversation_history = [
            {"role": "system", "content": agent_prompt},
            {"role": "user", "content": task},
        ]

        for i in range(self.max_iterations):
            # 调用大模型获取下一步行动
            response = self.llm_client.chat(
                messages=self.conversation_history,
                temperature=0.0,  # Agent场景需要高确定性
            )

            # 检查是否输出最终答案
            final_match = re.search(
                r"Final Answer:\s*(.*)", response, re.DOTALL
            )
            if final_match:
                return final_match.group(1).strip()

            # 解析工具调用
            action = self._parse_action(response)
            if action is None:
                # 模型未输出有效Action，引导其重新思考
                self.conversation_history.append(
                    {"role": "assistant", "content": response}
                )
                self.conversation_history.append(
                    {"role": "user", "content": "请按格式输出Thought和Action"}
                )
                continue

            # 执行工具
            tool_name = action["tool"]
            if tool_name not in self.tools:
                observation = f"错误：工具'{tool_name}'不存在，可用工具：{list(self.tools.keys())}"
            else:
                observation = self.tools[tool_name].execute(**action["params"])

            # 将推理过程和观察结果加入对话历史
            self.conversation_history.append(
                {"role": "assistant", "content": response}
            )
            self.conversation_history.append(
                {"role": "user", "content": f"Observation: {observation}"}
            )

        return "Agent达到最大迭代次数，任务未完成。"

3.2 多模态交互：让Agent"看"和"听"

import base64
from pathlib import Path

class MultimodalAgent(Agent):
    """支持多模态输入的Agent扩展"""

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._register_multimodal_tools()

    def _register_multimodal_tools(self):
        """注册多模态处理工具"""
        self.register_tool(Tool(
            name="analyze_image",
            description="分析图片内容，提取文字、物体、场景等信息",
            parameters={
                "type": "object",
                "properties": {
                    "image_path": {
                        "type": "string",
                        "description": "图片文件路径"
                    },
                    "question": {
                        "type": "string",
                        "description": "针对图片提出的问题"
                    }
                },
                "required": ["image_path"]
            },
            required_params=["image_path"],
            func=self._analyze_image,
        ))

    def _analyze_image(self, image_path: str,
                       question: str = "描述图片内容") -> str:
        """调用视觉模型分析图片"""
        path = Path(image_path)
        if not path.exists():
            return f"图片文件不存在：{image_path}"

        # 读取图片并编码为Base64
        image_data = base64.b64encode(path.read_bytes()).decode()
        mime_type = self._get_mime_type(path.suffix)

        # 构建多模态消息
        messages = [{
            "role": "user",
            "content": [
                {"type": "text", "text": question},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": f"data:{mime_type};base64,{image_data}"
                    }
                }
            ]
        }]

        try:
            response = self.llm_client.chat(
                messages=messages,
                temperature=0.1,
            )
            return response
        except Exception as e:
            return f"图片分析失败：{str(e)}"

    @staticmethod
    def _get_mime_type(suffix: str) -> str:
        """根据文件后缀推断MIME类型"""
        mime_map = {
            ".jpg": "image/jpeg",
            ".jpeg": "image/jpeg",
            ".png": "image/png",
            ".gif": "image/gif",
            ".webp": "image/webp",
        }
        return mime_map.get(suffix.lower(), "image/jpeg")

3.3 Agent编排：多Agent协作框架

class AgentOrchestrator:
    """多Agent编排器，协调多个专业Agent完成复杂任务"""

    def __init__(self):
        self.agents: dict[str, Agent] = {}
        self.routing_prompt = (
            "根据以下任务描述，判断应该由哪个Agent处理。\n"
            "可选Agent：{agent_list}\n"
            "任务：{task}\n"
            "请只输出Agent名称，不要输出其他内容。"
        )

    def register_agent(self, name: str, agent: Agent,
                       capability: str):
        """注册专业Agent及其能力描述"""
        self.agents[name] = {
            "agent": agent,
            "capability": capability,
        }

    def execute(self, task: str, llm_client: Any) -> str:
        """路由任务到合适的Agent执行"""
        if not self.agents:
            return "未注册任何Agent"

        # 构建Agent列表描述
        agent_list = "\n".join([
            f"- {name}: {info['capability']}"
            for name, info in self.agents.items()
        ])

        # 使用LLM做任务路由
        routing_input = self.routing_prompt.format(
            agent_list=agent_list, task=task
        )
        selected_name = llm_client.chat(
            messages=[{"role": "user", "content": routing_input}],
            temperature=0.0,
        ).strip()

        # 路由到选中的Agent
        if selected_name in self.agents:
            return self.agents[selected_name]["agent"].run(task)
        else:
            # 路由失败时降级到第一个Agent
            fallback = list(self.agents.keys())[0]
            return self.agents[fallback]["agent"].run(task)

四、Agent系统的架构权衡与边界分析

4.1 自主性 vs 可控性

Agent的自主性越强，能完成的任务越复杂，但失控的风险也越高。一个不受约束的Agent可能陷入死循环、执行危险操作、或产生不可预期的副作用。设计Agent时，必须在自主性和可控性之间找到平衡。实践中，通过限制最大迭代次数、设置工具权限白名单、加入人工确认环节，可以有效降低风险。

4.2 延迟 vs 准确性

Agent的每一步推理和工具调用都会增加延迟。一个需要5步推理的任务，延迟可能是单次调用的5倍以上。在对延迟敏感的场景中，需要权衡"多步推理的准确性"和"单步响应的时效性"。一种折中方案是：简单任务走快速路径（单次调用），复杂任务走推理路径（多步Agent）。

4.3 上下文窗口的消耗

Agent的ReAct循环会快速消耗上下文窗口。每一步的Thought、Action、Observation都占用Token。当任务步骤较多时，早期信息可能被截断，导致Agent"遗忘"关键上下文。解决方案包括：摘要压缩历史信息、只保留关键观察结果、使用向量数据库存储中间状态。

4.4 工具设计的Granularity

工具粒度太细，Agent需要更多步骤完成任务；粒度太粗，工具复用性降低。例如，"查询数据库"是一个粗粒度工具，"执行SQL"是细粒度工具。前者使用简单但灵活性低，后者灵活但需要Agent具备SQL编写能力。工具粒度的选择取决于Agent的推理能力和任务复杂度。

五、总结

AI Agent的设计，本质上是将大模型从"被动的文本生成器"升级为"主动的任务执行者"。ReAct范式提供了推理与行动交织的框架，工具系统赋予了模型与外部世界交互的能力，多模态扩展让Agent的感知维度从文本走向视觉和语音。但Agent系统也面临自主性与可控性的矛盾、延迟与准确性的权衡、上下文窗口的消耗等挑战。设计一个生产级Agent，不是简单地拼接API，而是在这些约束条件下做出合理的架构选择。就像八卦中的阴阳相生，Agent的"智能"和"约束"也是一体两面——没有约束的智能是危险的，没有智能的约束是无用的。找到那个平衡点，才是Agent设计的真正艺术。