拓十年匠心定制 · 商业建站与技术教学双线并行 咨询热线:400-886-1026 service@lmnt.cn
ARTICLE DETAIL

资讯详情

深耕网站建设与运营推广的一线实战洞察。

RAG、Agent、MCP实战:大模型应用开发核心概念与代码实现

RAG、Agent、MCP实战:大模型应用开发核心概念与代码实现

这两年大模型应用开发变化非常快,几乎每隔几个月就会冒出新的概念和框架。很多初学者跟我反馈,说在网上找资料时经常被 RAG、Agent、MCP 这几个词绕晕,不知道先学哪个,也不知道它们之间到底是什么关系。这篇文章想解决的,就是这个问题。

我会抛开营销号式的“三天入门、五天精通”话术,踏踏实实把 AI 大模型应用开发这条路上的三个核心主题一次讲透:RAG 知识库、Agent 智能体、MCP 协议。从概念原理讲到环境搭建,再给出完整可运行的实战代码和常见排错方案。无论你是刚开始接触大模型的新手,还是已经在做应用开发的工程师,都可以把这篇文章当作一份系统化的学习索引与实战手册。

1. 背景与核心概念

1.1 大模型应用开发到底在做什么

先理清一个大前提:大模型本身不是一个“开箱即用”的完整应用。

你有一个很强大的模型,它能写代码、能回答问题、能总结文档。但要把它变成一个真正面向用户的业务系统,还需要解决几件事:

  • 模型不知道你私有的业务数据,比如公司内部文档、产品说明书、用户手册。
  • 模型的知识有截止时间,它无法知道最近发生的事情。
  • 模型不能主动调用外部系统,比如查询天气、下单、操作数据库。
  • 模型无法访问实时网页,无法读你本地的文件。
  • 模型响应速度慢,用户体验需要流式输出。

这些工程问题,就是“大模型应用开发”的核心工作。而 RAG、Agent、MCP,恰好是从三个不同维度解决这些问题的方案。

1.2 RAG、Agent、MCP 分别解决什么问题

我们先打个比方,帮助理解:

  • RAG(Retrieval-Augmented Generation,检索增强生成)相当于给模型配了一本“参考书”。模型回答问题之前,先从这本参考书里找到相关段落,再基于这些材料组织答案。它解决的是“知识缺失”和“事实幻觉”问题。
  • Agent(智能体)相当于给模型配了“手脚”。模型不再只是说几句话,而是可以规划任务、调用工具、执行动作,比如查数据库、发请求、操作接口,最终完成任务。它解决的是“只会说不会做”的问题。
  • MCP(Model Context Protocol,模型上下文协议)相当于给模型和外部工具之间定了一份“标准化接口协议”。它解决的是“工具接入方式五花八门”的问题。以前每个工具都要写一套定制对接代码,现在按 MCP 标准暴露服务,任何支持 MCP 的客户端都能直接使用。

一张表总结:

概念解决的问题类比
RAG私有知识、时效性、幻觉参考书
Agent任务规划与工具调用手脚
MCP工具接入标准化插座与插头

需要注意的是,这三个概念不是互斥的,而是经常组合使用。RAG 让 Agent 有知识支撑,MCP 让 Agent 能接入更多工具。下文会有组合实战演示。

1.3 为什么新手容易走弯路

根据我观察到的新手学习路径,最容易踩的坑有两个。

第一个坑是“只学概念不写代码”。看了很多架构图、流程图,觉得自己明白了,一写代码就发现环境装不上、依赖冲突、接口报错。所以本文在概念讲完后,会立刻进入实战。

第二个坑是“过早陷入某个框架细节”。比如一上来就钻研某个 Agent 框架的源码,或者纠结向量数据库选型。对于零基础学习,更重要的是先建立完整链路认知:数据从哪来、模型怎么调用、答案怎么生成、工具怎么接入。框架只是工具,核心思想才是不变的。

