Agent入门

学习目标

学完本章你应该能够：

1. 说清 Agent 是什么：Agent = 规划（Planning） + 记忆（Memory） + 工具（Tools） + 行动（Action） 的自主代理，能自己去思考、去行动。
2. 区分短期记忆与长期记忆的存储与使用方式，讲清规划的「任务拆解 + 反思」两大组成部分，以及常见工具类型（搜索、RAG、代码解释器、第三方服务 API）。
3. 讲清来自吴恩达（Andrew Ng）分享的 Agent 四种设计模式：反馈、工具调用、规划、多智能体协作，以及各自的优点与缺陷。
4. 用 Python + LangChain 从零实现一个 RAG Agent：通过「向量检索 → 大模型反思是否能回答 → 不能则加块重检索」的循环，克服传统 RAG 「检索文档不相关」与「大模型幻觉」两大弱点。
5. 分析吴恩达开源 TranslationAgent 的三段式架构（初始翻译 → 反思 → 优化）与提示语设计，理解「反思工作流」，并对比其与直接用 ChatGPT 翻译的稳定性差异。
6. 掌握 LangGraph 三大核心概念 State / Node / Edge，会用它搭单节点 Chat Agent、串行多节点复刻 TranslationAgent，理解有向无环图（DAG）与有向有环图的区别。
7. 接入 LangSmith 做 Agent 开发调试日志与调用追踪，学会在平台上观察每一步的输入输出。
8. 用 LangGraph + conditional_edge 实现一个 ReAct 自主思考 Agent（获取→修改工作负载的工具循环），并理解 super step 并行与单节点执行次数的关系。
9. 掌握打造个人运维知识库/数字分身的三种方式（无代码自托管、Coze 等 SaaS、二次开发），并用 LangGraph 实现 Self-RAG 自适应 RAG + Web 搜索路由整合成完整的运维知识库 Agent。

前置知识：

• 能熟练使用 Jupyter Notebook、会写基础 Python（函数、循环、字典、字符串格式化）。
• 了解传统 RAG 的基本流程（文档切分 → embedding → 向量检索 → 拼 prompt 生成），建议先回看上一节「RAG 入门」。
• 有至少一个 OpenAI 兼容格式的 LLM API Key。
• 了解 Kubernetes / 工作负载（Deployment）的基本概念更佳，但不是必须。

本章你会动手做的事：

1. 从零实现一个 RAG Agent，用「循环加块 + 大模型反思」解决传统 RAG 答不出「谁管理的系统最多」这类问题。
2. 克隆并运行吴恩达开源 TranslationAgent，把一段英文《静夜思》翻译成中文，再对比直接调 ChatGPT 的翻译稳定性。
3. 用 LangGraph 分别搭一个「单节点 Chat Agent」和一个「三节点串行 翻译→反思→优化」的 Agent。
4. 用 LangGraph + conditional_edge 实现一个 ReAct 工具循环 Agent（先 get 工作负载、再 patch 修改），并接入 LangSmith 观察每一次调用。
5. 用 LangGraph 实现 Self-RAG 自适应检索（评估相关性 → 评估是否基于事实 → 评估是否解决问题 → 兜底重写问题），再叠加 Web 搜索路由，打造个人运维知识库 Agent。

什么是 Agent

先看一下 Agent 的能力可以被概括成一张图：它包含了四个组成部分——记忆（Memory）、工具（Tools）、规划（Planning），以及行动（Action）。把大模型类比成人类的大脑：如果只有大脑，没有手脚和记忆，大脑是无法独立行动的。Agent 就是大模型的「手脚 + 记忆」——有了它，大模型就相当于拥有了完整的身体，可以去跟外界交互。

用一句话概括：Agent 就是具有复杂推理能力、记忆，以及执行任务手段的自主代理。注意「自主」两个字——它能自己思考，然后自己去行动。

Agent = 规划(Planning)  +  记忆(Memory)  +  工具(Tools)  +  行动(Action)

下面把四个模块分别展开。

规划（Planning）

规划模块主要分两部分：任务拆解 与 反思。

• 任务拆解：为了能把一个复杂任务拆分成多个子任务，通过完成这些简单的子任务，从而实现复杂任务。比如「排查线上 P99 飙高」会被拆成「查指标 -> 查容器 -» 查依赖 -> 定位根因 -> 决定扩容还是重启」。
• 反思（Reflection）：赋予大模型一种自主决策、自我纠正错误的能力，典型实现就是 ReAct、Reflexion，以及思维链（Chain of Thought）、思维树（Tree of Thoughts）。目的都是提升大模型自主行动和决策的能力。

记忆（Memory）

记忆包含短期记忆和长期记忆：

类比：你的"短期记忆"是手头正在聊的事、聊完一局就忘；“长期记忆"是记在笔记本里、随时能翻出来用的东西。对应到工程里就是：短期记忆 = 大模型有限的上下文窗口（每次推理时都在里面），长期记忆 = 外部数据库（向量库 / 关系库），对话时模型从两者里取上下文继续推理。

• 短期记忆：主要通过当前的上下文对话学习。由于大模型的上下文（context window）是有限的，短期记忆自然也是有限的。
• 长期记忆：借助外部数据库来存储对话。可以从向量数据库获取知识库内容，也可以把对话历史放进关系型数据库。对话进行时，大模型从短期记忆和长期记忆里取上下文，基于它们继续推理。

工具（Tools）

工具模块用来增强大模型的能力：

• 搜索工具（如 Google Search）：受知识日期限定，大模型感知不到现实世界发生的事，搜索让它获取最新知识。
• RAG 私有知识库：和搜索类似，但检索的是你的私有知识库。
• 代码解释器：运行大模型生成的代码，在一些数据统计、数学运算场景能极大提升准确性——与其让大模型直接推理，不如让它生成一段 Python 代码、调用代码解释器运行，得到正确结果。
• 第三方服务 API：能把内部几乎所有的运维能力封装成工具提供给大模型，比如查询日志、查询指标、创建工单，甚至部署工作负载，都能抽象成 API 让大模型调用整合。

