AI Agent基础学习笔记

# 简要

本笔记基于小林面试笔记的 16 道 Agent 面试题扩展而来，旨在通过面试题作为入口，系统深入学习 AI Agent 的完整知识体系。

1. 先理解面试题的基础回答
2. 学习深入原理部分，掌握技术本质
3. 阅读实现细节和代码示例
4. 结合实际项目理解应用场景
5. 参考扩展资料进行深度研究

# 目录

1. 什么是 Agent？与大模型有什么本质不同？
2. Agent 的基本架构由哪些核心组件构成？
3. Workflow，Agent，Tools 这三个的概念和区别介绍一下？
4. 了解哪些其他的 Agent 设计范式？Agent 和 Workflow 的区别是什么？
5. Agent 推理模式有哪些？ReAct 是啥？具体是怎么实现的？
6. ReAct、Plan-and-Execute、Reflection 三种范式有什么核心区别？实际项目中该如何选型？
7. 复杂任务怎么做的任务拆分？为什么要拆分？效果如何提升？
8. 请你介绍一下 AI Agent 的记忆机制，并说明在实际开发中应该如何设计记忆模块？
9. Agent 的长短期记忆系统怎么做的？记忆是怎么存的？粒度是多少？怎么用的？
10. 什么是 Multi-Agent？
11. 说说 Single-Agent 和 Multi-Agent 的设计方案？
12. Agent 记忆压缩通常有哪些方法？
13. 在工程实践中，为什么有时候选择「手搓」Agent，而不是直接用成熟框架？
14. 如何赋予 LLM 规划能力？
15. 讲讲 Agent 的反思机制？为什么要用反思？具体怎么实现？
16. 如何设计多 Agent 的协作与动态切换机制？

# 1. 什么是 Agent？与大模型有什么本质不同？

# 面试题基础回答

我理解 Agent 本质上是一个能自主完成目标的 AI 系统，跟传统 AI 最核心的区别在于「自主性」和「能行动」。

传统 AI 是你问一个问题它回答一个问题，每次都是独立的，被动响应；而 Agent 有自己的规划能力，你给它一个复杂目标，它会自己把任务拆成多步，通过调工具、访问记忆、感知环境来一步步执行，直到完成。

它不只是输出文字，而是真的能做事。

# 深入原理：Agent 的准确定义与核心特性

# 学术与工业界定义

• 狭义 Agent：能够感知环境并自主行动以实现特定目标的计算实体（源自分布式人工智能领域）
• 广义 Agent（大模型驱动）：基于大语言模型作为核心决策大脑，集成工具调用、记忆管理、自主规划、多步推理能力的智能系统，能够主动执行复杂任务而非被动响应查询

# 核心特性

1. 自主性：无需人类持续干预，自主制定执行计划并跟踪任务进度
2. 反应性：实时感知环境变化并做出适应性响应
3. 主动性：主动发起目标导向的行为，而非仅响应外部请求
4. 社会性：能够与其他 Agent 或系统进行交互协作（多 Agent 场景）

# 与大模型的本质差异（深入对比）

# 发展历程

1. 早期智能体：专家系统、多智能体系统（如 1990 年代的 FA/CONTACTS）
2. 传统机器学习 Agent：基于强化学习的游戏 AI、机器人控制系统
3. 大模型驱动 Agent：AutoGPT、BabyAGI（2023 年）、GPT-4o 集成的原生 Agent 能力（2024 年）

# 实际应用案例

• 代码开发 Agent：GitHub Copilot X、CodeLlama Agent、Amazon CodeWhisperer
• 科研助手 Agent：ChatGPT Code Interpreter、ResearchGPT、SciPhi
• 自动化办公 Agent：Microsoft 365 Copilot、Notion AI、Slack AI
• 智能家居 Agent：Alexa、Google Home、Home Assistant

# 2. Agent 的基本架构由哪些核心组件构成？

# 面试题基础回答

我理解 Agent 的基本架构有四个核心组件：LLM、工具、记忆、规划模块。

LLM 是整个系统的大脑，负责理解任务和做决策；工具让 Agent 能跟外部世界交互，搜索、执行代码、调 API 都靠它；记忆让 Agent 在任务执行过程中保持状态，不会「失忆」；规划模块负责把复杂目标拆解成可执行的步骤。

# 深入原理：各组件详细解析

# 1. LLM 核心（大脑）

作为整个 Agent 的决策中枢，负责处理所有输入信息并生成行动指令。

关键工程问题：

• 模型选型：推理能力、工具调用稳定性、上下文窗口大小、成本控制
  • 强推理模型：GPT-4o、Claude 3.5 Opus、Gemini Advanced
  • 平衡模型：GPT-4o Mini、Claude 3.5 Sonnet
  • 开源模型：Llama 3、Mistral Large、Qwen 2
• System Prompt 设计：定义 Agent 角色、行为边界、输出格式、工具使用规则
• 模型适配：微调适配特定任务、工具调用格式兼容

代码示例：基础 LLM 初始化

from openai import OpenAI

client = OpenAI()

def init_llm(model: str = "gpt-4o"):

    return {

        "model": model,

        "max_tokens": 4096,

        "temperature": 0.7

    }

# 2. 工具系统（与外部世界交互的接口）

将外部能力封装为可被 LLM 调用的函数，是 Agent 从「文本生成器」变为「执行者」的关键。

工具分类：

• 内置工具：代码执行、文件读写、计算器等
• 外部工具：搜索引擎、数据库查询、API 调用等
• 第三方工具：GitHub、Slack、Notion 等平台集成

工具设计原则：

• 职责单一：每个工具只完成一个明确功能
• 描述精确：让 LLM 能准确理解工具用途和参数要求
• 错误友好：返回清晰的错误信息，便于 LLM 重试

代码示例：工具定义与注册

tools = [

    {

        "type": "function",

        "function": {

            "name": "search_web",

            "description": "搜索互联网上的最新信息",

            "parameters": {

                "type": "object",

                "properties": {

                    "query": {"type": "string", "description": "搜索关键词，越具体越好"}

                },

                "required": ["query"]

            }

        }

    },

    {

        "type": "function",

        "function": {

            "name": "execute_python",

            "description": "执行Python代码并返回结果",

            "parameters": {

                "type": "object",

                "properties": {

                    "code": {"type": "string", "description": "要执行的Python代码"}

                },

                "required": ["code"]

            }

        }

    }

]

# 3. 记忆系统（维持状态与知识）

负责保存和管理 Agent 在任务执行过程中的所有信息，分为短期记忆和长期记忆。

记忆类型：

• 短期记忆：当前任务上下文，存储在 LLM 的上下文窗口中
• 长期记忆：跨任务持久化存储，通常使用向量数据库
• 实体记忆：从对话中提取的结构化关键事实

记忆管理挑战：

• 上下文窗口限制：如何在有限窗口内保留关键信息
• 记忆检索效率：如何快速找到相关记忆
• 记忆更新与一致性：如何处理过时或矛盾的记忆

# 4. 规划模块（任务拆解与调度）

负责将复杂目标拆解为可执行的步骤，并管理整个任务的执行流程。

常见规划模式：

