AI Agent

发表于 | 分类于 | 阅读次数

目前，与AI交互遵循一种熟悉的流程。您都必须输入prompt，AI 模型会根据输入来响应。每次您想要新的输出时，您都必须提供prompt。总是有人来启动这个过程。

AI agent以不同的方式工作。他们被设计为独立思考和行动。您唯一需要提供的就是一个目标。他们将根据环境的反馈和自己的内心独白生成一个任务列表并开始工作。就好像AI Agent可以自我提示，不断发展和适应，可以在存在大量新信息的不可预测的环境中工作，以尽可能最好的方式实现他们的目标。

AutoGPT

https://aiedge.medium.com/autogpt-forge-e3de53cc58ec

img

Limitations

  • Wrong tool selection
  • Infinite loops 无限循环
  • Hallucinations

Why agent

  • LLM 是静态的文本生成器
  • Agent 是具备规划、记忆、行动能力的自治系统
  • 企业想要的不是聊天机器人，而是能真正 替人干活的数字员工
  • 这种能力必须依赖 Planning / Memory / Tool Use / Function Calling
  • Agent 技术才是企业真正愿意投入预算的方向

LLM 本质是“被动生成式”的。它能写计划、能回答问题，但缺乏：

  • 可执行能力（Execute）
  • 状态意识（Environment State）
  • 错误感知能力（Feedback Loop）

CoT 的核心问题：

  • 只能生成“思维链”，无法执行真实动作
  • 没有环境反馈，思考永远不会修正
  • 只是一段文本，无法保证结构化输出

ReAct 的核心：让模型先思考，再行动，再根据反馈继续思考:Thought → Action → Observation → Thought → Action

Profile

当我们人类专注于各种任务时，我们会为这些任务调整自己的状态。无论是写作、切菜、驾驶还是进行体育运动，我们都会集中注意力，甚至采取不同的心态。当讨论Agent时，概念上的profile指的就是这种适应性。研究表明，仅仅告知一个Agent程序它在某个特定任务上是专家，就能提高它的性能。

profile模块具有超越仅仅优化提示的潜在应用。它可以用于调整Agent程序的记忆功能、可用动作，甚至是驱动Agent程序的底层大型语言模型（LLM）。

Memory

对于一个机器人来说，记忆不仅仅是存储，它是构建其身份、能力和学习的基础。正如我们的记忆影响我们的决策、反应甚至个性一样，机器人的记忆是其过去互动、学习和反馈的积累。记忆主要分为长期记忆和短期记忆。

  • 长期记忆类似于机器人的基础知识，为 Agent 提供长时间保留和回忆信息的能力，这个时候需要借助外部问量存储和快速检索来实现(向量数据库)

  • 短期记忆（或工作记忆）关注的是即时的事务，处理短暂记忆，就像我们对最近事件的回忆一样。虽然对于实时任务至关重要，但并不是所有的短期记忆都能进入机器人的长期存储。（Prompt Engineering，上下文）

在这个领域出现了一个新兴的概念，即记忆反思。在这里，机器人不仅仅是存储记忆，还会主动回顾它们。这种内省使机器人能够重新评估、优先处理或甚至丢弃信息，就像人类回忆和从过去经验中学习一样。

Planning

规划是机器人解决问题的路线图。当面对复杂的挑战时，人类本能地将其分解为可管理的小任务，这种策略也被镜像在基于LLM的机器人中。这种有条不紊的方法使机器人能够以结构化的思维方式解决问题，确保全面而系统的解决方案。

机器人的规划工具包中有两种主要策略。

  • 第一种是带反馈的规划，这是一种自适应的方法。在这种方法中，机器人根据结果来优化其策略，就像根据用户反馈不断迭代设计版本一样。

  • 第二种是不带反馈的规划，将机器人视为一名策略家，仅依靠其现有知识和远见。将复杂任务分解为更小、更易于处理的子目标,从而实现对复杂任务的高效处理。

在大多数现有的 Agent 框架中，所谓的 “Planning（规划）”，并不是 LLM 自发产生的，而是通过 Prompt 或程序结构人工嵌入进去的。

换句话说，现在的 LLM，并不会自动思考“下一步该干什么”，而是我们先告诉它该怎么想、怎么拆、怎么执行。

场景LLM 自行规划（Prompt）

  • 问题不确定、场景多变：比如要写营销方案、生成创意文案、制定调研计划，这些任务本身没有固定解，模型发挥空间更大。
  • 希望模型具备灵活应变能力：当需求常变、数据来源复杂时，手写规则成本太高，不如让模型按原则生成方案。
  • 人力维护成本高：如果每天都要手动改规则，不如让模型通过 Prompt 自己“思考”并给出步骤。