用 RAG 直观理解 Agent

用一个 RAG 的例子理解「自主」的含义。传统 RAG 流程是：从向量数据库检索相关内容，然后直接基于内容生成回答。它有两个不确定性：

1. 检索出的文档不一定与问题相关——你得到的可能是一段无用的文档，回答不了你的问题。
2. 大模型回答时有概率产生幻觉——它答的未必解决了你最初的提问。

所以传统 RAG 效果并不好。RAG Agent 则结合了 Agent 的能力，解决这两个问题：

• 从向量数据库检索到文档后，先由大模型自行判断这个文档跟我的提问是否相关、是否匹配——这是 Agent「自我决策」的体现。
• 如果匹配，就把相关文档交给大模型问答；如果不匹配，Agent 会自动把问题重写（基于近似语义、语义一致性地重写），再基于重写后的问题重新检索，继续走这个循环。
• 这个循环一直进行，直到大模型认为当前文档与提问匹配为止。

flowchart TD
    Q[用户问题] --> R[向量检索文档]
    R --> E{大模型评估
文档与问题相关?}
    E -->|匹配| A[提供文档问答]
    E -->|不匹配| RW[重写问题]
    RW --> R

所以 RAG Agent 用「循环 + 自主决策」的方式，更有机会得到准确的答案——这就化解了传统 RAG 的两个弱势：检索可能不相关、回答可能没解决最初提问。

Agent 的四种设计模式

这一节内容来自吴恩达教授在一次公开演讲的分享。四种设计模式分别是：反馈（Reflection）、工具调用（Tool Use）、规划（Planning）、多智能体协作（Multi-Agent）。

模式一：反馈（Reflection）——自我反思与自我迭代

反馈的核心是自我反思和自我迭代。前面讲的 RAG Agent 例子，就是自我反思和自我迭代的体现：它能通过这两个过程不断优化输出，进而提高任务准确率。

这有点像人完成工作任务时的复盘：先对自己的工作做回顾、找到需要改进的地方；自我迭代则是基于反思结果去做迭代，让每次生成的结果都比上一次好。

最典型的例子是编写代码：让大模型不断基于代码的错误信息反复优化代码。它可能发现第 3 行有 bug，修复后运行又发现其他 bug，直到生成的代码没有错误、可以运行为止。

• 优点：提高准确性和回复质量。
• 缺点：
  • 多次迭代后整体耗时较长，token 成本更高；
  • 一般需要结合工具（如写代码就要用代码解释器运行，才能发现 bug）；
  • 反馈机制和流程设计较为复杂，写程序时也更复杂。

模式二：工具调用（Tool Use）——扩充模型能力范围

• 优点：数学计算场景可通过生成代码极大提高准确性；主要是扩充了模型的能力范围——原本大模型只能以文本方式交流（Chat），有了工具加持之后，可以去访问任何外部系统，基于外部系统数据交流。
• 缺点：
  • 工具调用会让模型产生新的外部依赖，一旦工具故障，模型可能无法回复，进入错误状态；
  • 复杂任务不能直接靠一个工具完成，可能需要结合 ReAct 或子目标拆解、让多个子工具协作；
  • 多工具协作会极大提升系统整体复杂性，甚至需要中心化的工具协调器来协调不同工具、聚合完成任务——这对整体实现架构带来更大挑战。

模式三：规划（Planning）——拆分步骤

把复杂任务拆分成更小、更简单的步骤。

• 优点：简化复杂任务，并提升复杂任务的准确性。
• 缺点：
  • 拆分出的步骤之间依赖性强：下一步极大依赖上一步的准确性，属于串行调用；
  • 一旦上游某个步骤调用出错，下游步骤也会跟着出错；
  • 拆分出更多小步骤后，整体任务耗时与资源消耗也更多。

模式四：多智能体协作（Multi-Agent）——分工协作

让多个智能体进行分工协作，提高执行效率与准确性。规划模式拆分出的子任务，既可以串行运行，也可以用多智能体协作模式让可并行的任务并行运行，从而提高效率。

典型例子是近期很火的 ChatDev 项目：用户只需提出一个目标（比如「开发一个五子棋游戏」），它就能自主完成所有任务——包括写需求文档（文档阶段）、设计阶段、编码阶段、测试阶段。每一个阶段都有一个单独的智能体 Agent 负责，而最核心的位置是一个指挥官（类似公司的 CEO），主要任务是给这些多智能体做任务协调、分配任务、推动任务。

• 缺点：整体设计起来的复杂性更高。

Agent 开发实战（一）：从零实现 RAG Agent

理论知识介绍得差不多，现在从零开发一个 Agent。要实现的 Agent 很简单：基于之前一个非常简单的 RAG 流程，进一步借助 Agent 增强 RAG 的能力。

流程是：输入问题后，通过向量检索搜索出相近似的文本块；找到后不直接回答，而是先让大模型判断这个文本块能否回答最初的问题；如果不能，就重新进入向量检索流程匹配更多的块。这是一个循环流程，直到找到能回答问题的文本块，才真正回答。

• 循环退出条件：设置最大循环次数为 15 次。
• 细节：每次循环做向量匹配时，会多匹配一个块——第一次匹配 1 个块，第二次 2 个块，第三次 3 个块，以此类推。
• 核心：用 Python while 循环，匹配到知识库后让大模型进行「反思」——参考后面要讲的 TranslationAgent，先反思，再检查是否能回答问题；可以就回答，不行就继续循环。

搭建环境与复用传统 RAG 代码

创建一个新的 Jupyter Notebook（例如 agent_rag.ipynb）。首先安装相关依赖、选择内核，然后把上一节 RAG 用到的代码直接复制过来——包括怎么 embedding 运维知识库。再创建一个 data/ 目录，把 Markdown 格式的运维知识库放进去；使用 MarkdownHeaderTextSplitter（源码里叫 md_splitter）对知识库做切片；最后把传统 RAG 流程的检索代码也复制过来：用 similarity_search_with_score 做向量检索，配一条系统提示语「基于上下文回答问题，如果不知道就说不知道，不要编造」。