• 反应式规划：边执行边规划（ReAct 模式）
• 预规划：先制定完整计划再执行（Plan-and-Execute 模式）
• 动态规划：根据执行结果实时调整计划

# 3. Workflow，Agent，Tools 这三个的概念和区别介绍一下？

# 面试题基础回答

我理解这三个概念是粒度从小到大的三层结构。

Tools 是最小的能力单元，就是封装好的可调用函数，比如搜索、执行代码、发邮件，它只负责「执行」，本身没有任何决策能力。

Agent 是一个完整的决策系统，内部用 LLM 做大脑，自己判断什么时候调哪个 Tool、要不要继续、什么时候结束，是主动的。

Workflow 是更上层的编排框架，把 Agent、LLM、Tools 组织成一条确定性流程，每个节点做什么、按什么顺序流转都是开发者事先写死的。

三者最核心的区别就一句话：Tools 不做决策只执行，Agent 自己做决策，Workflow 是开发者替所有节点把决策提前写好。

# 深入原理：三层结构的完整解析

# 1. Tools：最小能力单元

本质：封装好的可调用函数，为 Agent 提供特定能力
特点：

• 被动等待调用，不具备自主决策能力
• 有明确的输入输出格式
• 是整个系统的基础能力积木

常见工具类型：

• 信息获取工具：搜索引擎、数据库查询、API 调用
• 操作执行工具：代码执行、文件读写、邮件发送
• 计算工具：计算器、数据分析、机器学习模型调用

# 2. Agent：自主决策系统

本质：基于 LLM 的自主决策系统，能够感知环境、制定计划、调用工具完成任务
特点：

• 主动决策下一步行动
• 维持任务状态
• 可以调用多个工具完成复杂任务

Agent 与单工具调用的区别：

• 单工具调用：一次只调用一个工具，不维护状态
• Agent：多轮工具调用，维护完整任务状态，具备自主决策能力

# 3. Workflow：任务编排框架

本质：将多个 Agent、LLM、Tools 组织成确定性流程的框架
特点：

• 流程由开发者预先定义，确定性强
• 可以处理复杂的多步骤任务
• 提供统一的错误处理和监控机制

常见工作流模式：

• 顺序流：按固定顺序执行多个步骤
• 并行流：同时执行多个独立步骤
• 条件分支流：根据执行结果选择不同路径
• 循环流：重复执行某些步骤直到满足条件

# 三者的嵌套关系

三者不是互斥的，而是可以相互嵌套组合：

• 一个 Workflow 可以包含多个 Agent 和 Tools
• 一个 Agent 可以包含多个 Tools
• 一个 Workflow 可以调用另一个 Workflow

实际工程示例：

# 一个简单的工作流示例：先搜索信息，再生成报告

from langgraph.graph import Graph, END

def search_step(state):

    # 调用搜索 Agent

    return {"search_result": call_agent("search", state["query"])}

def write_report_step(state):

    # 调用写作 Agent

    return {"report": call_agent("write", state["search_result"])}

workflow = Graph()

workflow.add_node("search", search_step)

workflow.add_node("write_report", write_report_step)

workflow.add_edge("search", "write_report")

workflow.add_edge("write_report", END)

workflow.set_entry_point("search")

# 4. 了解哪些其他的 Agent 设计范式？Agent 和 Workflow 的区别是什么？

# 面试题基础回答

常见的 Agent 设计范式有 ReAct、Plan-and-Execute、Reflection 这几种。

Agent 和 Workflow 最核心的区别是「谁来决定下一步」。Workflow 是提前把流程写死的，每一步怎么走都是固定的，确定性高、好控制；Agent 是让 LLM 自己决定下一步做什么，灵活但不可控。

# 深入原理：主流 Agent 设计范式

# 1. ReAct（Reasoning + Acting）

核心思想：交替进行思考、行动、观察，形成闭环
执行流程：

1. Thought：分析当前情况，决定下一步行动
2. Action：调用工具执行决策
3. Observation：接收工具返回结果
4. 重复直到任务完成

优点：

• 实现简单，灵活性高
• 推理过程可见，便于调试
• 能够动态适应环境变化

缺点：

• 容易出现循环漂移（忘记初始目标）
• 错误容易传播
• 长任务中表现不佳

# 2. Plan-and-Execute

核心思想：先制定完整计划，再按计划逐步执行
执行流程：

1. 规划阶段：将目标拆解为步骤列表
2. 执行阶段：按步骤执行，每步更新状态
3. 重规划阶段：根据执行结果调整后续计划

优点：

• 全局视野清晰，不容易跑偏
• 便于并行执行子任务
• 可预测性强，便于监控

缺点：

• 初始计划可能不符合实际情况
• 动态调整增加复杂度
• 规划阶段额外的模型调用成本

# 3. Reflection（反思机制）

核心思想：在执行后进行自我评估，修正错误，提升质量
执行流程：

1. 执行任务步骤
2. 评估执行结果
3. 如果不满足要求则重试或调整

优点：

• 显著提升输出质量
• 减少错误和幻觉
• 适合对质量要求高的场景

缺点：

• 增加模型调用次数和成本
• 可能陷入无限循环

# Agent 与 Workflow 的本质区别

# 5. Agent 推理模式有哪些？ReAct 是啥？具体是怎么实现的？

# 面试题基础回答

Agent 的推理模式最基础的是直接输出答案，没有中间推理；CoT 是让 LLM 先把推理过程写出来再给答案，准确率更高；ReAct 是在 CoT 基础上加了「行动」，让 LLM 交替输出思考和工具调用，每次行动后再根据结果继续思考，形成一个循环。

# 深入原理：推理模式全解析

# 1. 直接输出模式

特点：LLM 直接生成最终答案，没有中间推理过程
适用场景：简单问答、内容生成等
缺点：多步推理任务容易出错，容易产生幻觉

# 2. CoT（Chain of Thought，思维链）

核心思想：让 LLM 显式地写出推理步骤，再生成最终答案
触发方式：

• Zero-shot CoT：在 prompt 末尾加「请一步步思考」
• Few-shot CoT：提供带有推理过程的示例

优点：

• 显著提升多步推理任务的准确率
• 推理过程可见，便于调试

缺点：

• 仅文本推理，无法与外部环境交互
• 单条推理链，容易走错方向

# 3. ReAct（Reasoning + Acting）

核心思想：将推理和行动结合，形成「思考 - 行动 - 观察」的闭环
与 CoT 的区别：CoT 仅生成文本推理，ReAct 可以调用工具并根据返回结果继续推理

完整实现流程：

import openai

import json

# 定义工具

def search_web(query):

    # 模拟搜索引擎调用

    return f"搜索结果：{query}的最新信息（2024年）"

def calculate(expr):

    # 模拟计算器调用

    try:

        return str(eval(expr))

    except:

        return "表达式计算错误"

# 工具列表

tools = [

    {

        "type": "function",

        "function": {

            "name": "search_web",

            "description": "搜索互联网信息",

            "parameters": {"query": "string"}

        }

    },

    {

        "type": "function",

        "function": {

            "name": "calculate",

            "description": "计算数学表达式",

            "parameters": {"expr": "string"}

        }

    }

]