场景人工或程序硬编码

  • 流程固定，可控性要求高：比如实名认证、法务审核、风控决策。这些环节要的是“绝对可靠”，不是灵活。
  • 不需要模型推理的逻辑：例如按钮点击、表单验证、API 调用流程，这类纯逻辑性代码交给程序更安全。
  • 安全与合规要求高：模型可能输出不合规的答案，而硬编码逻辑可以被审计、被复盘。

生产混合式规划：

  1. 主流程由人工或代码固定；确保关键步骤不会跑偏。
  2. 关键节点交给模型规划；比如内容生成、策略制定、风险分析等。
  3. Prompt 嵌入规则和约束条件；例如告诉模型“遇到非法输入要直接返回失败”，而不是随意发挥。
  4. LLM 规划后再经人工/程序校验；提高安全性与可解释性。

目前主流 Agent 框架中的规划（Planning），主要是通过 Prompt 模板 + 程序循环实现的。 LLM 负责局部的推理和决策，但总体流程由人工定义。

Action

在回忆和规划之后，最终到来的是行动。这是Agent认知过程转化为实际结果的阶段，运用Agent的能力。每个决策、每个思考都在行动阶段得到体现，将抽象概念转化为明确的成果。

无论是写下回应、保存文件还是启动新的流程，行动是Agent决策之旅的关键。它连接着数字认知和真实世界的影响，将Agent的电子冲动转化为有意义而有目标的结果。

Function call

难点不在工具本身，而在“决策”，模型到底什么时候调用、调用哪个、调用顺序是什么、缺信息时要不要追问、多轮对话怎么推进。

  • Prompt 本质是“规则”，无法覆盖分支逻辑，业务流程，不是自然语言能稳定表达的。
  • prompt 只能告诉模型“请调用工具”， 但无法让模型真正理解工具之间的依赖关系。
  • Prompt 不能让模型学习“追问逻辑”与“信息补全流程

提高调用采样率

常见错误

  • tool shema （函数名模糊、参数名相似、含糊不清）
    • 参数唯一
    • 功能无重叠
    • 语义明确
    • 格式严格
    • 例子清晰
  • Prompt 或上下文有歧义
    • 没告诉模型什么时候必须调用工具
    • 没告诉模型什么时候不能调用工具
    • 没要求模型必须先思考再调用
    • 没明确参数必须 JSON 格式且严格遵守 Schema
    • 没告诉它错误时要反思和修正
  • 模型采样策略太高 ，自由度太高 （或者模型太烂）
  • 没有防御机制：调用错了、JSON 格式不对、工具返回异常 等

Schema 设计

  • Schema 必须做成“无歧义、无默认、无模糊”的结构。

Prompt 或上下文有歧义

  • Prompt 中必须显式告诉模型“不能猜”，没有得到的信息不要自己猜，不要推断，必须向用户追问。
  • 追问机制是减少幻觉的唯一“补全手段”

动态函数

  • 意图识别 给与 工具子集，不然token 压力增大，决策空间更大、错误概率变高

CoT + Plan-Execute（强制模型按步骤来， 复杂任务不拆解＝必挂。）

  • 先plan, 逐步执行，每一步都带观察值（Observation），错了就 ReAct 自我纠正。

结果校验层（Result Validation Layer）

  • 参数是否完整？缺失参数 → 反馈错误 → Retry
  • JSON 格式是否符合 schema？：字段类型不对 → 清洗 / 重新生成
  • API 返回是否可用？ ： 429 → Retry（指数退避）401 → 重新认证，数据格式异常 → fallback

Memory（history） + 变量注入

1
2
3
4
5
当前会话信息：
- 出发地：
- 目的地：
- 出发日期：
- 行程天数：

日志驱动优化（Log-driven）

1
2
3
4
5
6
7
8
9
10
(query, tool_chosen, params, error_type)
再按 error_type 聚类：

参数缺失？ → schema 不明确

选错函数？ → 路由需要改

JSON 拼错？ → 思考链不完整

API 格式问题？ → 加清洗层

训练function call

Function Call 训练的目标不是让模型“会调用工具”，而是让它“根据业务逻辑正确调用工具”。

工具选择能力

  • 模型要学会： 什么场景选什么工具，什么场景不选工具。

参数绑定能力

  • 能正确解析用户自然语言里的参数