先跑一遍传统 RAG，问「谁管理的系统最多」。运维知识库里对业务负责人有描述，人类看很明显——小王维护的微服务最多（他同时负责 gateway 和 backend 两个服务）。但传统 RAG 跑出来的结果是：向量检索到的最近似的块并不是业务负责人对应的块，而是前面的某一块，所以大模型回答「不知道」。

这就演示了传统 RAG 的缺点：你以为匹配到了最近似的块，但它仍然不能回答问题。

实现 RAG Agent 的反思循环

现在实现增强的 RAG 循环。核心代码是一个「检索 + 反思 + 决定退出或继续」的循环。

from langchain_openai import ChatOpenAI
from langchain_community.vectorstores import FAISS

# 1. 定义相似性检索函数：每次取 k 块，循环时 k + 1
def retrieve(query: str, k: int = 1):
    # vectorstore.similarity_search_with_score(query, k=k)
    docs = vectorstore.similarity_search_with_score(query, k=k)
    return "\n---\n".join(doc.page_content for doc, _ in docs)

# 2. 定义"反思"系统提示语：让大模型判断上下文能否回答问题
REFLECTION_PROMPT = """根据上下文，识别问题是否能被回答。
如果不能回答，返回 JSON 字符串：{"can_answer": false}
如果可以回答，返回 JSON 字符串：{"can_answer": true}

上下文：
{context}

问题：
{question}
"""

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 3. 主循环：不断检索更多块，直到大模型判断能回答为止
query = "谁管理的系统最多"
k = 1
MAX_ITER = 15
can_answer = False

while not can_answer and k <= MAX_ITER:
    print(f"第 {k} 次检索")
    context = retrieve(query, k=k)
    prompt = REFLECTION_PROMPT.format(context=context, question=query)
    messages = [
        {"role": "system", "content": "开始检查上下文是否能够回答问题。"},
        {"role": "user", "content": prompt},
    ]
    resp = llm.invoke(messages)
    print(resp.content)
    if '"can_answer": true' in resp.content:
        can_answer = True
    else:
        k += 1  # 不能回答 -> 加一块，继续检索

# 4. 退出循环后：做最终推理，正式回答
FINAL_SYSTEM_PROMPT = "你是运维问答助手，请使用检索到的上下文回答用户提出的问题，如果不知道答案就说不知道，不要编造。"
final_ctx = retrieve(query, k=k)
final_messages = [
    {"role": "system", "content": FINAL_SYSTEM_PROMPT},
    {"role": "user", "content": f"上下文如下：\n{final_ctx}\n\n问题：{query}"},
]
answer = llm.invoke(final_messages)
print("最终回答：", answer.content)

运行结果是：第一次检索返回 false，第二、三……一直到第六次都是 false，直到第七次检索才检索到业务负责人相关的块。此时大模型返回 {"can_answer": true}，最终打印出「可以回答问题，开始回答」，给出正确的答案。

体会到关键差异了吧——这个 RAG Agent 弥补了传统 RAG 不能回答某些问题的缺陷，靠的是 Agent 最重要的特性：循环，即自主思考、自主决策。

RAG Agent 的三个改进方向

真实的 RAG Agent 开发不会这么简单。下面是几个可以自己实践的方向：

1. 切片策略：循环里不断加知识块时，没有判断块的大小。对于小模型，或上下文接近超限时，RAG Agent 会出问题。需要引入切片策略，对检索到的知识库先切片、再基于分片让大模型检查能否回答，避免上下文 token 超限。
2. 改进向量检索：可以通过增加 offset 参数，过滤掉已经验证过、完全没有用、回答不了问题的文本块（因为目前是不断往下加，把已确认无用的块也带上了）。
3. 更稳定的结构化输出：目前是直接通过字符串判断能否回答，可以用 JSON Mode 获得更稳定、更结构化的输出。

分析开源 TranslationAgent

接下来介绍吴恩达教授开源的一款翻译 Agent——通过这个例子看一个「好的 Agent」其架构设计和提示语是怎么写的，这个 Agent 有非常多值得参考的地方。

反思工作流：三段式结构

TranslationAgent 是基于反思工作流的翻译 Agent，模拟人类翻译专家的思考过程，分三个流程：

1. 初始翻译：先把输入的问题做初始翻译，得到初步译文。
2. 改进（反思）：基于翻译结果做一次改进——让大模型扮演另一个角色（翻译结果的评估/评判官），提出修改意见：哪里不准确、不流畅、或不符合用户所在国家/地区的语言习惯等建设性意见。
3. 优化：根据第二步的建设性意见，对结果重新翻译或再次优化，最终得到翻译的最终结果。

flowchart LR
    A[初始翻译] --> B[反思/评判
提出改进意见]
    B --> C[根据意见重新翻译]
    C --> D[最终译文]

完整流程还有多段（multi）翻译的处理：如果输入超长（大于模型最大 token），会做翻译分割（translation splitting）；每段翻译完先反思、再完善，最终合并结果。这跟我们前面提到的「切片策略」思路一致。

实际演示：翻译《静夜思》

• 安装依赖（用的是 uv 工具管理）、克隆 translation-agent 仓库，install 后使用 src 目录里的示例脚本。
• 把**英文版《静夜思》**放进 simple_text 目录，改写 example_script.py：source_language=英语，target_language=中文，country=中国，使其符合中文语言习惯。
• 运行后，它能真正把一段英文翻译成中国的古诗，翻译效果很好。

作为对照，用 ChatGPT 直接翻译同一段英文（最简接入，system prompt 只写「将以下英文翻译成中文」）——第一次能翻译，第二次却不行，输出不稳定。这正好体现 TranslationAgent 反思工作流的价值。

源码分析：三个核心函数与提示语

对应 PPT 里的三个流程，源码里有三个函数：