def react_agent(question: str, max_steps: int = 10):

    client = openai.OpenAI()

    messages = [{"role": "user", "content": question}]

    for step in range(max_steps):

        # 调用 LLM 获取响应

        response = client.chat.completions.create(

            model="gpt-4o",

            messages=messages,

            tools=tools,

            tool_choice="auto"

        )

        response_message = response.choices[0].message

        messages.append(response_message)

        # 检查是否完成任务

        if not response_message.tool_calls:

            return response_message.content

        # 执行工具调用

        for tool_call in response_message.tool_calls:

            function_name = tool_call.function.name

            function_args = json.loads(tool_call.function.arguments)

            # 执行工具

            if function_name == "search_web":

                tool_result = search_web(function_args["query"])

            elif function_name == "calculate":

                tool_result = calculate(function_args["expr"])

            else:

                tool_result = f"错误：未知工具 {function_name}"

            # 添加工具返回结果到消息

            messages.append({

                "role": "tool",

                "tool_call_id": tool_call.id,

                "content": tool_result

            })

    return "任务超时，未完成"

# 测试

print(react_agent("2024年中国GDP增长率是多少？和2023年相比如何？"))

关键特点：

• 循环不是 LLM 自动执行的，而是由代码驱动
• LLM 每次只生成思考和工具调用决策
• 代码负责工具执行和上下文管理

局限性：

• 循环漂移：容易忘记初始目标
• 错误传播：中间错误会影响后续步骤
• 长任务性能下降：上下文窗口限制

# 6. ReAct、Plan-and-Execute、Reflection 三种范式有什么核心区别？实际项目中该如何选型？

# 面试题基础回答

三者是 Agent 开发里最主流的三种设计范式，核心区别在于「决策和执行的关系」。

• ReAct 是边想边干，走一步看一步，单步迭代实时调整，灵活度最高
• Plan-and-Execute 是先想全再干，先定完整计划再分步执行，适合长流程复杂任务，不容易跑偏
• Reflection 不是独立的完整流程，而是给前两者加的「检查修正 buff」，用来提升输出质量

实际选型就看三个维度：任务复杂度、流程确定性、输出质量要求，新手入门首选 ReAct，复杂任务用 Plan-and-Execute，高要求场景再加 Reflection。

# 深入原理：范式对比与选型指南

# 核心范式对比

# 实际选型策略

1. 新手入门 / 快速验证：优先选择 ReAct，实现简单，灵活性高
2. 复杂长流程任务：优先选择 Plan-and-Execute，避免循环漂移
3. 高要求质量任务：在 ReAct 或 Plan-and-Execute 基础上添加 Reflection
4. 混合使用：实际项目中通常混合使用多种范式

示例选型场景：

• 简单客服问答：ReAct
• 代码生成：Plan-and-Execute + Reflection
• 科研文献综述：Plan-and-Execute + Reflection
• 自动化测试：ReAct + Reflection

# 7. 复杂任务怎么做的任务拆分？为什么要拆分？效果如何提升？

# 面试题基础回答

任务拆分的原因是 LLM 一次性处理太复杂的任务很容易出错，把大任务拆成小步骤，每步聚焦一件事，准确率会明显提升。

拆分方式主要有两种：一种是静态拆分，提前把步骤写死；另一种是动态拆分，让 LLM 自己根据目标规划步骤，更灵活但也更难控制。

拆完之后步骤之间可能有依赖关系，识别出哪些可以并行，端到端延迟能降很多，有时能降 40% 到 60%。

# 深入原理：任务拆分的完整方法论

# 为什么需要任务拆分

1. 上下文窗口限制：LLM 的上下文窗口有限，复杂任务的信息量容易超出限制
2. 降低认知负荷：LLM 同时处理多个任务容易出错，拆分后每步专注一件事
3. 便于调试和错误定位：拆分后可以单独验证每个步骤，更容易找到问题所在
4. 支持并行执行：独立的子任务可以并行处理，降低总耗时
5. 提升可维护性：拆分后的步骤更易于理解和修改

# 任务拆分的两种思路

# 1. 静态拆分

定义：开发者预先定义任务的执行步骤和顺序
特点：

• 实现简单，行为可预测
• 灵活性低，无法适应未预料的情况
• 适合流程固定的任务

示例：

# 静态拆分的报告生成任务

def generate_report(query):

    # 步骤 1：搜索信息

    search_result = search_web(query)

    # 步骤 2：分析信息

    analysis = analyze_data(search_result)

    # 步骤 3：撰写报告

    report = write_report(analysis)

    return report

# 2. 动态拆分

定义：让 LLM 根据目标动态规划执行步骤
特点：

• 灵活性高，适应各种情况
• 实现复杂，行为不可预测
• 适合复杂、不确定的任务

示例：

# 动态拆分的报告生成任务

def dynamic_generate_report(query):

    # 让 LLM 规划步骤

    plan = llm_call(f"请为任务'{query}'制定执行步骤")

    steps = parse_plan(plan)

    # 按步骤执行

    results = []

    for step in steps:

        result = execute_step(step, results)

        results.append(result)

    # 汇总结果

    final_report = summarize_results(results)

    return final_report

# 任务拆分的最佳实践

1. 原子步骤原则：每个步骤只做一件独立的事，边界清晰
2. 依赖分析：识别步骤之间的依赖关系，确定执行顺序
3. 并行优化：将无依赖的步骤并行执行，降低总耗时
4. 粒度适中：既不要太细（增加 overhead）也不要太粗（失去拆分优势）
5. 动态调整：根据执行结果实时调整拆分策略

# 并行执行优化

原理：识别无依赖的子任务，并行执行以降低总耗时
示例：

import asyncio

async def parallel_task_execution(tasks):

    # 并行执行多个独立任务

    results = await asyncio.gather(*[execute_task(task) for task in tasks])

    return results

# 示例：并行搜索多个信息源

tasks = ["搜索AI趋势", "搜索大模型进展", "搜索Agent应用"]

results = asyncio.run(parallel_task_execution(tasks))

效果提升：

• 端到端延迟可降低 40%-60%
• 资源利用率提升
• 总处理时间减少

# 8. 请你介绍一下 AI Agent 的记忆机制，并说明在实际开发中应该如何设计记忆模块？

# 面试题基础回答

Agent 需要记忆才能在多步任务中保持状态、跨任务积累知识。

记忆机制分四层：感知记忆（当前输入的原始内容）、短期记忆（context window 里的对话历史）、长期记忆（存在外部数据库、语义检索召回）、实体记忆（结构化提取的关键事实）。

实际设计时要解决三个核心问题：存什么、怎么存、什么时候取出来用，根据信息类型选合适的存储方式，再搭配主动检索和按需检索两种策略使用。

# 深入原理：记忆系统全解析

# 记忆的层级结构

# 1. 感知记忆（Sensory Memory）

定义：当前调用的原始输入，是最短暂的记忆
特点：

• 生命周期仅为单次调用
• 处理完成后即被丢弃
• 作为信息进入系统的入口

# 2. 短期记忆（Short-term Memory）