格式化能力（JSON / Schema）

  • 很多模型“工具选对了”，但 JSON 拼错了： 缺引号、少括号、值类型错误…，训练集要让它习惯按 schema 生成。

多轮上下文能力

  • 训练集必须告诉模型：如何利用记忆 / 上下文补全参数。

使用正负样本

1
2
3
4
5
6
7
8
9
10
User: 帮我查一下明天从上海到北京的机票。
Assistant:
{
  "tool": "search_flights",
  "args": {
    "origin": "上海",
    "destination": "北京",
    "date": "2023-10-30"
  }
}

负

1
2
Assistant: （不给工具调用）
北京未来三天的天气较冷...

困难例子

1
2
3
4
5
6
7
8
9
10
User: 帮我查一下后天魔都飞帝都的机票。
Assistant (tool_call):
{
  "tool": "search_flights",
  "args": {
    "origin": "上海",
    "destination": "北京",
    "date": "2023-10-31"
  }
}

训练数据

训练数据生成

多轮对话

构建一个“沙盒式数据生成系统”，把所有分支、变量、流程在数据层面定义清楚，然后一次性生成全量覆盖的数据集。

  • 用模板 / LLM 自动生成大量样本。

  • 真实对话日志（Real Logs）

  • 规则生成的边界样本（Rule-based Hard Cases）

    用于强化：

    • 日期边界（跨月、模糊日期）
    • 模糊表达（“明早”“后天晚上”）
    • 用户错别字、不完整句子
    • tool schema 边界值
    • 参数缺失、参数冲突
    • 模拟工具调用链

  • Badcase 驱动（Error-driven）样本

    • 每次模型调用失败，就把失败场景反向加入训练集。

比较通用的比例搭配：

  • 正样本：60% 正样本教模型工具的标准调用方式；
  • 难例：30%
  • 负样本：10% : 负样本用于告诉模型哪些场景不能调用工具。

全部样本通过合成数据、真实日志和 Badcase 驱动构建，形成持续闭环。

对成熟系统来说，难例比例会逐渐提高，因为边界错误会越来越多。

样本规模一般取决于工具数量和业务复杂度：

  • 简单业务：几千条
  • 中等复杂度（航旅、电商）：1–5 万条
  • 多轮 Agent（企业级助手）：10 万条以上

AutoGen

AutoGen 是一个框架，可以使用多个代理程序进行交流来解决任务，实现 LLM 应用的开发。AutoGen 代理程序可定制，可对话，并且能够无缝地与人类参与结合起来。它们可以以不同的方式运行，结合了 LLM、人类输入和工具的各种组合模式。

AutoGen Overview

AutoGen 可以通过多个代理程序之间的对话，以最小的努力构建下一代 LLM 应用。它简化了复杂的 LLM 工作流程的编排、自动化和优化。它最大限度地提高了 LLM 模型的性能，并克服了它们的不足之处。 它支持各种复杂工作流程的多种对话模式。借助可定制和可对话的代理程序，开发人员可以使用 AutoGen 构建各种不同的对话模式，包括对话自治性、代理程序数量和代理程序对话拓扑结构等。 它提供了一系列具有不同复杂度的工作系统。这些系统涵盖了各种不同领域和复杂度的应用。它们展示了 AutoGen 如何轻松支持不同的对话模式。 AutoGen 提供了 openai.Completion 或 openai.ChatCompletion 的即插即用替代方案，作为增强型推理 API。它可以轻松进行性能调优，提供了 API 统一化和缓存等实用工具，支持更高级的用法模式，例如错误处理、多配置推理、上下文编程等。

总结

Agent = LLM + 计划+执行+纠错

Agent

  • 分解任务并完成 (ToT,Reasoning 推理 + Action 行动 React)
  • 历史的动作进行自我反思并完善 (通过长期记忆 ，Reflexion 反思)

相比RAG，但可以调用更加多的工具或者权限（例如实时信息）去完成更多实际任务而不仅仅是调用向量数据库输出结果。

image-20240327213817219

将Agent视为在RAG顶部包裹一层的东西，Agent动态地丰富查询信息，基本上允许这种整体上的更高级抽象以正确的方式使用工具，试图为您提供响应

image-20240327214416854

ToT

image-20240330170435602

  1. Thought: 只有一个中间计划
  2. 生成思路：样例
  3. 评估思路：投票
  4. 搜索算法：宽度优先搜索（深度=2，广度=5）

multi-agent

why agent

  • 任务复杂度增加后，LLM 的推理深度会崩溃
  • 能力冲突:一个 Agent 无法同时擅长规划、执行、评估
  • 无法形成自我纠错闭环