1. 初始翻译函数：提示语写得很简单——「你是一个翻译专家，需要把 s_language 翻译成 t_language」+「原文」，直接翻译即可。
2. 结果反思函数：输入有三个东西——原文（s）、要翻译的语言（t）、初始翻译译文（translation）。它让翻译 Agent 自己思考并提出和改进建议。这个反思 prompt 写得比较复杂，给出了很多可执行的操作：比如句子不通顺、有错误、标点符号问题、有多余的说法等，让大模型基于这些方面思考，最终输出建设性、有帮助的建议。
3. 优化函数：把反思建议再传入第三段优化流程，基于建议重新翻译。它的入参更多：原始译文、反思结果（改进意见），最终输出最后的翻译结果。

三个函数串起来，就是前面看到的「翻译 → 反思 → 优化」流程。

优势与不足

• 优势：基于反思工作流，翻译效果明显优于普通翻译工具（包括直接 ChatGPT 里翻译）；流程可定制。
• 不足：
  • 翻译效果依赖大模型智力，目前只能使用 OpenAI 的模型；做源码定制改成国产大模型甚至小模型时，效果可能打折扣。
  • 相比商业翻译引擎，在特定专业领域可能略逊一筹（由大模型智力决定）。
  • 使用有门槛：要装一大堆依赖、还要写代码，并非产品化的工具，主要是给开发者参考 Agent 的开发模式和架构。
  • 基于固定流程：是链式思考（先翻译、再反思、再优化），并未使用 Agent 最核心的自我思考循环机制。可以自己加一层判断——最终翻出来的结果还能不能再优化？能就回到「提出意见」环节，继续思考，经过 N 轮优化拿到更好结果。

Agent 开发实战（二）：LangGraph 基础

到现在为止，Agent 的开发还比较原始，除了 LangChain 之外几乎没有用框架。LangChain 最近开源了一个新工具叫 LangGraph，能极大降低开发 Agent 的编码量和成本，是本节非常重要的内容。

LangGraph 是什么

LangGraph 是构建在 LangChain 之上的库，只为 Agent 添加循环（loop）运算能力——就是前面强调的「for / ReAct 循环」。

LangChain 主要面向定义一个有向无环图（DAG）：像 TranslationAgent 里那样「第一步做什么、第二步做什么、第三步做什么」，一旦前面的节点失败，后面的都会失败。做任务尤其是调用工具时，如果中间某个函数失效，后面的函数都会失效——它没有循环、回溯或重新运行的机制。

而 LangGraph 引入了循环功能，可以实现有向有环图：节点可以这样运行、那样运行、还能再回去（循环）。从而借助 LangGraph 实现更复杂的 Agent 开发。简而言之，LangGraph 就是构建 Agent 的框架，让你更简单地开发 Agent。

flowchart LR
    A[START] --> B[节点A]
    B -->|无环| C[节点B]
    C --> D[节点D]
    B -->|有环：可回溯| B

三大核心概念：StateGraph / Node / Edge

只需要知道 StateGraph、Node、Edge 三个东西，就能开始简单的 LangGraph 开发。

类比：把 LangGraph 想象成一张地铁线路图——StateGraph 是整张图纸和你随身带着的"当前在哪一站"的状态记录，Node 是每个站（一个计算步骤），Edge 是站与站之间的轨道，轨道还能带"信标"决定下一站往哪走。对应到工程里就是：StateGraph 定图与存状态，Node 跑业务逻辑，Edge 决定执行顺序并支持按条件动态跳转。

• StateGraph（状态图）：定义有向有环图以及保存其状态的类。需要往里面添加 Node 才能形成图。可以把它理解为一个可以容纳节点的空容器。
• Node（节点）：代表一个函数或一个计算步骤。既然代表函数，能干的事情就多了——可以写任何业务逻辑、任何运算步骤：处理输入、做决策、与外部 API 交互。节点之间有相互依赖关系，既有串行执行步骤，也有并行执行流程（比如 B 和 C 并行，都执行完再执行 D），甚至能做节点之间的循环运行。
• Edge（边）：控制节点往哪里走的条件，定义计算流程，包括起始节点（Entry Point）与终止节点。Edge 支持条件——可以根据 Edge 挂的一个函数返回值，动态决定下一个要执行的节点（比如一个 condition 函数的返回结果控制流程走到 node2 / node3 / node4）。

LangGraph 示例一：单节点 Chat Agent

先实现一个最简单的 Agent——Chat Agent，即把大模型的 Chat 能力放进 LangGraph 里。主要步骤：先定义一个 StateGraph（Graph State）用来存内部状态；再定义一个 Node（函数）；把节点添加进去；再 Set 一条 Edge（设置 Entry Point 起点为 chat、终点也为 chat）。

from typing import TypedDict, Annotated   # 类型标注
from langgraph.graph import StateGraph, END
from langchain_openai import ChatOpenAI
from langgraph.graph.message import add_messages

llm = ChatOpenAI(model="gpt-4o", temperature=0)

# 1. 声明状态：messages 用 add_messages 自动追加历史
class State(TypedDict):
    messages: Annotated[list, add_messages]

# 2. 定义 chat 节点：调用大模型，返回消息
def chat_node(state: State):
    return {"messages": [llm.invoke(state["messages"])]}

# 3. 构建图
workflow = StateGraph(State)
workflow.add_node("chat", chat_node)   # 添加节点
workflow.set_entry_point("chat")       # 起点 = chat
workflow.set_finish_point("chat")      # 终点 = chat
graph = workflow.compile()             # 必须编译才能生效

# 4. 与用户交互循环
while True:
    user_input = input("你：")
    if user_input.lower() in {"q", "quit"}:
        break
    for event in graph.stream({"messages": [{"role": "user", "content": user_input}]}):
        value = event["values"]
        last = value["messages"][-1]
        print("Agent：", last.content)

display(Image(graph.get_graph().draw_mermaid_png())) 可以打印出图：start -> chat -> end。这个 Agent 功能很简单，只有一个节点做模型推理，因为 messages 是列表且用 add_messages 追加，所以连上下文记录都不用手动处理。

LangGraph 示例二：复刻 TranslationAgent（三节点串行）