定义：维持当前任务执行状态的记忆，存储在 LLM 的上下文窗口中
特点：

• 容量受限于 LLM 的上下文窗口
• 随任务完成而被清除
• 包含用户输入、模型输出、工具返回结果

实现方式：

class ShortTermMemory:

    def __init__(self):

        self.messages = []

    def add(self, role: str, content: str):

        self.messages.append({"role": role, "content": content})

    def get_context(self):

        return self.messages

    def clear(self):

        self.messages = []

# 3. 长期记忆（Long-term Memory）

定义：跨任务持久化存储的记忆，用于保存对未来任务有价值的信息
特点：

• 容量几乎无限
• 任务完成后仍保留
• 需要通过语义检索获取

技术实现：

• Embedding：将文本转换为向量表示
• 向量数据库：存储和检索向量数据（Chroma, Pinecone, Weaviate）

示例代码：

import chromadb

from openai import OpenAI

client = OpenAI()

db = chromadb.Client()

collection = db.get_or_create_collection("agent_memory")

def save_long_term_memory(content: str, metadata: dict = None):

    # 生成 Embedding

    embedding = client.embeddings.create(

        input=content,

        model="text-embedding-3-small"

    ).data[0].embedding

    # 存储到向量数据库

    collection.add(

        embeddings=[embedding],

        documents=[content],

        metadatas=[metadata or {}],

        ids=[f"mem_{hash(content)}"]

    )

def retrieve_long_term_memory(query: str, top_k: int = 3):

    # 生成查询 Embedding

    query_embedding = client.embeddings.create(

        input=query,

        model="text-embedding-3-small"

    ).data[0].embedding

    # 检索相关记忆

    results = collection.query(

        query_embeddings=[query_embedding],

        n_results=top_k

    )

    return results["documents"][0]

# 4. 实体记忆（Entity Memory）

定义：从对话中提取的结构化关键事实
特点：

• 信息密度高
• 查询速度快
• 不受原始表述方式影响

示例：

# 提取用户偏好作为实体记忆

user_preferences = {

    "language": "Python",

    "code_style": "简洁",

    "comment_style": "英文注释"

}

# 记忆系统设计的核心问题

# 1. 存什么？

判断标准：这条信息对未来任务是否有价值
值得存储的内容：

• 用户偏好和习惯
• 任务执行中的关键结论和决策
• 外部知识和文档

不值得存储的内容：

• 中间推理过程
• 工具返回的原始数据
• 闲聊内容

# 2. 怎么存？

混合存储策略：

• 结构化数据：关系数据库或 Key-Value 存储
• 非结构化数据：向量数据库
• 关键实体：单独存储或与向量数据关联

# 3. 什么时候取？

两种检索策略：

1. 主动检索：任务开始前，用任务描述检索相关记忆，注入系统提示
2. 按需检索：任务执行中，LLM 根据需要主动调用检索工具

实际工程组合：

• 任务开始前主动检索用户偏好和背景知识
• 执行中按需检索专业知识和历史数据

# 9. Agent 的长短期记忆系统怎么做的？记忆是怎么存的？粒度是多少？怎么用的？

# 面试题基础回答

记忆系统分两层。

短期记忆就是 context window 里的对话历史，存当前任务的中间状态，任务结束就清掉；长期记忆用向量数据库存，把信息 embedding 后写入，用的时候做语义检索拿回来注入 prompt。

粒度上通常按「一次完整交互」或「一个关键事件」为单位存，太细碎检索噪音大，太粗糙又丢失细节，需要根据业务实际调整。

# 深入原理：长短期记忆系统详解

# 短期记忆系统

核心作用：维持当前任务的执行状态
实现原理：维护一个消息列表，包含用户输入、模型输出、工具返回结果
特点：

• 随任务结束而清除
• 容量受限于 LLM 的上下文窗口
• 每次调用 LLM 都需要传递完整的短期记忆

上下文管理挑战：

• 窗口溢出：长任务中上下文容易超出限制
• 信息丢失：早期重要信息可能被截断

解决方案：

1. 滑动窗口：只保留最近 N 轮对话
2. 摘要压缩：将早期对话压缩为摘要
3. 记忆卸载：将不常用的信息存入长期记忆

# 长期记忆系统

核心作用：跨任务保存知识和经验
技术栈：

• Embedding 模型：text-embedding-3-small, BGE, M3E
• 向量数据库：Chroma, Pinecone, Weaviate, Milvus
• 检索算法：近似最近邻搜索（ANN）

存储流程：

1. 内容预处理：清洗、格式化
2. 生成 Embedding：将文本转换为向量
3. 存储向量和元数据：保存到向量数据库
4. 检索：根据查询生成向量，查找相似内容

# 记忆粒度设计

粒度选择的权衡：

• 太细：检索噪音大，碎片化信息多
• 太粗：信息丢失严重，无法获取细节

最佳实践：

1. 一次完整交互：用户的一个具体请求加上处理结果
2. 独立知识点：用户偏好、关键结论、决策点
3. 实体记忆：结构化提取的关键事实

示例粒度选择：

• 好的粒度：「用户偏好 Python 语言，代码风格简洁」
• 不好的粒度：完整的 5000 字对话历史

# 记忆系统的完整工作流程

1. 任务开始前：

   • 检索长期记忆，获取用户偏好和历史信息
   • 将检索结果注入系统提示

2. 任务执行中：

   • 维护短期记忆，记录每一步的输入输出
   • 按需调用检索工具，获取相关知识

3. 任务结束后：

   • 提取关键结论和决策，存入长期记忆
   • 清除短期记忆，准备下一个任务

代码示例：

def run_agent_with_memory(user_request: str):

    # 步骤 1：检索长期记忆

    relevant_memories = retrieve_long_term_memory(user_request)

    # 步骤 2：初始化短期记忆

    short_memory = ShortTermMemory()

    short_memory.add("system", f"用户偏好：{relevant_memories}")

    short_memory.add("user", user_request)

    # 步骤 3：执行任务

    result = execute_task(short_memory.get_context())

    # 步骤 4：保存关键信息到长期记忆

    save_long_term_memory(f"用户请求：{user_request}，结果：{result}")

    return result

# 10. 什么是 Multi-Agent？

# 面试题基础回答

多智能体系统（Multi-Agent）就是多个 Agent 协作完成任务，每个 Agent 各有分工，有的负责搜索、有的负责写代码、有的负责做评审。

单个 Agent 主要受两个限制：一是 context 窗口大小限制，复杂任务信息量一多就撑爆了；二是单点能力，什么都让一个 Agent 做，每件事都是泛才。

Multi-Agent 通过专业分工和并行执行，能处理更复杂、更长流程的任务。

# 深入原理：多智能体系统全解析

# 核心动机与价值

# 1. 上下文窗口限制

单个 Agent 的上下文窗口有限，复杂任务的大量信息无法一次性装入

# 2. 单点能力瓶颈

单个 Agent 无法兼顾多种专业能力，如同时精通搜索、代码、测试等

# 3. 并行执行需求

多个独立子任务可以并行处理，提升整体效率

# 4. 专业分工优势

不同 Agent 专注于特定领域，提升任务完成质量