三大关键机制（真正的面试核心）

  • 角色分工（Role Assignment）——把复杂任务拆成技能颗粒度
    • 业务复杂度变高，必须拆角色
    • 企业需要可控性和解释性，而 Multi-Agent 天然具备
    • 工程团队更容易调优与扩展
  • 语言协作（Communication）——用自然语言对话完成协作
    • 符合大模型的天然工作方式（文本推理）
    • 可解释性极高
  • 协同闭环（Cooperation Loop）——多 Agent 形成“投票、质检、反思”机制

角色 → 对话 → 反思 → 监督 → 行动

why agnet需要调度

  • LLM 并不知道自己是“团队的一员”，它只会生成回复
  • 工程系统需要“可控性”和“可追踪性”
  • 必须有调度器提供：
    • 明确的消息路由
    • Agent 执行顺序
    • 错误恢复点
    • 可审计日志
    • 任务生命周期管理
  • 多 Agent 协作需要“角色自治”与“团队秩序”同时存在

中心化调度（Centralized Orchestrator）

企业级 Multi-Agent、流程自动化、全链路工具调用

  • 一个中央控制器（Orchestrator）统一调度

  • 决定谁先说、谁后说、决定任务流转顺序

  • 决定什么时候结束对话

优点

  • 稳定性最高、易 debug、审计清晰、性能可控、适合强约束业务（金融、政务、企业流程）

缺点

  • 灵活度不如分布式、扩展性需要设计

去中心但有“角色链”的协作图（Graph-Structured Multi-Agent）：如果任务流程天然是链式或 DAG 结构，用 Graph 模式最自然。

复杂推理、多步骤知识加工、需要“链式加工”的任务（如研报分析 → 观点提取 → 风险项总结）

  • Multi-Agent Graph 系统
  • Agent A → Agent B → Agent C
  • 或者分叉成多条链
  • 再由某个 Checker 汇总

优点

  • 灵活、适合复杂的流程编排、每个 Agent 知道自己上游和下游是谁

缺点

  • 图越复杂越难 debug、消息路由容易出现隐性循环、很依赖正确的图结构设计

解决sigle agent遇到的问题

sigle multi 备注
如何完成需要不同背景的复杂任务 多agent根据标准或自定义流程配合 流程可能复杂且多样，增加编程难度
如何提高应用的可靠性 多个agent讨论、复盘逻辑 依然建立在LLM能服从指令的前提下
如何灵活兼容多模态数据 招募不同擅长领域的agent合作 如何高效保存、分享多模态数据
如何提高解决问题的效率 优化、并行多个子任务agent执行 如何并行，可以自动化优化吗
  • 如何编排复杂流程(灵活、交流机制)：交流顺序、方式复杂多变，逐一枚举费时费力
  • 如何提高鲁棒性和可靠性：大模型幻觉和不稳定的指令跟随能力会影响应用运行效果
  • 如何处理多模态数据：需要在文本支持的基础上兼容多模态数据的传递，存储和展示
  • 如何提高运行效率：:需要分布式背景，并深度分析应用流程，对开发者而言优化难度高

langchain 参考

crewai

先思考两个问题

  • 每个agent的功能
  • agent之间怎样连接

An agent supervisor 路由到 individual agents.

img

多agent，互相协助

img

分层代理团队 , 上面两个的结合，多个graph 连接

img

目前的agent架构

ReAct 的核心循环

1
2
3
4
5
6
7
8
9
10
11
12
13
Reason
  ↓
Action       ← 我要做什么
  ↓
Observe      ← 做完以后发生了什么
  ↓
Reason       ← 根据结果决定下一步
  ↓
Action
  ↓
Observe
  ↓
...

例子

1
2
3
4
5
6
7
8
9
Action                      Observation

read_file("a.py")      →    文件内容
shell("pytest")        →    stdout + stderr + exit_code
web_search("xxx")      →    搜索结果
git_diff()             →    diff
write_file(...)        →    写入成功/失败
database_query(...)    →    查询结果
delegate(subagent)     →    Subagent 返回结果

现代 Harness 实际上是,ReAct 是 Engine，Harness 是 Engine 外面的控制系统。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
                    Harness
┌──────────────────────────────────────────┐
│                                          │
│  Context Manager                         │
│       ↓                                  │
│     LLM ←──────────────────────┐         │
│       ↓                        │         │
│   tool_call                    │         │
│       ↓                        │         │
│  PreTool Hook                  │         │
│       ↓                        │         │
│  Permission / Policy           │         │
│       ↓                        │         │
│     Tool                       │         │
│       ↓                        │         │
│  PostTool Hook                 │         │
│       ↓                        │         │
│ Observation ───────────────────┘         │
│                                          │
│ State / Trace / Budget / Checkpoint      │
│ Compact / Skill / Validator              │
│                                          │
└──────────────────────────────────────────┘