用 LangGraph 实现 TranslationAgent 的三个步骤：初始翻译 → 反思 → 优化。这里需要三个节点，以串行方式执行。

• 初始化大模型：source_language=中文、target_language=英文（或反过来）。
• 声明状态 messages（Annotated[list, add_messages]）。
• 定义三个函数，每个都从 state 里取上一步结果：
  • initialize_translation：第一次翻译。系统提示语从源码复制（“你是翻译专家，把 s_language 翻译成 t_language”）+ 翻译 prompt + 用户输入原文。
  • reflection：反思。注意原文不是用户第一次对话，而是在现有 messages 里取倒数第二条（index -2），译文取倒数第一条（index -1）。系统提示语 + 反思 prompt（包含可执行优化点），让大模型提出改进意见。
  • improve_translation：改进。入参更多——原文、初始译文中、反思建议，系统提示语 + improve prompt，基于反思建议重新优化译文。
• 构建图：set_entry_point("initialize_translation")，再 add_edge 依次 初始化 → 反思 → 优化，最后 add_edge("improve", END)，编译。

from langgraph.graph import StateGraph, START, END

workflow = StateGraph(State)
workflow.add_node("initial_translation", initialize_translation)
workflow.add_node("reflection", reflection)
workflow.add_node("improve_translation", improve_translation)

workflow.add_edge(START, "initial_translation")
workflow.add_edge("initial_translation", "reflection")
workflow.add_edge("reflection", "improve_translation")
workflow.add_edge("improve_translation", END)

graph = workflow.compile()

运行后，输入《静夜思》中文版：第一步输出初步译文，第二步反思提出建议，第三步根据建议输出最终优化后的译文。这样我们就用 LangGraph 复刻了 TranslationAgent——你画图时不用自己连接这么多状态，LangGraph 会帮你做调度。

Agent 开发实战（三）：LangSmith 与 ReAct Agent

从这开始 Agent 开发变得复杂了。比如想调试 Agent 的每一个步骤——每一步给大模型的提示语是什么、输出结果是什么，如果每一步都手动打印，会比较麻烦。这时可以集成 LangSmith 这个平台来简化调试。

集成 LangSmith 做调用追踪

LangSmith 是 Agent 开发的调试日志与追踪平台，跟 LangChain / LangGraph 原生集成（它是 LangChain 这家公司开发的）。它是一个 SaaS 服务，需要在它的官网申请 API Key，超过一定用量需要付费。集成非常简单。

from langsmith import Client  # 或通过环境变量配置

# 初始化 LangSmith
client = Client(
    api_key=LANGCHAIN_SMITH_API_KEY,   # 在 langsmith 平台 Setting 里申请
    project_name="default",            # 默认项目；自定义项目名要写对
)

集成后跑一遍翻译流程，在 LangSmith 平台点开最近一条 trace，就能看到 Agent 所有的追踪数据：先运行 initial_translation 节点，它调用的是 OpenAI、用的是 GPT 模型，输入输出分别是 system prompt 和 human 提示语、大模型输出；第二步反思的 system prompt、输入、大模型输出——全部一目了然。

LangSmith 还有个很好用的能力叫 playground（在线调试）：可以实时调试提示语、查看返回结果。不过目前只对单纯的大模型调用（如 chat.completion 接口）支持在线调试，Agent 暂不支持。LangSmith 功能很多，感兴趣可以深入了解。

LangGraph ReAct Agent：工具调用循环 + 条件边

了解了 LangGraph 简单使用后，看一个非常复杂的例子：Agent 会根据用户的输入自主思考，循环调用工具，直到有足够的信息解决用户的输入为止。这是一个循环——之前开发的 Agent 都是单向的（有向无环图），这里是一个有向有环图。

具体到例子：定义两个工具 get 和 patch。get 用来获取工作负载，patch 用来修改工作负载。当用户提问「请帮我修改 payment 服务工作负载镜像为……」：

1. Agent 首先自主思考：「要修改这个工作负载之前，得先获取它」——先去调用 get 获取 payment 工作负载的 yaml 内容。
2. 获取成功后再调用 patch 修改工作负载，最终返回修改成功。

调用工具时，get 没获取成功会一直循环获取直到成功；patch 也是同理。

怎么控制循环结束？ 借助 conditional_edge（条件边） 实现动态控制——这就是前面讲 Edge 时提到的「条件」。当条件函数返回字符串 tools 时，继续调用 tools 相关节点；当返回字符串 end 时就退出。条件来源于大模型返回的最后一条消息是否是工具调用。

实现代码要点：

from langgraph.graph import StateGraph, START
from langchain_openai import ChatOpenAI

# 工具1：获取工作负载（这里用静态 yaml 模拟真实 K8s API 读取）
def get_deployment(deployment_name: str) -> str:
    """Use tool to get deployment."""
    print(f"get deployment: {deployment_name}")
    return DEPLOYMENT_YAML   # 返回一份静态 yaml

# 工具2：修改工作负载
def patch_deployment_to_json(deploy_name: str, patch: str) -> str:
    """Patch a deployment to a new image."""
    print(f"apply deployment: {patch}")
    return "deployment applied"

# 调用大模型的核心节点
def call_model(state: State):
    messages = state["messages"]
    response = model_with_tools.invoke(messages)
    return {"messages": [response]}

# 决定"是否继续"的条件函数：看大模型最后一条是否还在调用工具
def should_continue(state: State):
    last_message = state["messages"][-1]
    if last_message.tool_calls:
        return "tools"   # 继续调用工具节点
    return END           # 结束

# 把工具绑定到模型
model = ChatOpenAI(model="gpt-4o", temperature=0)
model_with_tools = model.bind_tools([get_deployment, patch_deployment_to_json])

# 构建图（tools_node 会把工具直接转成节点）
from langgraph.prebuilt import ToolNode
workflow = StateGraph(State)
workflow.add_node("chat", call_model)
workflow.add_node("tools", ToolNode([get_deployment, patch_deployment_to_json]))
workflow.add_edge(START, "chat")                       # 起点 -> 聊天
workflow.add_conditional_edge("chat", should_continue) # 条件边控制循环
workflow.add_edge("tools", "chat")                     # 工具 -> 聊天（循环回来）
graph = workflow.compile()