2. 环境准备与版本说明

2.1 开发环境与语言版本

本文示例以 Python 3.10+ 为例,操作系统使用 Windows / macOS / Linux 均可。建议使用虚拟环境隔离项目依赖,避免污染系统 Python。

python -m venv venv # Windows venv\Scripts\activate # macOS / Linux source venv/bin/activate

涉及前端渲染的部分,会用到原生 JavaScript,不需要额外框架。你只需要一个现代浏览器即可查看效果。

2.2 安装依赖

本教程会用到的 Python 库如下:

  • openai:调用 OpenAI 格式的大模型接口,国内大多数厂商的 API 也兼容该格式。
  • langchain:RAG 链路的快速搭建工具,注意当前版本迭代较快,以你安装时的最新稳定版为准。
  • faiss-cpu或chromadb:向量存储与检索。
  • requests/httpx:HTTP 请求。
  • flask或fastapi:用于搭建本地演示服务。

安装命令:

pip install openai langchain faiss-cpu chromadb flask httpx

这里有一个非常实用的建议:不要一次性装一堆库,而是根据下面的实战章节需要,一个一个安装。否则很容易出现版本冲突,且你根本不知道是哪个库出了问题。

2.3 示例项目结构

我们整个实战部分会围绕一个“智能客服助手”来展开,项目结构如下:

ai-tutorial/ ├── venv/ ├── requirements.txt ├── 01_llm_basic.py # 大模型基础调用 ├── 02_stream_output.py # 流式输出 ├── 03_rag_pipeline.py # RAG 知识库完整链路 ├── 04_agent_tool.py # Agent 工具调用 ├── 05_mcp_server.py # MCP Server 示例 ├── docs/ # 存放知识库文档 │ └── product_manual.txt # 产品说明文档 └── server/ ├── app.py # Flask 演示服务 └── templates/ └── index.html # 流式渲染页面

这个结构是逐步演进的,每新增一个能力就对应新增一个脚本文件,方便你对照学习。

3. 大模型接入与 Stream 流式输出

无论做 RAG、Agent 还是 MCP,第一步都是学会与大模型对话。这一节我们解决三个问题:怎么调用 API、怎么拿到流式输出、怎么在网页上实时渲染。

3.1 大模型 API 调用基础

现在绝大多数大模型厂商都提供 OpenAI 兼容接口,所以使用openai库是最通用的方式。

# 文件路径:01_llm_basic.py from openai import OpenAI client = OpenAI( api_key="your-api-key", base_url="https://your-model-provider.com/v1" ) response = client.chat.completions.create( model="your-model-name", messages=[ {"role": "system", "content": "你是一个严谨的技术助手。"}, {"role": "user", "content": "用一句话解释什么是RAG。"} ] ) print(response.choices[0].message.content)

这里有几个重要参数需要解释。

model参数需要替换为你所使用厂商提供的模型名称,例如常见的gpt-4o-mini、qwen-plus、deepseek-chat等,具体以你的 API 文档为准。base_url是 API 服务地址,如果你使用的是国内云厂商的模型服务,就填对应平台的地址。

messages是对话消息列表,包含system(系统提示词,用于设定角色和规则)和user(用户输入)两种角色。系统提示词非常关键,它决定了模型的行为边界和回答风格。

response.choices[0].message.content是模型返回的文本内容。这种方式是“一次性返回”,也就是要等模型把内容全部生成完毕后才能拿到结果。对于短文本没问题,但对长回答体验较差,我们需要使用流式输出。

3.2 使用 SSE 流式输出

流式输出的原理是:模型每生成一小段文本,服务端就通过 SSE(Server-Sent Events,服务器发送事件)推送给客户端。客户端收到后立即渲染。这样用户看到的效果是“字一个一个蹦出来”,不用干等。

服务端使用 Flask 实现 SSE:

# 文件路径:server/app.py from flask import Flask, render_template, Response, request from openai import OpenAI app = Flask(__name__) client = OpenAI( api_key="your-api-key", base_url="https://your-model-provider.com/v1" ) @app.route("/") def index(): return render_template("index.html") @app.route("/chat") def chat(): user_input = request.args.get("msg", "") def generate(): response = client.chat.completions.create( model="your-model-name", messages=[ {"role": "system", "content": "你是一个乐于助人的助手。"}, {"role": "user", "content": user_input} ], stream=True ) for chunk in response: delta = chunk.choices[0].delta if delta and delta.content: yield f"data: {delta.content}\n\n" yield "data: [DONE]\n\n" return Response(generate(), mimetype="text/event-stream") if __name__ == "__main__": app.run(port=5000, debug=True)

关键点在于设置了stream=True,然后遍历response中的每一个chunk。每个chunk里包含一小段增量文本delta.content。我们用 SSE 格式data: 内容\n\n推送出去。

3.3 前端流式渲染与 AbortController

前端我们使用fetch读取 SSE 流,并用AbortController实现“停止生成”功能。

<!-- 文件路径:server/templates/index.html --> <!DOCTYPE html> <html lang="zh"> <head> <meta charset="UTF-8"> <title>大模型流式问答</title> </head> <body> <h2>流式问答演示</h2> <div id="output" style="white-space: pre-wrap; border: 1px solid #ccc; padding: 12px;"></div> <input id="input" placeholder="请输入问题" style="width: 300px; margin-top: 10px;"> <button id="sendBtn">发送</button> <button id="stopBtn">停止</button> <script> let controller = null; async function sendMessage() { const input = document.getElementById('input'); const output = document.getElementById('output'); const msg = input.value.trim(); if (!msg) return; output.textContent = ''; controller = new AbortController(); try { const response = await fetch('/chat?msg=' + encodeURIComponent(msg), { signal: controller.signal }); const reader = response.body.getReader(); const decoder = new TextDecoder(); while (true) { const { done, value } = await reader.read(); if (done) break; const text = decoder.decode(value, { stream: true }); const lines = text.split('\n'); for (const line of lines) { if (line.startsWith('data: ')) { const data = line.slice(6); if (data === '[DONE]') continue; output.textContent += data + ' '; } } } } catch (err) { if (err.name === 'AbortError') { output.textContent += '\n[已停止生成]'; } else { console.error(err); } } } document.getElementById('sendBtn').addEventListener('click', () => { if (controller) controller.abort(); sendMessage(); }); document.getElementById('stopBtn').addEventListener('click', () => { if (controller) controller.abort(); }); </script> </body> </html>

这里的关键设计:

  • AbortController用于中断fetch请求。当用户点击“停止”按钮时,调用controller.abort(),浏览器的网络请求会被取消,服务端生成也会中断。
  • reader.read()循环读取响应流,每次拿到一个 Uint8Array 类型的数据块,用TextDecoder解码成字符串。
  • 由于 SSE 数据可能被拆分成多个 Fragment,所以解析时注意按换行符切分,并且只处理data:开头的行。

这个场景非常典型:基于 SSE 流式输出实现大模型回答实时渲染,配合 abort 实现中断,几乎是目前所有大模型对话产品的前端标配。

4. RAG 知识库实战

4.1 RAG 的完整链路

RAG 的完整链路可以拆成两个阶段:

离线索引阶段:

  1. 加载文档(PDF、TXT、Word、网页等)。
  2. 将长文档切分成片段(Chunk)。
  3. 对每个片段做向量化(Embedding),把文本转成向量。
  4. 将向量存入向量数据库。

在线问答阶段:

  1. 用户输入问题。
  2. 对问题做同样的向量化。
  3. 在向量数据库中做相似度检索,找到最相关的几个片段。
  4. 把问题和相关片段一起拼进 Prompt,发给大模型。
  5. 大模型基于片段内容生成回答。