# 多智能体系统的核心优势

1. 处理更复杂任务：可以完成远超单个 Agent 能力范围的复杂任务
2. 专业分工：每个 Agent 专注于特定领域，提升任务质量
3. 并行执行：多个 Agent 可以同时处理不同子任务，提升效率
4. 容错性：单个 Agent 失败不会导致整个系统崩溃
5. 可扩展性：可以方便地添加新的 Agent 类型

# 多智能体系统的挑战

1. 协作复杂度：Agent 之间的通信和协调变得复杂
2. 一致性问题：多个 Agent 之间的状态同步
3. 调试难度：分布式系统的调试比单个系统更困难
4. 性能开销：Agent 之间的通信和协调会带来额外开销

# 典型应用场景

1. 软件开发：需求分析、代码编写、代码审查、测试等不同角色的 Agent
2. 科研辅助：文献搜索、数据分析、结果整理等 Agent
3. 智能客服：问题分类、知识检索、工单处理、人工转接等 Agent
4. 自动驾驶：感知、规划、控制、决策等不同模块的 Agent
5. 金融交易：市场分析、风险评估、交易执行、监控等 Agent

# 多智能体系统的发展趋势

1. 标准化通信协议：如 Google 的 A2A 协议
2. 跨框架协作：不同框架开发的 Agent 之间的互操作
3. 自主协作：Agent 能够自主发现和协作伙伴
4. 人机协作：人类与多个 Agent 的协同工作

# 11. 说说 Single-Agent 和 Multi-Agent 的设计方案？

# 面试题基础回答

Single-Agent 适合任务流程清晰、复杂度适中的场景，实现简单、好维护；Multi-Agent 适合需要专业分工、任务量大或者需要并行执行的复杂场景。

Multi-Agent 架构上主要有两种拓扑：

• 中心化方案：由一个主 Agent（Orchestrator）统一调度各个 Worker，工程上用得更多，好控制、好调试
• 去中心化方案：Agent 之间直接通信，听起来灵活但工程实践中很少用，存在任务分配协调、执行顺序保证、失败感知等问题

# 深入原理：单智能体与多智能体设计

# Single-Agent 设计

# 核心特点

• 单一 LLM 作为决策核心
• 所有任务由单个 Agent 处理
• 实现简单，链路清晰
• 调试和监控方便

# 适用场景

1. 任务流程清晰、复杂度适中
2. 不需要多种专业能力
3. 快速原型开发和验证
4. 小型项目和个人使用

# 优点

• 架构简单，易于实现
• 调试和监控方便
• 通信开销小
• 行为可预测

# 缺点

• 上下文窗口限制
• 单点能力瓶颈
• 无法并行处理任务
• 难以处理复杂任务

# Multi-Agent 设计

# 中心化架构（Orchestrator 模式）

核心组件：

1. Orchestrator：总调度者，负责任务分配和结果汇总
2. Worker Agents：执行具体任务的 Agent

工作流程：

1. Orchestrator 接收用户任务
2. 将任务拆分为多个子任务
3. 将子任务分配给合适的 Worker Agent
4. 收集 Worker 的结果
5. 汇总结果并返回给用户

优点：

• 调度清晰，易于控制
• 调试和监控方便
• 任务分配明确

缺点：

• Orchestrator 成为单点瓶颈
• 灵活性相对较低
• Worker 之间无法直接通信

代码示例：

class CentralizedMultiAgentSystem:

    def __init__(self):

        self.agents = {

            "search": SearchAgent(),

            "code": CodeAgent(),

            "review": ReviewAgent()

        }

    def route_task(self, task_type: str):

        # 根据任务类型路由到合适的 Agent

        return self.agents.get(task_type, DefaultAgent())

    def execute_task(self, user_request: str):

        # 分析任务类型

        task_types = self.analyze_task(user_request)

        # 分配任务并收集结果

        results = {}

        for task_type in task_types:

            agent = self.route_task(task_type)

            results[task_type] = agent.run(task_types[task_type])

        # 汇总结果

        return self.summarize_results(results)

# 去中心化架构（Peer-to-Peer）

核心特点：没有中央调度者，Agent 之间直接通信
工作流程：

1. 每个 Agent 独立工作
2. Agent 之间直接发送消息和请求
3. 任务通过 Agent 之间的协商完成

优点：

• 灵活性高
• 没有单点瓶颈
• 易于扩展

缺点：

• 调试和监控困难
• 任务协调复杂
• 容易出现死循环
• 行为不可预测

适用场景：

• 学术研究场景
• 高度自治的分布式系统
• 不需要集中控制的场景

# 选型决策指南

实际工程建议：

1. 先从 Single-Agent 开始，验证需求
2. 当 Single-Agent 无法满足需求时，再演进到 Multi-Agent
3. 优先选择中心化架构，便于控制和调试
4. 去中心化架构仅在特定场景下使用

# 12. Agent 记忆压缩通常有哪些方法？

# 面试题基础回答

记忆压缩常见有四种方法：

1. 滑动窗口：只保留最近 N 轮对话，最简单但会硬截断关键信息
2. 摘要压缩：把长对话总结成简短摘要，丢之前先提炼一遍
3. 重要性过滤：按价值筛选记忆内容，不按时间筛选
4. 结构化抽取：把关键信息抽成结构化数据存起来，信息密度最高

工程里最常用的是摘要压缩和滑动窗口组合使用，滑动窗口控制总长度，摘要压缩保留关键信息。

# 深入原理：记忆压缩技术全解析

# 为什么需要记忆压缩

1. 上下文窗口限制：LLM 的上下文窗口有限，长任务中需要压缩历史信息
2. 降低成本：减少输入 token 数量，降低模型调用成本
3. 提升效率：减少模型处理时间，提升响应速度
4. 减少噪音：去除无关信息，提升模型决策质量

# 主流记忆压缩方法

# 1. 滑动窗口（Sliding Window）

核心思想：只保留最近 N 轮对话，更早的历史直接丢弃
实现方式：

def sliding_window(messages: list, max_length: int = 10):

    # 只保留最近 max_length 条消息

    return messages[-max_length:]

优点：

• 实现简单，无需额外模型调用
• 计算开销小

缺点：

• 硬截断，可能丢失重要信息
• 无法区分信息的重要性
• 长期记忆丢失严重

适用场景：

• 短对话场景
• 历史信息不重要的场景
• 快速原型开发

# 2. 摘要压缩（Summarization）

核心思想：将长对话历史压缩为简短摘要，保留关键信息
实现方式：

def summarize_history(messages: list, llm) -> str:

    # 调用 LLM 生成对话摘要

    prompt = f"请将以下对话历史压缩为简短摘要：\n{messages}"

    summary = llm.call(prompt)

    return summary

优点：

• 保留关键信息
• 减少 token 使用

缺点：

• 摘要过程会丢失细节
• 依赖 LLM 的摘要质量
• 额外的模型调用成本

进阶：层级式摘要：

• 最近 N 轮保留原文
• 中期历史压缩为中期摘要
• 早期历史压缩为长期摘要

# 3. 重要性过滤（Importance Filtering）