为什么它能按顺序先调 get_deployment、再调 patch_deployment_to_json？因为把这两个工具绑定到了模型上，由大模型的 ReAct 机制（自我思考）自行决定先调哪个、再调哪个来实现目标。

运行后：tool call 先执行 get_deployment，得到 payment 服务的 yaml；接着调用 patch_deployment_to_json，返回「已成功将 payment 工作负载更新为……」。整个运行流程符合预期。在 LangSmith 里能看到：每次调用传入的 function call 定义（就是我们 bind 的工具）、should_continue 返回 tools 就继续调工具、最终返回 end 后大模型重新推理给出最终回答。

flowchart LR
    S[START] --> C[chat
调用大模型]
    C -->|should_continue
还有工具调用| T[tools
执行工具]
    T --> C
    C -->|should_continue
返回 END| E[END]

LangGraph 并行与 super step

LangGraph 还有一个重要概念叫 super step，用来理解每个步骤运行完之后，流程是怎么实现调度的。通过一个非常简单的例子体会：

• 定义 A、B、C、D 四个节点，B 和 C 在 A 运行完之后并行。
• 往 A 下面添加 B 和 C 会出现两条分叉，之后分别到 D（A->B->D、A->C->D），最后 D -> END。
• 运行会发现：D 最终只被执行一次。

接着一个「神奇」的事情发生：当给 B 节点的下游再定义一个节点 E 时，会发现 D 被 执行了两次。为什么？

• 这是因为 super step 的概念：B 和 C 处于同一个 super step（同一并行层）。
• 但下面这组分叉的节点数量不一致（C 和 E 数量不同），于是产生了两个 super step：C 运行完通知 D 运行一次；E 运行完又让 D 运行一次。一次运行跑两次 D。
• 而在一个 super step 里的节点只会被运行一次，所以前面那个例子 D 只运行一次。

解决方案：既希望并行、又希望 D 只运行一次，就在 add_edge 时不要把它当整体——把 C 和 E 包起来当作同一个 super step，D 就只会被运行一次了。

个人运维知识库 Agent：三种落地方式

LangGraph 的基础知识讲完，进入进阶——用 LangGraph 实现一个个人运维知识库 Agent，它就像你个人的运维数字分身：可以帮你解答业务方提出的普通问题，或解答那些你没必要亲自思考的问题。只要知识库沉淀下来，就可以用这种方法打造自己的数字分身，给业务方提供服务。

三种落地方式：

1. 不需要代码：自托管部署之前课程提过的 Flowise / Dify / OpenWebUI，把运维知识库放进去做 embedding，就可以对外提供服务。
2. 借助 SaaS 服务：比如字节的 Coze（扣子）、或 Dify，能基于个人运维知识库打造个人运维数字分身（现场会演示怎么用）。
3. 二次开发：需要做二次开发，目的是深度接入公司的业务流程——比如想让 Agent 在内部系统里创建运维工单、读取内部运维数据（日志、指标）、打通 IM 工具（飞书/企业微信），甚至基于日志和指标做联动分析。

方式二演示：用 Coze 打造运维知识库机器人

Coze 是 SaaS 服务，先注册，流程很简单：

1. 登录官网，进入 Personal → Knowledge（知识库），创建一个知识库（如「个人运维知识库」）。
2. Load document：通过上传本地文件创建知识库。把前面用到的运维知识库（data.md）上传，等待处理完成。
3. 创建一个机器人（Bot）（如「个人运维知识库」），在 Knowledge 里点选展开并 Add knowledge，把这个知识库添加进 bot。
4. 可在设置里开启展示原文，这样回答时能看到内容来自哪个知识库。
5. 调试：问「payment 服务是谁维护的？」，它基于知识库直接回答出来——用 Coze 不需要做复杂的开发。
6. 点 Publish 会生成一个公网可访问链接，直接把链接发给业务方，就能基于这个知识库问答。

参数配置（了解即可）：

• 召回策略：默认 auto coze（自动调用 RAG，根据问题自动识别是否需要从 RAG 检索）；on demand（按需调用，但要在提示语里描述什么时候使用知识库，比如遇到运维相关问题就调用，给一些示例）。默认自动方式的一个缺点：有时不够精准——想问知识库的问题，它却偏偏不去知识库查。
• 搜索策略：检索向量库时怎么检索——text（全文检索）、semantic（语义检索）、hybrid（结合全文与语义的优势），保持默认即可。
• 最大召回数量：给大模型多少个片段（默认给 3 或 5 或 10）。注意：传统 RAG 检索出来就拿不到的问题，你配置多少个都没用，它只是一个值的配置。
• 最小匹配度：做向量匹配时的距离/相似度阈值，高于此相似度的块才会被返回，低于的被过滤。

用 LangGraph 实现 Self-RAG 自适应 RAG

最后用 LangGraph 实现一个终极的个人运维知识库数字分身——自适应 RAG Agent（Self-RAG）。它是自适应 RAG Agent，非常强大，且有论文基础支撑。

自适应 RAG 的整体流程

一个问题提出后，先有一个**路由器（Router）**做意图识别与流量分发：

类比：就像公司前台的接待员——先问清楚你来干吗，“要报销"就带到财务、“要换设备"就带到 IT。对应到工程里就是：Router 先用结构化输出判断问题是"知识库检索类"还是"网络搜索类”，然后分流到 Self-RAG 或 Web 搜索两条支线，让每类问题都走最合适的通道。

• 判断问题是知识库检索类问题（向量数据库相关）、还是网络搜索类问题（如「深圳天气怎么样」），分发到 Self-RAG 或 Web 搜索。

为什么叫「自适应 RAG」？因为它的 RAG 过程设计得非常好（内部有多个子循环）：