这个链路看起来不复杂,但每一步都有细节。下面我们用一个本地文档来走通全流程。

4.2 文档加载与切分

首先准备一个简单的产品文档,放到docs/product_manual.txt里。内容可以是一些与你的业务相关的说明文字,这里我们假设是产品使用手册。

用户手册: 本产品支持录音转写功能。录音文件格式支持 MP3、WAV、M4A,单文件最大 500MB。 转写任务提交后,系统会异步处理。任务状态包括:排队中、识别中、完成、失败。 识别完成后,用户可以在控制台下载转写文本和字幕文件。 本产品支持自动识别说话人,最多支持 10 个说话人分离。 隐私说明:录音文件在识别任务完成后 24 小时内自动删除。

下面我们写 RAG 链路的核心代码:

# 文件路径:03_rag_pipeline.py from langchain.text_splitter import RecursiveCharacterTextSplitter from openai import OpenAI import numpy as np client = OpenAI( api_key="your-api-key", base_url="https://your-model-provider.com/v1" ) def load_and_split(file_path, chunk_size=200, chunk_overlap=50): with open(file_path, "r", encoding="utf-8") as f: text = f.read() splitter = RecursiveCharacterTextSplitter( chunk_size=chunk_size, chunk_overlap=chunk_overlap, separators=["\n\n", "\n", "。", "!", "?", " "] ) chunks = splitter.split_text(text) return chunks chunks = load_and_split("docs/product_manual.txt") print(f"切分出 {len(chunks)} 个片段") for i, c in enumerate(chunks): print(f"--- 片段 {i+1} ---") print(c)

这里解释两个核心参数。

chunk_size是每个片段的字符数上限。片段太长会导致检索不精准,因为一个片段里混杂了多个主题;片段太短则可能导致上下文不足。一般经验值是 200~500 字,需根据你的文档类型调整。

chunk_overlap是相邻片段之间的重叠字符数。设置重叠可以避免“一句话被拦腰截断”的问题,保证关键信息不丢。

RecursiveCharacterTextSplitter是 LangChain 提供的常用切分器,它会优先按段落切,再按句子切,再按字符切,尽可能保持语义完整。

4.3 向量化与检索

有了文本片段后,下一步是对片段做向量化。这里我们使用模型提供商的 Embedding 接口来生成向量,然后用 NumPy 数组模拟向量库,避免引入过重的数据库依赖。

# 继续在 03_rag_pipeline.py 中追加 def embed_texts(texts): resp = client.embeddings.create( model="your-embedding-model-name", input=texts ) return [item.embedding for item in resp.data] def cosine_similarity(a, b): a = np.array(a) b = np.array(b) return np.dot(a, b) / (np.linalg.norm(a) * np.linalg.norm(b)) def search(query, chunk_list, top_k=2): query_vec = embed_texts([query])[0] chunk_vecs = embed_texts(chunk_list) scores = [] for i, vec in enumerate(chunk_vecs): score = cosine_similarity(query_vec, vec) scores.append((score, i, chunk_list[i])) scores.sort(key=lambda x: x[0], reverse=True) return scores[:top_k] query = "录音文件支持什么格式?" results = search(query, chunks, top_k=2) print("检索结果:") for score, idx, text in results: print(f"{score:.4f} -> {text}")

Embedding模型和对话模型一般是分开的。对话模型负责生成回答,Embedding 模型负责把文本转换成向量。具体的模型名称需要参考你的平台文档,例如text-embedding-3-small或国产平台对应的 Embedding 模型名。

余弦相似度是常用的向量相似度计算方法。值越接近 1,说明两个向量在语义空间越接近。Embedding 的核心思想是:语义相近的文本,向量距离也近。

在实际项目中,你不会手动用 NumPy 实现向量检索。当数据量增长到几万条以上时,需要使用专门的向量数据库,例如 Chroma、FAISS、Milvus 或 pgvector。但理解上面的原理,对你选择合适的数据库、调参和排错都很有帮助。