核心思想：根据信息的重要性进行筛选，而非按时间
实现方式：

1. 为每条记忆分配重要性分数
2. 过滤掉低于阈值的记忆

打分方式：

• 规则打分：包含关键词、被引用次数等
• LLM 打分：让 LLM 评估每条记忆的重要性

代码示例：

def filter_by_importance(memories: list, threshold: float = 0.5):

    # 过滤掉重要性低于阈值的记忆

    return [mem for mem in memories if mem["importance"] >= threshold]

优点：

• 保留更有价值的信息
• 避免按时间一刀切

缺点：

• 额外的打分成本
• 需要定义重要性标准

# 4. 结构化抽取（Structured Extraction）

核心思想：将关键信息提取为结构化数据，而非保留原始文本
实现方式：

1. 从对话中提取实体、关系、事实
2. 存储为结构化格式

代码示例：

def extract_structured_memory(messages: list, llm) -> dict:

    prompt = "请从以下对话中提取用户偏好和关键事实，返回JSON格式：\n{messages}"

    result = llm.call(prompt)

    return json.loads(result)

优点：

• 信息密度高
• 查询速度快
• 不受原始表述影响

缺点：

• 需要预先定义结构
• 灵活性较低
• 初始抽取成本高

# 压缩方法的组合应用

工程最佳实践：组合使用多种压缩方法

示例组合：

1. 使用滑动窗口控制总长度
2. 对早期历史使用摘要压缩
3. 对关键信息使用结构化抽取
4. 定期进行重要性过滤

实际流程：

def compressed_memory_management(messages: list, llm):

    # 步骤 1：如果超过长度限制，先摘要早期历史

    if len(messages) > 20:

        early_summary = summarize_history(messages[:-10], llm)

        messages = [early_summary] + messages[-10:]

    # 步骤 2：提取结构化记忆

    structured_mem = extract_structured_memory(messages, llm)

    # 步骤 3：重要性过滤

    filtered_mem = filter_by_importance(structured_mem)

    return filtered_mem

# 13. 在工程实践中，为什么有时候选择「手搓」Agent，而不是直接用成熟框架？

# 面试题基础回答

框架用起来快，但有几个实际痛点。

第一是抽象层太多，调试的时候不知道哪步出了问题，得一层层往下扒；第二是版本升级经常有破坏性变更，线上稳定性难保证；第三是框架的通用设计往往和具体业务需求有偏差，定制起来反而更费劲。

手搓的代码完全在自己掌控之内，可观测性好、出问题好排查，也更方便做性能优化。所以我现在的策略是核心逻辑手写，只在边缘功能上用框架的工具。

# 深入原理：框架与手搓的博弈

# 框架的价值与痛点

# 框架的优势

1. 快速开发：封装了大量样板代码，提升开发效率
2. 功能丰富：内置了工具调用、记忆管理、工作流等功能
3. 社区支持：有大量的文档、示例和社区资源
4. 持续更新：框架开发者会不断添加新功能和优化

# 框架的痛点

1. 抽象层过多：调试需要穿透多层框架代码
2. 版本兼容性：升级可能带来破坏性变更
3. 性能开销：通用设计带来不必要的性能损耗
4. 灵活性不足：难以满足高度定制化的需求
5. 黑盒问题：框架内部逻辑不透明，难以理解和调试

实际痛点案例：

• LangChain 的 AgentExecutor 在新版中被废弃，导致大量代码需要重构
• 框架内部的隐性逻辑导致难以定位的 bug
• 性能敏感场景下框架的额外开销

# 手搓 Agent 的优势与挑战

# 手搓的优势

1. 完全掌控：所有代码都在自己的掌控之中
2. 链路透明：可以在任意位置添加日志和监控
3. 性能优化：可以针对具体场景进行极致优化
4. 稳定性高：不受框架升级的影响
5. 定制灵活：可以完全按照需求定制功能

# 手搓的挑战

1. 开发效率低：需要编写大量样板代码
2. 需要专业知识：需要深入理解 Agent 的各个组件
3. 维护成本高：需要自己维护所有功能
4. 容易重复造轮子：可能实现框架已经有的功能

# 工程实践策略：混合模式

最佳实践：核心逻辑手写，边缘功能使用框架

分层策略：

1. 核心层：手写，包括任务调度、状态管理、关键决策逻辑
2. 工具层：使用框架，包括工具调用、向量检索、基础组件
3. 监控层：手写，包括日志、监控、调试工具

实际示例：

# 手搓核心逻辑，使用 LangChain 的工具封装

from langchain.tools import DuckDuckGoSearchRun

class CustomAgent:

    def __init__(self):

        self.search_tool = DuckDuckGoSearchRun()

        self.memory = ShortTermMemory()

    def run(self, task: str):

        # 手搓核心决策逻辑

        # 使用框架的工具

        search_result = self.search_tool.run(task)

        # 手搓后续处理逻辑

        return self.process_result(search_result)

# 项目演进路径

1. 原型阶段：使用框架快速开发，验证需求
2. 验证阶段：识别瓶颈和问题点
3. 生产阶段：将核心逻辑替换为手写代码，保留边缘功能使用框架
4. 优化阶段：持续优化性能和可维护性

# 决策指南

什么时候用框架：

• 原型开发和快速验证
• 团队不熟悉 Agent 开发
• 边缘功能和工具集成

什么时候手搓：

• 核心业务逻辑和关键决策点
• 性能敏感的场景
• 需要高度定制化的功能
• 线上稳定运行的系统

# 14. 如何赋予 LLM 规划能力？

# 面试题基础回答

给 LLM 加规划能力主要靠这几种思路：

• CoT（思维链）：让 LLM 把推理步骤写出来，线性地一步步推导到答案，加一句「请一步步思考」即可
• ToT（思维树）：同时探索多条推理路径，选最优的继续深入，解决 CoT 走错方向的问题
• GoT（思维图）：图结构推理，推理节点可以复用和合并，适合更复杂的任务，目前还比较学术

工程上我用 CoT 最多，实现成本最低；ToT 效果更好但调用次数多，成本大概是 3 到 5 倍；GoT 目前还没有成熟的生产落地。

# 深入原理：LLM 规划能力全解析

# 规划能力的本质

核心问题：如何让 LLM 避免直接跳到答案，而是系统地思考问题
目标：减少推理错误，提升复杂任务的处理能力
实现路径：将隐式的推理过程显式化

# 1. CoT（Chain of Thought，思维链）

核心思想：让 LLM 显式地写出推理步骤，再生成最终答案
触发方式：

• Zero-shot CoT：在 prompt 末尾添加「请一步步思考」
• Few-shot CoT：提供带有推理过程的示例

工作原理：
LLM 的输出是顺序生成的，先写出的推理步骤会成为后续生成的上下文，帮助 LLM 避免跳步和错误。

代码示例：

def cot_prompt(question: str) -> str:

    return f"""

请解决以下问题，一步步思考：

{question}

请先写出你的推理过程，再给出最终答案。

"""

优点：

• 实现简单，几乎零成本
• 显著提升多步推理能力
• 推理过程可见，便于调试

缺点：