1. 匹配到对应块后，不直接问答，先判断匹配到的块是不是与问题相关。
2. 如果相关，再做下一步：让大模型基于块尝试生成答案。
3. 生成答案后再判断这个答案能否解决我的问题（是否有幻觉）。没有幻觉就真正回答；有幻觉就重新生成一次答案。
4. 如果向量检索出的文档与问题不匹配，就基于语义重写问题，再重新进入向量检索阶段。
5. 到最后如果发现大模型回答的内容不能解决最初的问题，也重新进入问题重写。

所以 Self-RAG 里有很多个子循环，最终在大部分情况下一定能找到向量库里能提供的那一段或多段内容——效果非常好。

flowchart TD
    Q[用户问题] --> D{Router 路由}
    D -- 运维/知识库相关 --> R[向量检索]
    D -- 通识/时事 --> W[Web 搜索]
    R --> G1{文档与问题相关?}
    G1 -- 不相关 --> RW[重写问题]
    RW --> R
    G1 -- 相关 --> GEN[生成答案]
    GEN --> G2{答案是否基于事实
有无幻觉?}
    G2 -- 有幻觉 --> GEN
    G2 -- 无幻觉 --> G3{是否解决用户问题?}
    G3 -- 否 --> RW
    G3 -- 是 --> ANS[最终回答]

它解决的痛点是：传统 RAG 粗暴检索固定数量段落、准确性差；Self-RAG 显著提升长文本检索的准确性，尤其当块之间还有逻辑关系时更能体现价值。

Web 搜索能力（Tavily）

要实现「网络搜索 + Self-RAG」结合，首先要实现网络搜索能力，让大模型感知现实世界。可以用 Tavily 这个工具执行网络搜索。特点：

• 相比传统搜索引擎返回更快、且自带摘要——不需要加载页面所有内容就能做简单回答。
• 搜索结果针对大模型优化，提高了准确率。
• 是 SaaS 服务，需要在 Tavily 官网申请 API Key。

集成到 LangGraph（Tavily 本身就是一个 Tool，可以直接转成节点）：

from langchain_community.tools.tavily_search import TavilySearchResults

search = TavilySearchResults(max_results=1)   # 搜索工具
model_with_tools = ChatOpenAI(model="gpt-4o", temperature=0).bind_tools([search])

# langgraph 里把工具转成 tools_node，其余图和 ReAct 一样

运行问「2024 年深圳程序员的平均薪酬」——知识库里没有，它会去调用搜索工具，返回结果带链接。搜索能解答知识库和大模型里没有的问题。

完整实现 Self-RAG 的四个评估步骤

用 LangGraph 实现 Self-RAG。这里示例写代码较多，重点讲清四个评估/重写环节：

第一步：评估检索文档与用户问题是否相关（grade_documents_node）

from typing import Literal
from pydantic import BaseModel, Field

class gradeDocuments(BaseModel):
    """Binary score for relevance of retrieved documents."""
    binary_score: str = Field(description="文档与问题相关则 yes，否则 no")

llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)
structured_llm = llm.with_structured_output(gradeDocuments)

# 提示语：你是评分员，评估检索文档与用户问题的相关性，含关键词或语义含义则评为相关，给出二进制 yes/no
relevance_prompt = "你是评分员，负责评估检索到的文档与用户问题的相关性。若包含相关关键字或语义含义，则评为相关，给出二进制 yes 或 no。\n\n文档：\n{document}\n\n问题：\n{question}\n"

# 对每个检索出的块逐个评分，只保留 yes 的块

测试：问相关问题「payment backend 服务是谁维护的」返回 yes；问不相关问题「深圳天气」返回 no；把所有块打印出来都可以用相关性判断。

第二步：生成回复（generate）：复用之前 RAG 的 prompt，用 rag_chain（langchain 链）基于 top 相关的块生成回答。测试「payment backend 服务由小王维护」正常。

第三步：评估大模型回答是否基于事实（有没有幻觉）（grade_generation_vs_documents）：

class gradeModel(BaseModel):
    binary_score: str = Field(description="回答是否有事实支持")

# 提示语：你是评分员，评估 LLM 生成是否基于一组检索到的事实，给出 yes/no
fact_prompt = "评估这个回答是否基于给定文档生成，给出 yes（有事实支持）或 no（无事实支持）。\n\n文档：\n{document}\n\n回答：\n{generation}\n"

测试返回 yes，说明答案基于事实。

第四步：评估是否真正解决了用户问题（grade_answer）：把提示语改成「评估答案是否解决了某个问题，给出 yes/no」。注意一个容易踩的坑——generation 要用第二步生成的 generation 变量，而不是误用评分模型对象，否则会错误返回 no。

重写问题（transform_query）：提示语「你是一个问题重写器，把输入问题转化成针对向量数据库检索进行优化的更好版本，查看输入并推断潜在含义/意图，使用用户语言回复」。运行对比：把「payment backend 服务是谁维护的」改写成「谁负责维护 payment backend 服务」（加了「负责」两字），若文档里有「负责」就更可能被匹配到——改写合理且有效。

组装节点与构建图

定义 GraphState（question、generation、document 三个字段，前两者字符串、document 是列表），然后把上面这些函数封装成 LangGraph 节点：

from langgraph.graph import StateGraph, START, END

def retrieve(state): ...            # 取 question，向量检索，返回 document
def generate(state): ...            # 取 document + question，返回 generation
def grade_documents(state): ...     # 逐个评估，保留相关的块
def transform_query(state): ...     # 重写问题
def decide_to_generate(state): ...
    # 若相关文档为空 -> 返回 "transform_query"；否则 -> 返回 "generate"
def grade_generation(state): ...    # 评估回复（幻觉判定）

workflow = StateGraph(GraphState)
workflow.add_node("retrieve", retrieve)
workflow.add_node("grade_documents", grade_documents)
workflow.add_node("generate", generate)
workflow.add_node("transform_query", transform_query)