普遍 Agent Framework 可以拆成 7 个模块

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
Agent Runtime
│
├── ① Model Adapter
│      GPT / Claude / Gemini
│
├── ② Context Manager
│      messages
│      compact
│      memory
│      context injection
│
├── ③ Tool Registry
│      read_file
│      shell
│      search
│      browser
│      MCP
│
├── ④ ReAct Loop
│      LLM → Action → Observation
│
├── ⑤ Hook / Middleware
│      before_model
│      after_model
│      before_tool
│      after_tool
│
├── ⑥ State / Persistence
│      task
│      checkpoint
│      artifacts
│      trace
│
└── ⑦ Policy / Verification
       permission
       budget
       test
       validator
       max_steps

skill tool mcp

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
Skill
  ↓
改变 ReAct 的行为策略

Tool
  ↓
提供 ReAct 的行动能力

Hook
  ↓
限制 ReAct 的行为边界

State
  ↓
保存 ReAct 的真实进度

Verifier
  ↓
判断 ReAct 是否真的完成

Harness
  ↓
管理以上所有东西

multi agent

1
2
3
4
5
6
7
8
9
10
11
12
13
于是：

Main ReAct
│
├── Tool
├── Tool
│
├── Subagent
│      │
│      └── ReAct Loop
│
└── Tool

纯 ReAct, 非常灵活，但容易乱跑 , 通常两者和 Plan-and-Execute混合，外层 Plan-Execute，内层 ReAct，最外面再由 Harness 管住。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
                 Task
                  ↓
              High-level Plan
                  ↓
       ┌──────── Step 1 ────────┐
       │                        │
       │       ReAct Loop       │
       │ LLM → Tool → Observe   │
       │        ↑        ↓      │
       │        └────────┘      │
       └──────────┬─────────────┘
                  ↓
              Verify Step
                  ↓
               Step 2
                  ↓
              ReAct Loop

Plan-and-Execute Agents

参考

plan-and-execute agents 对比 Reasoning and Action (ReAct)-style agents会更好，目前还没完美实现。

  • 通过强制规划器明确“思考”完成整个任务所需的所有步骤，它们可以在整体上表现更好（任务完成率和质量方面）。生成完整的推理步骤是一种改进结果的可靠提示技术。将问题细分还可以实现更有针对性的任务执行。
  • 可以分散并行地做，效率更高
  • token费用更少，每一步也不一定依赖调用大模型，只有(re-)planning steps and to generate the final response才需要。

总结

  • ReAct处理复杂问题时候： 流程长，有Long term Memory 需要额外存储，看LLM性能（token多），时间长，效果不一定好。
  • Plan-and-Execute 对于planner 要求高，任务执行管理麻烦

一般分成3块

  • 计划制定：LLM生成文本，直接回应用户或传递给函数。
  • 执行：你的代码调用其他软件执行操作，比如查询数据库或调用API。
  • 分析：根据工具调用的响应做出反应，要么调用另一个函数，要么回应用户。

image-20240331153650462

Plan-and-Execute 实现

核心思想是首先制定一个多步计划，然后逐个执行计划中的任务。在完成特定任务之后，您可以重新审视计划并根据需要进行修改。

  • 只能串行，不能产生并行计划，需要计划编排生成技巧，加快执行速度

paper

Reasoning WithOut Observations 实现

可以避免在每个任务中都需要使用一个LLM（语言模型）的问题，同时允许任务依赖于前一个任务的结果。这可以通过在规划器的输出中允许变量赋值来实现。下面是代理系统设计的示意图。ReWOO

image-20240331160444661

根据问题，规划器在工具响应之前组合了一份全面的互联计划蓝图。该蓝图指导worker使用外部工具并收集证据。最后，计划和证据被配对并传递给求解器以获取答案。

  • 关键在于任务，列出需要的Evidence，有标准输出
  • 还是得串行

LLMCompiler

paper

image-20240331161915159

  • executor可以让 tasks 并行
  • planner 流式输出
  • solver 有replan功能

img

Reflection Agents

参考

Basic Reflection

  • generator ：尝试直接响应用户的请求。
  • reflector：提示扮演教师的角色，并对最初的反应提出建设性的批评。

循环进行固定次数，并返回最终生成的输出。