• 单条推理链，容易走错方向
• 无法探索多个可能的解决方案
• 长推理链中容易出现错误累积

# 2. ToT（Tree of Thoughts，思维树）

核心思想：同时探索多条推理路径，评估每个路径的质量，选择最优的继续深入
与 CoT 的区别：CoT 是单条路径，ToT 是多条路径并行探索

执行流程：

1. 生成候选思路：针对问题生成多个可能的推理方向
2. 评估思路质量：对每个思路进行评估，筛选出最有希望的
3. 深入探索：对筛选后的思路继续展开推理
4. 剪枝：丢弃质量低的路径

代码示例：

tot_prompt = """

请解决以下问题，探索多种可能的推理路径：

{question}

对于每个路径，先写出你的思路，然后评估其合理性，最后选择最优路径继续。

"""

优点：

• 避免单条路径的错误
• 可以探索多个解决方案
• 更适合复杂问题

缺点：

• 增加模型调用次数，成本上升 3-5 倍
• 增加复杂度和调试难度
• 可能陷入组合爆炸

# 3. GoT（Graph of Thoughts，思维图）

核心思想：将推理结构从树升级为图，允许不同推理路径之间的信息共享和复用
与 ToT 的区别：ToT 是树形结构，节点只有一个父节点；GoT 是图结构，节点可以有多个父节点

核心优势：

• 支持推理节点的复用和合并
• 可以建模更复杂的推理关系
• 支持多路径信息融合

当前状态：

• 主要处于学术研究阶段
• 生产环境落地较少
• 实现复杂度较高

# 4. Plan-and-Execute 模式

核心思想：将规划和执行分离，先制定完整计划，再按计划执行
与前面三种方法的关系：是一种更高层次的规划模式

执行流程：

1. 规划阶段：将目标拆解为步骤列表
2. 执行阶段：按步骤执行，每步更新状态
3. 重规划阶段：根据执行结果调整后续计划

代码示例：

def plan_and_execute(task: str, llm):

    # 步骤 1：生成计划

    plan_prompt = f"请为任务'{task}'制定详细的执行步骤："

    plan = llm.call(plan_prompt)

    steps = parse_plan(plan)

    # 步骤 2：执行计划

    results = []

    for step in steps:

        result = execute_step(step, results)

        results.append(result)

    # 步骤 3：重规划（可选）

    replan_prompt = f"原计划：{plan}\n执行结果：{results}\n是否需要调整计划？"

    need_replan = llm.call(replan_prompt)

    if need_replan:

        new_plan = llm.call(f"请调整计划：{replan_prompt}")

        steps = parse_plan(new_plan)

    return results

# 规划能力的工程实践

选型指南：

1. 简单任务：CoT，实现简单，成本低
2. 中等复杂度任务：CoT + 少量调整
3. 复杂任务：Plan-and-Execute + CoT
4. 高要求质量任务：Plan-and-Execute + CoT + Reflection

性能与成本权衡：

• CoT：成本最低，效果中等
• ToT：成本较高，效果较好
• Plan-and-Execute：成本中等，效果好
• Reflection：成本较高，效果最好

实际工程建议：

1. 从最简单的 CoT 开始，逐步升级
2. 根据任务复杂度选择合适的方法
3. 混合使用多种方法，扬长避短
4. 关注成本和性能的平衡

# 15. 讲讲 Agent 的反思机制？为什么要用反思？具体怎么实现？

# 面试题基础回答

反思机制是在 Agent 完成一步或者完成整个任务之后，再让一个 LLM 来判断做得好不好、结果是否符合预期。如果评估不通过，就重试或者换一种策略。

它能显著提升输出质量，尤其是在代码生成、文案写作这类「质量要求高但一次做对很难」的场景下效果特别明显。代价是增加了 token 消耗和延迟。

# 深入原理：反思机制全解析

# 反思机制的核心价值

解决的问题：

1. 减少幻觉和错误：LLM 容易生成看似正确但实际错误的内容
2. 提升输出质量：对生成的内容进行审查和优化
3. 自我修正：发现错误并自动修正
4. 迭代改进：逐步优化结果直到满足要求

适用场景：

• 代码生成和审查
• 文案写作和编辑
• 数据分析和验证
• 决策制定和评估
• 复杂问题解决

# 反思机制的基本架构

核心流程：生成 → 评估 → 修正 → 循环直到满足要求

┌───────────────────────────────────────────────────────────┐
│                                                       │
│  用户任务 → 生成初始输出 → 评估输出质量 → 如果不满足则修正   │
│                                                       │
└───────────────────────────────────────────────────────────┘

# 基础 Reflection 实现

代码示例：

def reflection_loop(task: str, generate_func, evaluate_func, max_retries: int = 3):

    # 初始生成

    result = generate_func(task)

    for retry in range(max_retries):

        # 评估结果

        evaluation = evaluate_func(task, result)

        # 如果满足要求，返回结果

        if evaluation["pass"]:

            return result

        # 生成修正提示

        correction_prompt = f"""

任务：{task}

初始结果：{result}

评估问题：{evaluation['feedback']}

请修正结果，解决上述问题。

"""

        # 生成修正后的结果

        result = generate_func(correction_prompt)

    # 达到最大重试次数

    return result

# Reflexion 进阶实现

核心思想：不仅评估错误，还记录错误原因和改进建议，作为后续任务的上下文

增强流程：生成 → 评估 → 记录反思 → 修正 → 循环

代码示例：

def reflexion_loop(task: str, generate_func, evaluate_func, memory=None, max_retries: int = 3):

    if memory is None:

        memory = []

    # 初始生成

    result = generate_func(task, memory)

    for retry in range(max_retries):

        # 评估结果

        evaluation = evaluate_func(task, result)

        if evaluation["pass"]:

            return result

        # 生成反思

        reflection = f"""

任务：{task}

尝试结果：{result}

评估反馈：{evaluation['feedback']}

改进建议：{evaluation['suggestions']}

"""

        memory.append(reflection)

        # 生成修正提示

        correction_prompt = f"""

任务：{task}

历史反思：{memory}

请基于历史反思修正结果。

"""

        result = generate_func(correction_prompt)

    return result, memory

# 反思机制的实际应用

代码生成场景：

def code_reflection(task: str, code: str):

    evaluation_prompt = f"""

请评估以下代码是否满足要求：

任务：{task}

代码：{code}

请检查：

1. 是否正确实现了任务要求

2. 是否存在bug和安全问题

3. 是否符合最佳实践

返回JSON格式，包含pass（布尔值）、feedback（字符串）、suggestions（数组）

"""

    evaluation = llm.call(evaluation_prompt)

    return json.loads(evaluation)

def generate_code(task: str, memory=None):

    prompt = f"请为以下任务生成Python代码：{task}"

    if memory:

        prompt += f"\n参考历史反思：{memory}"

    return llm.call(prompt)

# 使用反思机制生成代码

final_code, reflections = reflexion_loop(

    task="实现一个快速排序算法",

    generate_func=generate_code,

    evaluate_func=code_reflection,

    max_retries=3

)

# 反思机制的优化策略