4.4 基于检索结果生成回答

检索到相关片段后,最后一步是把片段和用户问题组装成一个 Prompt,交给大模型生成最终答案。

# 继续在 03_rag_pipeline.py 中追加 def rag_answer(query, chunk_list): results = search(query, chunk_list, top_k=2) context = "\n\n".join([text for _, _, text in results]) prompt = f"""你是产品客服助手。请基于以下参考资料回答问题。 如果参考资料中没有答案,请直接说明资料中没有相关信息,不要编造。 参考资料: {context} 用户问题: {query} """ response = client.chat.completions.create( model="your-model-name", messages=[ {"role": "system", "content": "你是严格的客服助手。"}, {"role": "user", "content": prompt} ] ) return response.choices[0].message.content answer = rag_answer(query, chunks) print("回答:", answer)

很多初学者会直接把检索出来的大段文本全部塞给模型。这个做法有两个问题:一是超出上下文窗口限制,二是无关信息会干扰模型判断。正确的做法是“先精选,再拼接”,只选择相关性最高的 2~3 个片段。

在实际生产系统中,还需要做更多的工程优化,比如混合检索、重排、元数据过滤等。但先把这条最简单的 RAG 链路跑通,是理解后续一切优化的基础。

5. Agent 智能体实战

5.1 Agent 的核心机制

Agent 与普通 Prompt 调用的最大区别在于:Agent 可以让模型“决定”调用哪些工具,然后根据工具返回的结果继续推理。

典型的工作循环是:

  1. 用户提出任务。
  2. 模型分析任务,决定是否需要调用工具。
  3. 如果需要调用工具,模型输出一个结构化的“工具调用请求”。
  4. 程序执行该工具,把结果返回给模型。
  5. 模型根据工具结果生成最终回答,或者继续调用下一个工具。

这个能力在 API 层面通常被称为 Function Calling(函数调用)或 Tool Calling。很多国产模型也支持这一能力,只是名字可能略有不同。

5.2 Function Calling 示例

我们先定义一个简单的工具函数:查询订单状态。

# 文件路径:04_agent_tool.py import json from openai import OpenAI client = OpenAI( api_key="your-api-key", base_url="https://your-model-provider.com/v1" ) # 模拟订单数据库 orders_db = { "1001": {"status": "已发货", "eta": "2026-01-20", "carrier": "顺丰"}, "1002": {"status": "待支付", "eta": None, "carrier": None}, "1003": {"status": "已签收", "eta": "2026-01-18", "carrier": "中通"}, } def query_order_status(order_id: str) -> str: """查询订单状态""" if order_id in orders_db: return json.dumps(orders_db[order_id], ensure_ascii=False) return json.dumps({"error": "订单不存在"}, ensure_ascii=False)

接着定义工具描述,并调用模型:

tools = [ { "type": "function", "function": { "name": "query_order_status", "description": "查询订单状态,输入订单ID,返回物流状态和预计到达时间", "parameters": { "type": "object", "properties": { "order_id": { "type": "string", "description": "订单ID,例如 1001" } }, "required": ["order_id"] } } } ] response = client.chat.completions.create( model="your-model-name", messages=[ {"role": "system", "content": "你是电商客服助手。查询订单时请使用工具。"}, {"role": "user", "content": "帮我查一下订单 1002 到哪了?"} ], tools=tools, tool_choice="auto" ) message = response.choices[0].message print("模型返回:", message) # 如果模型决定调用工具 if message.tool_calls: tool_call = message.tool_calls[0] func_name = tool_call.function.name args = json.loads(tool_call.function.arguments) print(f"调用函数: {func_name}, 参数: {args}") result = query_order_status(args["order_id"]) # 把工具结果传回给模型 response2 = client.chat.completions.create( model="your-model-name", messages=[ {"role": "system", "content": "你是电商客服助手。"}, {"role": "user", "content": "帮我查一下订单 1002 到哪了?"}, message, {"role": "tool", "tool_call_id": tool_call.id, "content": result} ], tools=tools ) final_answer = response2.choices[0].message.content print("最终回答:", final_answer)