由于反射步骤不基于任何外部过程，因此最终结果可能不会比原始结果好得多。让我们探索一些可以改善这种情况的其他技术。

image-20240331165845052

Reflexion

在反思中，行动者智能体明确批评每个响应，并将其批评建立在外部数据的基础上。它被迫生成引用并明确列举生成的响应中多余和缺失的方面。这使得反思的内容更具建设性，并更好地引导生成器响应反馈。

它只追求一个固定的轨迹，所以如果它犯了一个错误，这个错误可能会影响后续的决策。

img

paper 实现

它结合了反射/评估和搜索（特别是蒙特卡罗树搜索），与 ReACT、Reflexion 甚至 Tree of Thoughts 等类似技术相比，可以实现更好的整体任务性能。它采用标准的强化学习 (RL) 任务框架，通过调用 LLM 来替换 RL 代理、价值函数和优化器。这旨在帮助代理适应复杂任务并解决问题，避免陷入重复循环。

img

  1. 选择：根据下面步骤 (2) 中的总奖励选择最佳的下一步行动。要么做出响应（如果找到解决方案或达到最大搜索深度），要么继续搜索。
  2. 扩展和模拟：生成 N（在我们的例子中为 5）个潜在操作以并行执行并执行它们。
  3. 反思+评估：观察这些行动的结果，并根据反思（以及可能的外部反馈）对决策进行评分
  4. 反向传播：根据结果更新根轨迹的分数。

img

LlamaIndex

router-query-engine example

1
2
3
4
5
6
7
8
9
10
11
12
from llama_index.core.query_engine.router_query_engine import RouterQueryEngine
from llama_index.core.selectors import LLMSingleSelector

# 路由
query_engine = RouterQueryEngine(
    selector=LLMSingleSelector.from_defaults(),
    query_engine_tools=[
        summary_tool,
        vector_tool,
    ],
    verbose=True
)

tool calling

Define a Simple Tool

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
from llama_index.core.tools import FunctionTool

def add(x: int, y: int) -> int:
    """Adds two integers together."""
    return x + y

def mystery(x: int, y: int) -> int: 
    """Mystery function that operates on top of two numbers."""
    return (x + y) * (x + y)


add_tool = FunctionTool.from_defaults(fn=add)
mystery_tool = FunctionTool.from_defaults(fn=mystery)


# 调用function tool
from llama_index.llms.openai import OpenAI

llm = OpenAI(model="gpt-3.5-turbo")
response = llm.predict_and_call(
    [add_tool, mystery_tool], 
    "Tell me the output of the mystery function on 2 and 9", 
    verbose=True
)
print(str(response))

Define the Auto-Retrieval Tool

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
from typing import List
from llama_index.core.vector_stores import FilterCondition


def vector_query(
    query: str, 
    page_numbers: List[str]
) -> str:
    """Perform a vector search over an index.
    
    query (str): the string query to be embedded.
    page_numbers (List[str]): Filter by set of pages. Leave BLANK if we want to perform a vector search
        over all pages. Otherwise, filter by the set of specified pages.
    
    """

    metadata_dicts = [
        {"key": "page_label", "value": p} for p in page_numbers
    ]
    
    query_engine = vector_index.as_query_engine(
        similarity_top_k=2,
        filters=MetadataFilters.from_dicts(
            metadata_dicts,
            condition=FilterCondition.OR
        )
    )
    response = query_engine.query(query)
    return response
    

vector_query_tool = FunctionTool.from_defaults(
    name="vector_tool",
    fn=vector_query
)

Setup Function Calling Agent

image-20240510215117242

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from llama_index.core.agent import FunctionCallingAgentWorker
from llama_index.core.agent import AgentRunner

agent_worker = FunctionCallingAgentWorker.from_tools(
    [vector_tool, summary_tool], 
    llm=llm, 
    verbose=True
)
agent = AgentRunner(agent_worker)
response = agent.query(
    "Tell me about the agent roles in MetaGPT, "
    "and then how they communicate with each other."
)


response = agent.chat(
    "Tell me about the evaluation datasets used."
)

task debug

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
agent_worker = FunctionCallingAgentWorker.from_tools(
    [vector_tool, summary_tool], 
    llm=llm, 
    verbose=True
)
agent = AgentRunner(agent_worker)

task = agent.create_task(
    "Tell me about the agent roles in MetaGPT, "
    "and then how they communicate with each other."
)

step_output = agent.run_step(task.task_id)

completed_steps = agent.get_completed_steps(task.task_id)
print(f"Num completed for task {task.task_id}: {len(completed_steps)}")
print(completed_steps[0].output.sources[0].raw_output)


