大模型Agent开发宝典：从基础架构到高级实践，一篇搞定智能体系统开发，建议收藏反复学习！

本文详细介绍了智能体系统的概念、分层架构和核心组件（模型、工具、中间件、上下文等），对比了仅模型调用与Agent编排的差异。文章指出，Agent系统在需要外部工具调用、结构化输出和高可靠性场景下更具优势，提供了代码示例和最小交付要素，强调了其在可控性、可观测性和稳定性方面的工程价值，并给出了适用场景边界和取舍原则。

m0_63171455

680人浏览 · 2025-11-10 08:15:00

m0_63171455 · 2025-11-10 08:15:00 发布

简介

学习目标

学习目标：理解智能体系统的必要性与分层架构；区分内容块（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/日志支持复盘，利于持续演进

参考与延伸阅读

LangChain（Python）官方文档概览：https://docs.langchain.com/oss/python/langchain/overview
Agents 模块（Python）：https://docs.langchain.com/oss/python/langchain/agents
Tools 工具集（Python）：https://docs.langchain.com/oss/python/langchain/tools
结构化输出（Structured Output）：https://docs.langchain.com/oss/python/langchain/structured-output
RAG 教程（Python）：https://docs.langchain.com/oss/python/langchain/rag

小结

工程化价值：通过工具、中间件、上下文与结构化输出，获得可控、可观测与稳定性。
取舍原则：当需外部动作、结构化结果或明确 SLA 时采用 Agent；单轮纯文本优先仅模型调用（参见“差异摘要”）。
分层主线：接口/编排 → 策略/中间件 → 模型/工具 → 上下文/结构化输出 → 观测与稳定性（职责清晰、可治理）。
最小交付要素：invoke/stream、@tool 原子能力、中间件治理、context 注入与结构化输出。

如何学习大模型 AI ？

由于新岗位的生产效率，要优于被取代岗位的生产效率，所以实际上整个社会的生产效率是提升的。

但是具体到个人，只能说是：

“最先掌握AI的人，将会比较晚掌握AI的人有竞争优势”。

这句话，放在计算机、互联网、移动互联网的开局时期，都是一样的道理。

我在一线互联网企业工作十余年里，指导过不少同行后辈。帮助很多人得到了学习和成长。

我意识到有很多经验和知识值得分享给大家，也可以通过我们的能力和经验解答大家在人工智能学习中的很多困惑，所以在工作繁忙的情况下还是坚持各种整理和分享。但苦于知识传播途径有限，很多互联网行业朋友无法获得正确的资料得到学习提升，故此将并将重要的AI大模型资料包括AI大模型入门学习思维导图、精品AI大模型学习书籍手册、视频教程、实战学习等录播视频免费分享出来。

一直在更新，更多的大模型学习和面试资料已经上传带到CSDN的官方了，有需要的朋友可以扫描下方二维码免费领取【保证100%免费】👇👇

在这里插入图片描述