执行逻辑比较清晰,但新手容易在以下三点出错。

第一,tool_choice="auto"表示由模型自主决定要不要调用工具。如果你希望某个场景必须调用某一个工具,可以指定tool_choice={"type": "function", "function": {"name": "query_order_status"}}。

第二,第二轮请求必须把第一轮返回的完整message对象加入messages列表中,然后用role: "tool"按tool_call_id关联工具结果。漏掉这一步模型会不理解工具结果的来路。

第三,工具函数的description和参数的description要写得足够明确。模型是靠这些描述来决定何时调用、传什么参数的。描述写得越清晰,工具调用准确率越高。

5.3 Agent 的进阶方向

上面只是一个“一次工具调用”的最小示例。真实的 Agent 需要处理多轮工具调用、错误重试、任务规划、记忆管理等复杂场景。这些能力可以在多个 Agent 框架中找到现成实现,比如 LangChain 的 Agent 模块、LangGraph、AutoGen、MetaGPT 等。

我的建议是:先自己用 Function Calling 手工实现一次 Agent 循环,理解底层流转逻辑;再去使用框架。如果一上来就用框架,遇到问题你很难判断是模型能力问题、Prompt 问题还是框架配置问题。

6. MCP 协议实战

6.1 什么是 MCP

MCP 全称 Model Context Protocol,模型上下文协议。它由 Anthropic 在 2024 年底提出,目标是解决大模型工具接入的“碎片化”问题。

在没有 MCP 之前,每接入一个外部工具,开发者都要针对具体工具和大模型平台写一套定制逻辑。工具一变,代码就要改。MCP 提供了一套统一标准:工具提供方按照 MCP 协议实现一个 Server,模型客户端按照 MCP 协议连接 Server,两边就能自动完成工具发现、调用和结果返回。

你可以把 MCP 理解成“大模型世界的 USB 接口”。USB 让不同设备可以通过统一接口接入电脑;MCP 让不同工具可以通过统一协议接入大模型。

MCP 的核心角色有两个:

  • MCP Server:暴露工具、资源或提示词的服务端,可以本地运行,也可以远程部署。
  • MCP Client:连接 Server 的大模型应用端,负责发现工具并调用。

6.2 用 Python 实现一个简易 MCP Server

当前 Python 生态中比较常用的 MCP 开发库是mcp,你可以先安装它。由于 MCP 的 SDK 仍在演进中,API 可能会有变化,以下代码演示的是当前版本的核心写法。

pip install mcp
# 文件路径:05_mcp_server.py from mcp.server.fastmcp import FastMCP mcp = FastMCP("demo-server") @mcp.tool() def add(a: int, b: int) -> int: """计算两个整数之和""" return a + b @mcp.tool() def get_time() -> str: """返回当前服务器时间字符串""" from datetime import datetime return datetime.now().isoformat() if __name__ == "__main__": mcp.run()

运行以后,这个 Server 默认会通过标准输入输出与 Client 通信,这是本地集成最常用的方式。如果你的 MCP Server 需要远程访问,可以使用mcp.run(transport="streamable-http")或mcp.run(transport="sse")等方式。

MCP Server 本身不复杂,复杂的地方在于与具体的大模型应用框架集成。不同的框架对 MCP Client 的支持程度不同,配置方式也不同,需要查阅对应框架的文档。

6.3 MCP 的典型应用场景