upcoming_steps = agent.get_upcoming_steps(task.task_id)
print(f"Num upcoming steps for task {task.task_id}: {len(upcoming_steps)}")
upcoming_steps[0]


step_output = agent.run_step(
    task.task_id, input="What about how agents share information?"
)
step_output = agent.run_step(task.task_id)
print(step_output.is_last)
response = agent.finalize_response(task.task_id)

graph agent example

https://langchain-ai.github.io/langgraph/tutorials/customer-support/customer-support/

Part 1 Diagram

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

builder = StateGraph(State)


# Define nodes: these do the work
builder.add_node("assistant", Assistant(part_1_assistant_runnable))
builder.add_node("action", create_tool_node_with_fallback(part_1_tools))
# Define edges: these determine how the control flow moves
builder.set_entry_point("assistant")
builder.add_conditional_edges(
    "assistant",
    tools_condition,
    # "action" calls one of our tools. END causes the graph to terminate (and respond to the user)
    {"action": "action", END: END},
)
builder.add_edge("action", "assistant")

# checkpoint 保存持久话
# The checkpointer lets the graph persist its state
# this is a complete memory for the entire graph.
memory = SqliteSaver.from_conn_string(":memory:")
part_1_graph = builder.compile(checkpointer=memory)

打印图

1
2
3
4
5
6
7
from IPython.display import Image, display

try:
    display(Image(part_1_graph.get_graph(xray=True).draw_mermaid_png()))
except:
    # This requires some extra dependencies and is optional
    pass

session 管理 同一个人 不同 session

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import shutil
import uuid

# Let's create an example conversation a user might have with the assistant
tutorial_questions = [
    "Hi there, what time is my flight?",
    .....
    "OK great pick one and book it for my second day there.",
]

# Update with the backup file so we can restart from the original place in each section
shutil.copy(backup_file, db)
thread_id = str(uuid.uuid4())

config = {
    "configurable": {
        # The passenger_id is used in our flight tools to
        # fetch the user's flight information
        "passenger_id": "3442 587242",
        # Checkpoints are accessed by thread_id   session 管理 同一个人 不同 session
        "thread_id": thread_id,
    }
}


_printed = set()
for question in tutorial_questions:
    events = part_1_graph.stream(
        {"messages": ("user", question)}, config, stream_mode="values"
    )
    for event in events:
        _print_event(event, _printed)

Part 2 diagram

需要用户去确认

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    # 保留用户信息
    user_info: str
    
from langgraph.checkpoint.sqlite import SqliteSaver
from langgraph.graph import END, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

builder = StateGraph(State)


def user_info(state: State):
    return {"user_info": fetch_user_flight_information.invoke({})}


# NEW: The fetch_user_info node runs first, meaning our assistant can see the user's flight information without
# having to take an action
builder.add_node("fetch_user_info", user_info)
builder.set_entry_point("fetch_user_info")
builder.add_node("assistant", Assistant(part_2_assistant_runnable))
builder.add_node("action", create_tool_node_with_fallback(part_2_tools))
builder.add_edge("fetch_user_info", "assistant")
builder.add_conditional_edges(
    "assistant", tools_condition, {"action": "action", END: END}
)
builder.add_edge("action", "assistant")

memory = SqliteSaver.from_conn_string(":memory:")
part_2_graph = builder.compile(
    checkpointer=memory,
    # 在
    # NEW: The graph will always halt before executing the "action" node.
    # The user can approve or reject (or even alter the request) before
    # the assistant continues
    interrupt_before=["action"],
)

我们看到了“广泛”的聊天机器人如何依靠单个提示和 LLM 来处理各种用户意图，让我们走得更远。然而，使用这种方法很难为已知意图创建可预测的出色用户体验。

您的图表可以检测用户意图并选择适当的工作流程或“技能”来满足用户的需求。每个工作流程都可以专注于其领域，允许单独的改进，而不会降低整体助手的性能。

在本节中，我们将把用户体验分成单独的子图，形成如下结构：

第 4 部分图

意图的上下文切换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
from typing import Annotated, Literal, Optional

from typing_extensions import TypedDict

from langgraph.graph.message import AnyMessage, add_messages


def update_dialog_stack(left: list[str], right: Optional[str]) -> list[str]:
    """Push or pop the state."""
    # 进退stack  出入状态
    if right is None:
        return left
    if right == "pop":
        return left[:-1]
    return left + [right]