1. 限制重试次数：避免无限循环，设置最大重试次数
2. 成本控制：反思会增加模型调用次数，需要平衡质量和成本
3. 针对性提示：明确指出需要改进的方向，避免模糊的反馈
4. 记忆持久化：将反思结果保存到长期记忆，用于后续任务
5. 自动化评估：对于可自动化评估的任务，使用自动化工具替代 LLM 评估

# 反思机制的挑战与局限

1. 额外成本：每次反思都需要额外的模型调用，增加成本和延迟
2. 评估偏差：评估 LLM 本身也可能产生错误和偏见
3. 循环风险：可能陷入生成 - 评估 - 修正的无限循环
4. 复杂度增加：系统变得更复杂，调试和维护难度增加

# 工程实践建议

1. 按需使用：仅在对质量要求高的场景使用反思机制
2. 分层反思：对于复杂任务，分阶段进行反思
3. 混合评估：结合自动化工具和 LLM 评估，提升效率和准确性
4. 监控与调试：添加详细的日志，便于调试和分析
5. 成本控制：设置合理的重试次数和成本上限

# 16. 如何设计多 Agent 的协作与动态切换机制？

# 面试题基础回答

协作靠两件事：

• 消息传递：Agent 完成工作后把结果发送到消息队列，下游 Agent 订阅取用，解耦性好
• 共享状态：所有 Agent 共同读写一个状态对象，记录任务进展和中间结果，直接性强

动态切换靠 Orchestrator 来做，有两种方式：

• 静态路由：提前写好规则，稳定可预测
• 动态路由：让 LLM 根据当前情况实时判断，灵活但增加成本

工程上最稳健的做法是主流程用静态路由保底，边缘情况交给 LLM 动态决策。

# 深入原理：多 Agent 协作与切换机制

# 多 Agent 协作的核心问题

1. 通信机制：Agent 之间如何传递信息
2. 任务分配：如何将任务分配给合适的 Agent
3. 状态管理：如何维护共享的任务状态
4. 协调机制：如何协调多个 Agent 的行动
5. 错误处理：如何处理 Agent 失败和异常情况

# 1. 通信机制

# 消息传递模式

核心思想：Agent 之间通过发送消息进行通信
特点：

• 解耦性好，发送方不需要知道接收方
• 需要消息中间件支持
• 适合分布式场景

实现方式：

• 消息队列：Redis、RabbitMQ、Kafka
• 发布 - 订阅模式：Agent 订阅感兴趣的主题

代码示例：

class MessageQueue:

    def __init__(self):

        self.topics = {}

    def publish(self, topic: str, message: any):

        if topic not in self.topics:

            self.topics[topic] = []

        self.topics[topic].append(message)

    def subscribe(self, topic: str) -> list:

        return self.topics.get(topic, [])

# 共享状态模式

核心思想：所有 Agent 共享一个状态对象，读写共同的状态
特点：

• 直接性强，不需要额外的中间件
• 适合紧密协作的场景
• 需要处理并发访问和数据一致性问题

实现方式：

• 内存状态：适合单进程场景
• 数据库：适合多进程分布式场景
• 分布式缓存：Redis 等

代码示例：

class SharedState:

    def __init__(self):

        self.state = {

            "task_progress": {},

            "intermediate_results": {},

            "errors": []

        }

    def update(self, key: str, value: any):

        self.state[key] = value

    def get(self, key: str) -> any:

        return self.state.get(key)

# 2. 任务分配与动态切换

# 静态路由

核心思想：提前定义任务分配规则，根据任务类型和当前状态选择合适的 Agent
特点：

• 稳定可预测，易于调试和监控
• 灵活性低，无法处理未预料的情况
• 适合流程固定的场景

代码示例：

def static_router(task_type: str) -> str:

    routes = {

        "search": "SearchAgent",

        "code": "CodeAgent",

        "review": "ReviewAgent",

        "test": "TestAgent"

    }

    return routes.get(task_type, "DefaultAgent")

# 动态路由

核心思想：让 LLM 根据当前情况动态决定任务分配给哪个 Agent
特点：

• 灵活性高，能够处理未预料的情况
• 增加额外的模型调用成本
• 行为不可预测，调试难度大

代码示例：

def dynamic_router(task_context: str, available_agents: list[str]) -> str:

    prompt = f"""

当前任务上下文：{task_context}

可用的Agent：{available_agents}

请选择最合适的Agent来处理当前任务，只返回Agent名称。

"""

    selected_agent = llm.call(prompt)

    return selected_agent.strip()

# 混合路由策略

工程最佳实践：主流程使用静态路由，边缘情况使用动态路由

优势：

• 稳定可靠，大多数情况行为可预测
• 灵活处理边缘情况和未预料的任务
• 平衡了稳定性和灵活性

实现流程：

1. 尝试使用静态路由
2. 如果没有匹配的规则，使用动态路由
3. 记录路由决策，用于后续分析和优化

# 3. 协作模式分类

# 流水线模式

核心思想：Agent 之间按固定顺序依次执行，前一个完成后交给下一个
特点：

• 流程清晰，易于理解和调试
• 灵活性低，适合固定流程的任务
• 适合顺序依赖的任务

# 层级模式

核心思想：有一个 Orchestrator 负责任务分配和结果汇总，其他 Agent 执行具体任务
特点：

• 调度清晰，易于控制
• Orchestrator 成为单点瓶颈
• 适合大多数工程场景

# 协商模式

核心思想：Agent 之间没有严格的上下级，通过协商和辩论达成一致
特点：

• 灵活性最高，适合复杂的决策场景
• 实现复杂，调试难度大
• 容易陷入僵局

# 4. 多 Agent 系统的工程实践

# 设计原则

1. 明确职责边界：每个 Agent 有清晰的职责范围
2. 最小通信：只传递必要的信息，减少通信开销
3. 状态一致性：确保多个 Agent 之间的状态同步
4. 错误处理：设计完善的错误处理和恢复机制
5. 可观测性：添加详细的日志和监控，便于调试

# 实现框架

• LangGraph：基于 LangChain 的多 Agent 框架
• CrewAI：多 Agent 协作框架
• AutoGen：微软的多 Agent 框架
• AgentScope：阿里巴巴的多 Agent 框架

# 监控与调试

1. 日志记录：记录每个 Agent 的输入、输出和决策过程
2. 可视化：使用工具可视化 Agent 之间的通信和任务流转
3. 追踪：跟踪每个任务的完整执行链路
4. metrics：收集性能指标，如响应时间、成功率等

# 5. 实际应用案例

软件开发团队：

• 需求分析 Agent：理解用户需求，生成需求文档
• 设计 Agent：生成系统设计方案
• 编码 Agent：编写代码
• 审查 Agent：代码审查
• 测试 Agent：编写测试用例，执行测试

智能客服团队：

• 路由 Agent：根据问题类型分配给合适的客服 Agent
• 知识库 Agent：检索相关知识
• 回复 Agent：生成回复内容
• 转接 Agent：复杂问题转接人工客服

科研助手团队：

• 文献搜索 Agent：搜索相关文献
• 数据分析 Agent：分析数据
• 图表生成 Agent：生成图表
• 报告撰写 Agent：撰写研究报告