现在 MCP 生态里已经出现了大量 Server,覆盖浏览器操作、数据库访问、设计工具、开发调试、文件系统等领域。比如:

  • Playwright MCP:让模型可以控制浏览器进行自动化测试和网页操作。
  • Chrome DevTools MCP:让模型读取浏览器调试数据。
  • Blender MCP:把 MCP 与 3D 建模软件连接起来。
  • Burp Suite MCP、Yakit MCP:在安全测试场景中辅助漏洞排查与请求分析。

这些工具的本质是一致的:通过标准协议把外部能力暴露给大模型,让模型能像调用函数一样使用它们。

需要特别说明的是,MCP 不解决安全认证问题。当你通过 MCP 暴露一个工具时,必须自己做好权限控制与身份校验,尤其是涉及文件系统、数据库和生产环境的操作时,要遵循最小权限原则。后续最佳实践部分还会强调这一点。

6.4 RAG + Agent + MCP 的组合关系

这三个技术不是各自孤立的。一个比较典型的生产级架构是:

  • 用 RAG 为 Agent 提供私有知识库检索能力。
  • 用 MCP Server 暴露外部业务工具,比如订单查询、CRM 接口、内部系统 API。
  • Agent 作为调度中枢,根据用户意图选择调用 RAG 检索还是 MCP 工具。

也就是说,RAG 是整个系统“知识来源”的一部分,MCP 是整个系统“工具来源”的标准化通道,Agent 负责在两者之上做决策。理解这层关系,你就不会再把它们当成三条不相干的学习路线了。

7. 常见问题与排查思路

7.1 高频问题表

问题现象常见原因解决思路
API 调用返回 401API Key 错误或未生效检查环境变量和平台控制台,确认 Key 状态
调用返回 404/模型不存在模型名称拼写错误或未开通权限对照平台文档确认模型名,检查模型服务开通状态
流式输出在前端不显示SSE 解析格式错误或被代理缓冲确认text/event-stream,检查反向代理是否关闭缓冲
Embedding 维度不匹配文本向量化时使用了不同模型统一 Embedding 模型和版本
检索结果不相关chunk_size 过大或文档分割不合理调整切分参数,检查文档内容质量
工具调用参数解析失败Function Calling 返回的 JSON 格式异常对参数做 JSON 解析容错,给工具函数增加日志
MCP Server 无法被 Client 发现传输方式不匹配或路径配置错误确认 Server 与 Client 使用相同的 transport 和地址
Agent 陷入死循环缺少最大步数限制或推理逻辑不闭环设置最大迭代次数,增加终止条件判断

7.2 排查思路

以最常见的“SSE 前端不显示”为例,排查顺序是:

  1. 先用curl直接访问流式接口,确认服务端输出是否正确。
  2. 检查浏览器开发者工具里的 Network 面板,查看响应头和响应体。
  3. 确认 Nginx 等反向代理是否设置了proxy_buffering off,因为默认缓冲会破坏 SSE 的实时性。
  4. 确认前端解析逻辑是否对数据进行正确的增量拼接。

以“Agent 工具调用失败”为例,排查顺序是:

  1. 查看模型返回的tool_calls内容和参数。
  2. 手动调用工具函数看是否本身有 Bug。
  3. 检查第二轮请求的messages结构是否符合 API 要求。
  4. 简化工具描述,减少模型的误解空间。

在这里我补一个值得注意的原则:不要为了排查问题反复修改 Prompt 或者盲目升级模型版本。先记录原始报错,复现最小场景,通过日志定位是哪一层出了问题,再针对性修改。

8. 最佳实践与工程建议

8.1 代码与配置管理

大模型应用开发中,API Key 是最高频的泄露点。绝对不要把 Key 硬编码在代码文件里,更不要传到公开仓库。建议通过环境变量加载,并在.gitignore中排除相关配置文件。

export LLM_API_KEY="your-key" export LLM_BASE_URL="https://your-provider.com/v1" export LLM_MODEL="your-model-name"

在 Python 中读取:

import os api_key = os.environ.get("LLM_API_KEY")