class State(TypedDict):
    messages: Annotated[list[AnyMessage], add_messages]
    user_info: str
    dialog_state: Annotated[  # 处于什么状态
        list[
            Literal[
                "assistant",
                "update_flight",
                "book_car_rental",
                "book_hotel",
                "book_excursion",
            ]
        ],
        update_dialog_stack,
    ]

每个步骤都要加入判断任务状态，结束 cancel 与原因信息

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
class CompleteOrEscalate(BaseModel):
    """A tool to mark the current task as completed and/or to escalate control of the dialog to the main assistant,
    who can re-route the dialog based on the user's needs."""

    cancel: bool = True
    reason: str

    class Config:
        schema_extra = {
            "example": {
                "cancel": True,
                "reason": "User changed their mind about the current task.",
            },
            "example 2": {
                "cancel": True,
                "reason": "I have fully completed the task.",
            },
            "example 3": {
                "cancel": False,
                "reason": "I need to search the user's emails or calendar for more information.",
            },
        }

不同的子图数据格式也不一样

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
class ToBookCarRental(BaseModel):
    """Transfers work to a specialized assistant to handle car rental bookings."""

    location: str = Field(
        description="The location where the user wants to rent a car."
    )
    start_date: str = Field(description="The start date of the car rental.")
    end_date: str = Field(description="The end date of the car rental.")
    request: str = Field(
        description="Any additional information or requests from the user regarding the car rental."
    )

    class Config:
        schema_extra = {
            "example": {
                "location": "Basel",
                "start_date": "2023-07-01",
                "end_date": "2023-07-05",
                "request": "I need a compact car with automatic transmission.",
            }
        }


class ToHotelBookingAssistant(BaseModel):
    """Transfer work to a specialized assistant to handle hotel bookings."""

    location: str = Field(
        description="The location where the user wants to book a hotel."
    )
    checkin_date: str = Field(description="The check-in date for the hotel.")
    checkout_date: str = Field(description="The check-out date for the hotel.")
    request: str = Field(
        description="Any additional information or requests from the user regarding the hotel booking."
    )

    class Config:
        schema_extra = {
            "example": {
                "location": "Zurich",
                "checkin_date": "2023-08-15",
                "checkout_date": "2023-08-20",
                "request": "I prefer a hotel near the city center with a room that has a view.",
            }
        }

彻底结束dialog的标记, 或者上下文切换

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# Flight booking assistant
builder.add_node(
    "enter_update_flight",
    create_entry_node("Flight Updates & Booking Assistant", "update_flight"),
)
builder.add_node("update_flight", Assistant(update_flight_runnable))
builder.add_edge("enter_update_flight", "update_flight")
builder.add_node(
    "update_flight_sensitive_tools",
    create_tool_node_with_fallback(update_flight_sensitive_tools),
)
builder.add_node(
    "update_flight_safe_tools",
    create_tool_node_with_fallback(update_flight_safe_tools),
)


def route_update_flight(
    state: State,
) -> Literal[
    "update_flight_sensitive_tools",
    "update_flight_safe_tools",
    "leave_skill",
    "__end__",
]:
    route = tools_condition(state)
    if route == END:
        return END
    tool_calls = state["messages"][-1].tool_calls
    did_cancel = any(tc["name"] == CompleteOrEscalate.__name__ for tc in tool_calls)
    if did_cancel:
        return "leave_skill"
    safe_toolnames = [t.name for t in update_flight_safe_tools]
    if all(tc["name"] in safe_toolnames for tc in tool_calls):
        return "update_flight_safe_tools"
    return "update_flight_sensitive_tools"


builder.add_edge("update_flight_sensitive_tools", "update_flight")
builder.add_edge("update_flight_safe_tools", "update_flight")
builder.add_conditional_edges("update_flight", route_update_flight)


# 彻底退出的标记
# This node will be shared for exiting all specialized assistants
def pop_dialog_state(state: State) -> dict:
    """Pop the dialog stack and return to the main assistant.

    This lets the full graph explicitly track the dialog flow and delegate control
    to specific sub-graphs.
    """
    messages = []
    if state["messages"][-1].tool_calls:
        # Note: Doesn't currently handle the edge case where the llm performs parallel tool calls
        messages.append(
            ToolMessage(
                content="Resuming dialog with the host assistant. Please reflect on the past conversation and assist the user as needed.",
                tool_call_id=state["messages"][-1].tool_calls[0]["id"],
            )
        )
    return {
        "dialog_state": "pop",
        "messages": messages,
    }


builder.add_node("leave_skill", pop_dialog_state)
builder.add_edge("leave_skill", "primary_assistant")