一、学习目标
- 学习目标：理解智能体系统的必要性与分层架构；区分内容块（ContentBlocks）与上下文（Context）；
- 说明何时采用 Agent 编排而非单次模型调用。
关键术语
- Agent：智能体，组合模型、工具与策略形成可执行的系统。
- Tool：工具，以
@tool声明的原子能力函数，支持类型校验与独立测试。 - Middleware（中间件）：包裹模型/工具调用，实现路由、裁剪、审计、错误防护等。
- Content Blocks（内容块）：标准化消息内容块，统一文本、工具调用与多模态引用的表示。
- Context（执行上下文）：通过
invoke/stream的context参数注入（如租户、用户画像、灰度桶），不进入提示词文本本身；若将上下文写入消息内容，则会计入 token 与成本。 - Static/Dynamic Model（静态/动态模型）：静态模型在生命周期内固定；动态模型在运行时依据策略与状态进行路由与降级。
二、什么是智能体系统
- 从“单次模型调用”发展为“具备工具、策略、状态与可观测的系统”。
- 核心构件：模型（Model）、工具（Tool）、中间件（Middleware）、上下文（Context）、结构化输出（Structured Output）、可观测（Observability）。
- 价值：可控与可扩展（错误防护、能力路由、合规审计）、稳定与韧性（重试、超时、断路器、降级与回退）、可观测（日志、指标、Trace）与可运维（报警、SLA 管理）。
三、为什么不仅仅是模型调用
- 场景复杂度：现实任务常需要读取外部数据、执行操作、返回结构化结果，而非纯文本。
- 不确定性治理：模型速率限制、短时失败与响应波动，需要重试、降级与断路器等工程策略。
- 组织需求：审计、合规、权限与安全隔离需要统一的系统化工程模式。
四、分层架构
智能体系统分层架构（LangChain 1.0，Python）。
数据流：用户输入在接口层被标准化为消息与执行上下文 → Agent 编排层按策略选择模型与工具并进入主推理循环 → 中间件在模型/工具调用前后拦截，执行路由、裁剪、审计与降级 → 主循环生成并解析结构化输出，返回结果与引用信息。
拦截与可观测：wrap_model_call负责动态路由与稳定性（重试、超时、断路器、降级与回退），wrap_tool_call负责参数校验、错误防护与审计留痕；所有关键事件打点到日志、指标与 Trace，支持问题回放与合规审计。
- 接口层：对外提供 API/UI；任务输入输出与会话管理。
- 输入/输出与职责：接收用户输入与会话状态，输出标准化任务与结果展示。
- Agent 编排层：将模型、工具与策略组合为可执行的工作流。
- 输入/输出与职责：输入任务与上下文，输出可执行工作流与结构化结果。
- 策略/中间件：在模型与工具调用前后注入规则（提示裁剪、脱敏、审计、动态模型路由等）。
- 输入/输出与职责：输入待执行调用与治理规则，输出经治理的调用与审计记录。
- 模型层：统一的
invoke/stream/batch调用，返回标准化消息内容块（content blocks）。
- 输入/输出与职责：输入标准化消息与上下文，输出 content blocks。
- 工具层：以
@tool暴露原子能力，参数校验与错误防护可在中间件实现。
- 输入/输出与职责：输入参数，输出执行结果或错误信息。
- 上下文与状态：通过
context注入租户、用户画像与灰度信息（不计入 token）；会话状态用于策略决策。
- 输入/输出与职责：输入租户/画像/灰度，输出策略决策参考（不计入 token）。
- 结构化输出：在主推理循环产生与解析结构化结果（避免预绑定工具限制）。
- 输入/输出与职责：输入原始文本与工具结果，输出 JSON/Schema 结构。
- RAG/记忆与缓存：知识检索与重排、对话记忆与缓存命中提升效率与质量。
- 输入/输出与职责：输入查询与历史，输出候选知识与缓存命中结果。
- 可观测与稳定性：日志、指标与 Trace；重试、超时、断路器与降级策略保障 SLA。
- 输入/输出与职责：输入关键事件，输出日志、指标、Trace 与稳定性动作。
五、适用场景与边界
场景适配与边界：
| 对比项 | 适用示例 | 非适用示例 | 迁移建议 |
|---|---|---|---|
| 任务流程化 | 多步骤执行、需外部工具与数据协作 | 单轮纯文本生成 | 使用最小模型调用，减少编排开销 |
| 结构化结果 | 表格/JSON 响应与引用信息 | 仅需自由文本 | 在后处理阶段进行简单模板化即可 |
| 高可靠性 | 需可观测、降级与回退的业务逻辑 | 强事务一致性核心交易写入 | 采用传统事务系统封装，与智能体系统异步集成 |
| 策略治理 | 敏感操作需审计与人工审核（HITL） | 无敏感操作或低风险场景 | 简化为模型服务 + 轻量鉴权层 |
六、最小能力集（交付最小可用系统）
- 模型调用：统一
invoke/stream，返回content_blocks。 - 工具集成：以
@tool声明至少 2 个原子能力，并单测覆盖边界与异常路径。 - 中间件防护：
wrap_model_call路由/降级，wrap_tool_call错误处理与审计。 - 上下文注入：
context提供租户/画像/灰度，不计入 token。 - 结构化输出：主推理循环生成 JSON/Schema，并提供解析与容错策略。
- 可观测：最小日志/指标/Trace 打点，记录关键事件（模型切换、工具调用、解析失败）。
七、示例：最小 invoke 与 content_blocks 归一
from langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage# 依赖：OPENAI_API_KEY，版本以示例仓库的 requirements.txt 为准model = ChatOpenAI(model="gpt-4o-mini", temperature=0)prompt = HumanMessage(content=[{"type": "text", "text": "用两点说明智能体系统的价值。"}])# 单次调用（invoke）ai_msg = model.invoke([prompt])# 归一化为 content_blocks 列表，统一不同返回形态（字符串/列表）def normalize_content_blocks(message): if isinstance(message.content, list): return message.content return [{"type": "text", "text": str(message.content)}]blocks = normalize_content_blocks(ai_msg)for b in blocks: if b.get("type") == "text": print(b.get("text", "")) else: print(f"{b.get('type')}: {b}")# 流式调用（stream，可选）：按片段打印文本for chunk in model.stream([prompt]): text = chunk.content if isinstance(chunk.content, str) else"" if text: print(text, end="")八、示例：仅模型调用 vs Agent 编排
对比目标：展示“仅模型调用”与“带工具的 Agent 编排”在能力与可控性上的差异。
- 仅模型调用：直接生成文本，无法执行算术/外部动作，结果不可验证。
- Agent 编排：通过
@tool暴露原子能力，主循环处理tool_calls，可执行、可审计、可回放。
from langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, ToolMessagefrom langchain_core.tools import tool# 工具：返回两数之和（原子能力，可单测）@tooldef add(a: int, b: int) -> int: """返回两数之和。""" return a + b# 仅模型调用：无工具、直接生成文本plain_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0)plain_res = plain_llm.invoke([HumanMessage(content="计算 12+30，并解释思路。")])print("仅模型调用:", plain_res.content)# Agent 编排：绑定工具并在主循环处理 tool_callsagent_llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([add])msgs = [HumanMessage(content="请用工具计算 12+30，并给出一步步解释。")]ai = agent_llm.invoke(msgs)msgs.append(ai)for tc in ai.tool_calls or []: if tc["name"] == "add": result = add.invoke(tc["args"]) # 执行工具 msgs.append(ToolMessage(content=str(result), tool_call_id=tc["id"]))final = agent_llm.invoke(msgs)print("Agent 编排:", final.content)九、示例：外部 API 工具
目标：体现“外部动作”与工程可观测/可回放。示例以汇率查询 API 为工具，展示调用日志与简单回放。
import timeimport requestsfrom langchain_openai import ChatOpenAIfrom langchain_core.messages import HumanMessage, ToolMessagefrom langchain_core.tools import tool# 外部 API 工具：查询两币种间的实时汇率（使用公开接口）@tooldef get_exchange_rate(base: str, target: str) -> float: """返回 base 到 target 的近实时汇率。""" url = f"https://api.exchangerate.host/latest?base={base}&symbols={target}" try: resp = requests.get(url, timeout=8) resp.raise_for_status() data = resp.json() return float(data["rates"][target]) except Exception as e: # 将异常抛回，交由工具中间件或主循环记录审计 raise RuntimeError(f"get_exchange_rate_failed: {e}")# 绑定工具，演示可观测日志与回放llm = ChatOpenAI(model="gpt-4o-mini", temperature=0).bind_tools([get_exchange_rate])msgs = [HumanMessage(content="请用工具查询 USD→CNY 汇率，并解释来源可靠性。")]ai = llm.invoke(msgs)msgs.append(ai)for tc in ai.tool_calls or []: if tc["name"] == "get_exchange_rate": start = time.monotonic() try: rate = get_exchange_rate.invoke(tc["args"]) # 外部 API 调用 latency_ms = int((time.monotonic() - start) * 1000) print(f"[tool_log] name=get_exchange_rate args={tc['args']} result={rate} latency_ms={latency_ms}") msgs.append(ToolMessage(content=str(rate), tool_call_id=tc["id"])) except Exception as err: print(f"[tool_error] name=get_exchange_rate args={tc['args']} error={err}") msgs.append(ToolMessage(content=f"ERROR: {err}", tool_call_id=tc["id"]))# 回放：打印消息序列（可用于审计/复盘）for m in msgs: role = m.type if hasattr(m, "type") else m.__class__.__name__ print(f"[replay] role={role} content={str(m.content)[:80]}")final = llm.invoke(msgs)print("外部工具示例:", final.content)注意：实际代码需要考虑以下情况
- 失败重试至少 2 次（含指数退避与错误分类）。
- 工具异常有审计记录（包含工具名、参数、错误类型与时间）。
- 结构化解析失败进入回退分支（安全输出与日志留痕）。
差异摘要：
| 对比维度 | 仅模型调用 | Agent 编排 |
|---|---|---|
| 能力范围 | 只能描述，无法执行外部动作 | 可调用工具执行具体动作 |
| 可验证性 | 结果难以验证 | 工具返回值可观测与审计 |
| 失败处理 | 难以工程化治理 | 支持中间件重试、超时、降级与回退 |
| 成本与复杂度 | 成本低、实现简单 | 工程成本更高、管线更复杂 |
| 延迟与并发 | 延迟稳定、并发简单 | 工具/治理增加延迟，可用并发与缓存优化 |
| 安全与合规 | 内容审核较弱 | 审计日志、权限与 HITL 审核可插桩 |
| 可维护性 | 难以回放与定位问题 | Trace/日志支持复盘，利于持续演进 |
如何学习大模型 AI ？
由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。
但是具体到个人，只能说是：
“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。
这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。
我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。
我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。
第一阶段（10天）：初阶应用
该阶段让大家对大模型 AI有一个最前沿的认识，对大模型 AI 的理解超过 95% 的人，可以在相关讨论时发表高级、不跟风、又接地气的见解，别人只会和 AI 聊天，而你能调教 AI，并能用代码将大模型和业务衔接。
- 大模型 AI 能干什么？
- 大模型是怎样获得「智能」的？
- 用好 AI 的核心心法
- 大模型应用业务架构
- 大模型应用技术架构
- 代码示例：向 GPT-3.5 灌入新知识
- 提示工程的意义和核心思想
- Prompt 典型构成
- 指令调优方法论
- 思维链和思维树
- Prompt 攻击和防范
- …
第二阶段（30天）：高阶应用
该阶段我们正式进入大模型 AI 进阶实战学习，学会构造私有知识库，扩展 AI 的能力。快速开发一个完整的基于 agent 对话机器人。掌握功能最强的大模型开发框架，抓住最新的技术进展，适合 Python 和 JavaScript 程序员。
- 为什么要做 RAG
- 搭建一个简单的 ChatPDF
- 检索的基础概念
- 什么是向量表示（Embeddings）
- 向量数据库与向量检索
- 基于向量检索的 RAG
- 搭建 RAG 系统的扩展知识
- 混合检索与 RAG-Fusion 简介
- 向量模型本地部署
- …
第三阶段（30天）：模型训练
恭喜你，如果学到这里，你基本可以找到一份大模型 AI相关的工作，自己也能训练 GPT 了！通过微调，训练自己的垂直大模型，能独立训练开源多模态大模型，掌握更多技术方案。
到此为止，大概2个月的时间。你已经成为了一名“AI小子”。那么你还想往下探索吗？
- 为什么要做 RAG
- 什么是模型
- 什么是模型训练
- 求解器 & 损失函数简介
- 小实验2：手写一个简单的神经网络并训练它
- 什么是训练/预训练/微调/轻量化微调
- Transformer结构简介
- 轻量化微调
- 实验数据集的构建
- …
第四阶段（20天）：商业闭环
对全球大模型从性能、吞吐量、成本等方面有一定的认知，可以在云端和本地等多种环境下部署大模型，找到适合自己的项目/创业方向，做一名被 AI 武装的产品经理。
- 硬件选型
- 带你了解全球大模型
- 使用国产大模型服务
- 搭建 OpenAI 代理
- 热身：基于阿里云 PAI 部署 Stable Diffusion
- 在本地计算机运行大模型
- 大模型的私有化部署
- 基于 vLLM 部署大模型
- 案例：如何优雅地在阿里云私有部署开源大模型
- 部署一套开源 LLM 项目
- 内容安全
- 互联网信息服务算法备案
- …
学习是一个过程，只要学习就会有挑战。天道酬勤，你越努力，就会成为越优秀的自己。
如果你能在15天内完成所有的任务，那你堪称天才。然而，如果你能完成 60-70% 的内容，你就已经开始具备成为一名大模型 AI 的正确特征了。
 - 集成企业微信，使用企业微信扫码登录soular)