配置项尽量与代码分离。模型名称、温度参数、超时时间、检索的 top_k 值,都应该归入配置文件,方便不同环境切换。

8.2 提示词与工具描述

提示词的质量直接决定大模型应用的效果。写系统提示词时,尽量明确以下内容:

  • 角色的边界:你能做什么、不能做什么。
  • 回答的风格要求:简洁还是详细,是否限制字数。
  • 知识来源的限制:是否只能基于检索结果回答。
  • 不确定时的行为:是直接承认不知道,还是给出猜测。

工具函数的描述同样重要。一个好的工具描述应该包含:工具用途、何时调用、参数含义、返回值结构。描述中的“何时调用”尤其关键,因为模型是根据描述做决策的。

8.3 安全与权限边界

RAG 知识库中可能包含敏感数据。如果要在多人系统里使用 RAG,必须做权限控制,保证不同角色只能检索到各自授权范围内的文档。这在工程上称为“权限感知检索”。

MCP Server 暴露工具时要坚持最小权限原则。不要为了让模型方便,就把一个能操作生产数据库的接口直接暴露给模型。合理的做法是:

  • 对 MCP 工具调用进行身份鉴权。
  • 对高风险操作增加二次确认。
  • 记录全部调用日志。
  • 对工具的读取和写入权限做拆分。

如果 Agent 能够执行写操作,比如发送邮件、修改数据、删除文件,务必要设置确认环节。在自动化程度提高的同时,出错代价也在提高,安全兜底必不可少。

8.4 性能与成本优化

大模型 API 按 Token 计费。RAG 的上下文拼接、Agent 的多轮工具调用,都会显著增加 Token 消耗。实际项目中可以从几个方向优化:

  • 限制检索片段数量,降低 Prompt 长度。
  • 使用小型模型处理简单任务,关键任务才路由到大模型。
  • 对相同或相似的问题做缓存,避免重复调用。
  • 对 Embedding 结果做持久化,避免每次检索都重新计算。
  • 设置请求超时和重试策略,避免不必要的等待和浪费。

8.5 可观测性

大模型应用是典型的“黑盒”系统,问题往往出在无法复现的中间环节。因此从第一天起就要完善日志,至少记录:

  • 每次请求的输入和输出。
  • 检索命中的片段及得分。
  • 模型调用的工具和参数。
  • 每次 AP 调用耗时和 Token 消耗。
  • 错误信息与堆栈。

有了这些日志,复盘问题时就能还原当时的完整决策链,而不是靠猜。

9. 学习路线与后续建议

9.1 从本文出发的进阶路径

如果你已经跟着本文代码跑通了所有示例,下一步建议按照这个顺序继续深入:

  1. 选一个国产或海外大模型平台,完整看一遍官方文档,把对话、流式、Embedding、Function Calling 四个核心接口都手动调一遍。
  2. 深入学习 LangGraph 或其他 Agent 编排框架,理解多智能体协作与状态管理。
  3. 把本地的 NumPy 向量检索替换为真正的向量数据库,比如 Chroma 或 Milvus,处理几万条文档数据。
  4. 研究 RAG 的进阶优化:混合检索、重排、查询改写、多路召回。
  5. 自己实现一个 MCP Server,接入一个真实业务系统,感受工具标准化的价值。

9.2 动手是最好的学习方式

技术学习本质上是一个“建立反馈回路”的过程。看一遍文章、收藏十个仓库,都不如把环境搭起来跑通一个最小 demo 有价值。你在实战中踩过的每一个依赖冲突、参数报错、解析异常,都会变成你对这个技术最直观的理解。

如果这篇文章对你有帮助,建议先收藏,然后立刻动手把环境搭起来,把 01 到 05 这几个脚本逐个跑通。遇到问题欢迎在评论区留言,我会根据高频反馈持续补充排错内容。写代码这件事,最怕的就是只看不动手。

返回列表