workflow.add_edge(START, "retrieve")
workflow.add_edge("retrieve", "grade_documents")
# condition 边：判断是生成回答还是重写问题
workflow.add_conditional_edges(
    "grade_documents", decide_to_generate,
    {"transform_query": "transform_query", "generate": "generate"},
)
workflow.add_edge("transform_query", "retrieve")   # 重写 -> 重新检索（循环）
workflow.add_conditional_edges("generate", grade_generation, {
    "not_supported": "generate",       # 有幻觉 -> 重新生成
    "useful": END,                      # 有用 -> 结束
    "not_useful": "transform_query",    # 无用 -> 重写问题回到检索
})

最终生成的图：retrieve -> grade_documents -> (相关则 generate / 不相关则 transform_query 回 retrieve) -> generate -> (有幻觉重新生成 / 无用重写 / 有用 END)——一个很复杂的循环流程。

运行「谁管理的服务最多」（传统 RAG 回答不了的）：Agent 会经历「改写问题 -> 没有解决 -> 继续改写 -> ……」最终得到正确回复。改写过程虽复杂但最终收敛到正确答案。

整合 WebSearch + Self-RAG：完整运维知识库 Agent

最后把 Web 搜索和 Self-RAG 两个独立流程整合起来，真正实现个人运维知识库 Agent。这样问答时：不懂的知识库内问题走向量检索；知识库外的（如天气）走网络搜索。

• 把一个**路由器（Router）**加在 transform/检索之前。用结构化输出模型定义 RouteQuery：data_source 字段有两个取值——vectorstore（向量数据库）和 web_search。
• 提示语：将用户问题路由到向量数据库或 Web 搜索专家。向量数据库包含运维工单、微服务、网关、工作负载、日志等相关内容，用向量库回答有关主题；否则用 Web 搜索。
• 检查 data_source 分流：是 web_search 就调用 Web 搜索节点（然后 generate）；是 vectorstore 就进入 retrieve。其余 self-RAG 的节点都不变。

flowchart TD
    START --> RQ{route_query 路由}
    RQ -- web_search --> WS[web_search]
    WS --> GE[generate]
    RQ -- vectorstore --> RV[retrieve]
    RV --> GD[grade_documents]
    GD -->|不相关| TQ[transform_query]
    TQ --> RV
    GD -->|相关| GE
    GE --> GG[grade_generation]
    GG -->|not_supported| GE
    GG -->|not_useful| TQ
    GG -->|useful| END

提问两类问题验证：

• 「谁管理的服务数量最多」→ 分流到向量知识库，沿自适 RAG 流程检索到答案；
• 「深圳最近的气温」→ 分支到网络搜索，通过 Tavily 拿到实时信息。

到这里，模块 5 的 Agent 开发内容就结束了：从理解 Agent、四种设计模式，到 RAG Agent 实战、TranslationAgent 分析、LangGraph 框架、LangSmith 追踪，再到个人运维知识库 Agent 的完整落地。

总结

Agent 开发的核心是把「大模型推理」与「工具/知识库/循环」结合成一个能自主闭环的工程框架，不是一句 prompt 就能跑起来的。本章主线：

• Agent 本质 = 规划 + 记忆 + 工具 + 行动；是把大模型从"光有大脑"扩展成"有手有脚有记忆、能独立行动”。短期记忆来自有限上下文，长期记忆靠外部数据库；工具层把查询日志/指标/工单/部署等运维能力封装成 API。
• 四种设计模式（吴恩达）：反馈（反思+迭代）、工具调用、规划（拆子任务）、多智能体协作——各有优劣，实际 Agent 组合使用。
• RAG Agent 实战：用「循环加块 + 大模型反思」解决传统 RAG「检索不相关 + 幻觉」两大短板；改进方向是切片防超限、offset 过滤无用块、JSON Mode 稳定输出。
• TranslationAgent：三节点串行的反思工作流（初始翻译 → 反思 → 优化），提示语设计值得参考；但它无循环，改进空间是加入"还能再优化就回炉"的自循环。
• LangGraph：State/Node/Edge 三大概念；add_node 建节点、add_edge/conditional_edge 串起来；set_entry_point/finish_point 定起止；用 add_messages 管理消息状态；能实现单节点、串行、工具循环（ReAct），并用 super step 理解并行节点的执行次数。
• LangSmith：Agent 开发的调试日志/追踪平台，四行代码集成，可观察每节点输入输出与调用链。
• 运维数字分身：三种落地（无代码自托管 Flowise/Dify/OpenWebUI、SaaS 的 Coze、二次开发）；用 LangGraph 实现 Self-RAG 自适应 RAG（评估相关性 → 评估是否基于事实 → 评估是否解决问题 → 兜底重写问题），再叠加 Tavily Web 搜索路由，形成完整的个人运维知识库 Agent。

说实话，这一章最值钱的不是某一句提示语怎么写的，而是**“循环 + 自主决策"这套思维方式**。传统 RAG 是"一把梭检索完就答”，RAG Agent 学会了"查不到就换个查询再试"；TranslationAgent 学会"翻完自己复查一遍"。把这些循环落到 LangGraph 上，你才真正配得上"我在开发 Agent"这句话。

自测题与动手练习

动手练习（建议真做一遍）：

1. 从零实现一个 RAG Agent，用「循环加块 + 大模型反思」回答传统 RAG 答不出的问题（如"谁管理的系统最多"），并尝试给它加上"切片防超限 / offset 过滤 / JSON Mode"三个改进之一。
2. 克隆运行吴恩达 TranslationAgent，翻译一段英文古诗成中文，再写一个最简 ChatGPT 翻译脚本，运行两次对比稳定性。
3. 用 LangGraph 搭一个单节点 Chat Agent 和一个三节点串行"翻译→反思→优化"Agent，用 display(Image(graph.get_graph().draw_mermaid_png())) 画出两张图对比。
4. 用 LangGraph + conditional_edge 实现一个 ReAct 工具循环 Agent（get → patch 修改工作负载），接入 LangSmith 观察每一步的输入输出与函数调用。
5. 用 LangGraph 实现 Self-RAG（相关性/事实/解决问题三级评估 + 重写兜底），再在检索前加一个 Web 搜索路由，分别问"谁管理的服务数量最多"和"深圳最近的气温"，观察分流结果。

本章小结