六边云
65.72M · 2026-02-13
本文深度剖析AI领域五大热门技术名词(Skill、MCP、RAG、Agent、OpenClaw)的真实底层逻辑,用通俗易懂的大白话揭穿技术包装,带你理解模型上下文协议、检索增强生成、智能代理的工作原理与实际应用场景。适合AI初学者、开发者、产品经理深入了解AI技术栈,避免被新概念迷惑。全文配有12张架构图,5个实战案例,彻底搞懂AI Agent新范式。
MCP协议、RAG检索增强生成、AI Agent、OpenClaw、Clawdbot、模型上下文协议、智能代理、Skill技能、大模型应用、AI技术栈、向量检索、工具调用、自动化编程
在AI领域,每隔几个月就会冒出一堆新概念:Skill、MCP、RAG、Agent、OpenClaw...这些术语听起来高大上,但很多人用它们只是为了"包装"和"炒作"。
真相是:这些概念背后的技术原理并不复杂,只是被赋予了新的名字和营销话术。本文将用大白话拆穿它们,让你彻底看懂AI技术的底层逻辑。
graph TB
User[用户] <--> OpenClaw[OpenClaw平台]
OpenClaw <--> LLM[大模型]
MCP[MCP<br/>协议层] --> Skill[Skill工具集<br/>统一调用接口]
RAG[RAG<br/>检索引擎] --> Skill
Agent[Agent<br/>智能体] --> Skill
OpenClaw --> MCP
OpenClaw --> RAG
OpenClaw --> Agent
style MCP fill:#e1f5ff
style RAG fill:#fff3e0
style Agent fill:#f3e5f5
style Skill fill:#e8f5e9
style OpenClaw fill:#fce4ec
图1: AI技术栈全景 - MCP统一工具调用,RAG提供知识,Agent负责决策
官方说法: Skill是AI模型的具体能力模块,可以通过训练或插件扩展。
大白话: 就是给AI贴的能力标签,比如"会写代码"、"会画图"、"会翻译"。
# 所谓的"天气查询Skill"本质上就是:
def weather_skill(city):
api_url = f"https://api.weather.com/v1/{city}"
response = requests.get(api_url)
return response.json()
# 这和传统的API调用有什么区别?没有!
结论: Skill不是新技术,只是功能模块化的新名字。重要的是如何设计和组织这些能力,而不是名字本身。
官方定义: Model Context Protocol(模型上下文协议)是Anthropic在2024年11月发布的开放标准,用于连接AI模型与外部系统。
大白话: 一套标准化的接口规范,让AI能统一地调用外部工具(数据库、文件、邮件等),不用为每个工具写不同的代码。
graph TD
Client[AI应用/模型<br/>Client]
Client -->|MCP标准协议<br/>JSON-RPC 2.0| Protocol[MCP协议层<br/>统一接口]
Protocol --> Server1[MCP Server<br/>文件系统]
Protocol --> Server2[MCP Server<br/>数据库]
Protocol --> Server3[MCP Server<br/>邮件服务]
Protocol --> Server4[MCP Server<br/>搜索引擎]
Protocol --> Server5[MCP Server<br/>...]
style Client fill:#e3f2fd
style Protocol fill:#fff9c4
style Server1 fill:#f1f8e9
style Server2 fill:#f1f8e9
style Server3 fill:#f1f8e9
style Server4 fill:#f1f8e9
style Server5 fill:#f1f8e9
图2: MCP协议架构 - 一个协议,连接所有工具
在MCP出现之前,每个AI应用都要单独实现与外部工具的对接:
混乱的现状:
AI助手A → 自定义协议1 → 文件系统
AI助手B → 自定义协议2 → 数据库
AI助手C → 自定义协议3 → 邮箱
MCP统一后:
AI助手A/B/C → MCP标准协议 → 文件/数据库/邮箱
# MCP服务配置示例
mcp_servers:
- name: "filesystem"
protocol: "mcp/v1"
endpoint: "http://localhost:8080/mcp"
tools:
- read_file
- write_file
- list_directory
- name: "database"
protocol: "mcp/v1"
endpoint: "http://localhost:8081/mcp"
tools:
- sql_query
- sql_execute
# AI调用MCP工具的流程
from mcp_client import MCPClient
client = MCPClient(server_url="http://localhost:8080/mcp")
# 统一接口调用不同工具
file_content = client.call_tool("read_file", {"path": "/data/report.txt"})
query_result = client.call_tool("sql_query", {"sql": "SELECT * FROM users"})
优势:
局限:
官方定义: Retrieval-Augmented Generation(检索增强生成),在生成答案前先从外部知识库检索相关信息。
大白话: AI在回答问题前先Google一下,然后根据搜到的资料生成答案,而不是靠死记硬背的训练数据。
graph TD
subgraph Offline[离线阶段 - 数据准备]
Doc[1. 提取文档资料] --> Chunk[2. 分块 Chunk]
Chunk --> Embed[3. 向量化 Embed]
Embed --> Store[4. 入库 向量库]
end
subgraph Online[在线阶段 - 检索生成]
Query[用户问题] --> QueryEmbed[问题向量化]
QueryEmbed --> Search[向量相似度检索<br/>Top-K最相关文档]
Store -.存储查询.-> Search
Search --> Prompt[构造增强Prompt<br/>问题+检索到的文档]
Prompt --> LLM[LLM生成答案]
LLM --> Answer[最终答案]
end
style Offline fill:#e8f5e9
style Online fill:#e1f5fe
style Store fill:#fff9c4
style Answer fill:#f3e5f5
图3: RAG工作流程 - 先查资料,再生成答案
传统大模型的三大痛点:
RAG的解决方案: 不改变模型本身,而是动态补充外部知识。
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import OpenAIEmbeddings
from langchain.vectorstores import Chroma
# 1. 数据提取
documents = load_documents("./knowledge_base/") # 加载PDF、Word、TXT等
# 2. 数据分块(Chunking)
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=500, # 每块500字
chunk_overlap=50 # 块之间重叠50字,避免语义断裂
)
chunks = text_splitter.split_documents(documents)
# 3. 向量化(Embedding)
embeddings = OpenAIEmbeddings()
vector_chunks = embeddings.embed_documents([c.page_content for c in chunks])
# 4. 入库存储(Vector Database)
vectorstore = Chroma.from_documents(
documents=chunks,
embedding=embeddings,
persist_directory="./chroma_db"
)
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
# 1. 用户提问
user_question = "如何使用Python实现RAG系统?"
# 2. 问题向量化
question_vector = embeddings.embed_query(user_question)
# 3. 向量相似度检索(Top-K)
relevant_docs = vectorstore.similarity_search(user_question, k=3)
# 4. 构造增强Prompt
context = "nn".join([doc.page_content for doc in relevant_docs])
prompt = f"""
基于以下参考资料回答问题:
参考资料:
{context}
问题: {user_question}
答案:
"""
# 5. LLM生成答案
llm = OpenAI(temperature=0)
answer = llm(prompt)
print(answer)
| 技术环节 | 关键技术 | 作用 |
|---|---|---|
| 数据分块 | RecursiveCharacterTextSplitter | 避免文本过长超过模型上下文限制 |
| 向量化 | OpenAI Embeddings / BGE / M3E | 将文本转为高维向量,计算语义相似度 |
| 向量存储 | Chroma / Pinecone / Weaviate | 支持高效的相似度检索(ANN算法) |
| 重排序 | Cross-Encoder / Cohere Rerank | 对检索结果二次排序,提高准确性 |
graph LR
subgraph Documents[文档向量化]
DocA["文档A: Python编程<br/>[0.8, 0.2, 0.5, ...]"]
DocB["文档B: 机器学习<br/>[0.7, 0.6, 0.3, ...]"]
DocC["文档C: 做饭技巧<br/>[0.1, 0.9, 0.7, ...]"]
end
Query["用户问题:<br/>如何用Python做AI?<br/>[0.75, 0.3, 0.4]"]
subgraph Similarity[余弦相似度计算]
SimA["文档A<br/>相似度: 0.92 <br/>最相关"]
SimB["文档B<br/>相似度: 0.85 "]
SimC["文档C<br/>相似度: 0.23 <br/>不相关"]
end
Result["返回Top-2:<br/>[文档A, 文档B]"]
Query --> Similarity
Documents --> Similarity
Similarity --> Result
style DocA fill:#e8f5e9
style DocB fill:#fff3e0
style DocC fill:#ffebee
style SimA fill:#c8e6c9
style SimB fill:#ffe0b2
style SimC fill:#ffcdd2
style Result fill:#e1f5fe
图4: 向量检索 - 通过语义相似度找到最相关的文档
最基础版本,直接检索 → 生成。
问题:
增加了:
# 查询改写示例
def query_rewrite(original_query):
"""将用户问题改写为更适合检索的形式"""
rewrite_prompt = f"将以下问题改写为更精确的搜索查询:n{original_query}"
rewritten = llm(rewrite_prompt)
return rewritten
# 混合检索
def hybrid_search(query, vectorstore, k=5):
"""结合向量检索和关键词检索"""
vector_results = vectorstore.similarity_search(query, k=k)
keyword_results = bm25_search(query, k=k) # BM25算法
# 融合排序(RRF算法)
merged = reciprocal_rank_fusion(vector_results, keyword_results)
return merged[:k]
最强形式,RAG + Agent结合:
# Agentic RAG示例
class AgenticRAG:
def answer(self, question):
# 1. 规划:分析问题需要哪些知识
plan = self.planner.plan(question)
# 2. 迭代检索
context = []
for step in plan:
docs = self.retrieve(step.query)
context.extend(docs)
# 判断是否需要继续检索
if self.is_sufficient(context, question):
break
# 3. 生成答案
answer = self.generate(question, context)
return answer
| 维度 | RAG | Fine-tuning |
|---|---|---|
| 知识更新 | 实时更新,只需替换知识库 | 需要重新训练,成本高 |
| 可解释性 | 可追溯引用来源 | 黑盒模型 |
| 成本 | 低成本,无需训练 | 需要GPU、数据标注 |
| 准确度 | ️ 依赖检索质量 | 深度掌握知识 |
| 适用场景 | 事实性问答、文档理解 | 风格模仿、任务专用 |
官方定义: AI Agent是具备感知、决策、执行能力的自主系统,能通过调用工具完成复杂任务。
大白话: 一个能自己做计划、用工具、修正错误的AI助手,不只是回答问题,而是真正完成任务。
graph TD
Start([开始]) --> P1[1. 感知 Perception<br/>理解用户意图和环境状态]
P1 --> P2[2. 规划 Planning<br/>分解任务,制定执行步骤<br/>使用思维链CoT]
P2 --> P3[3. 执行 Execution<br/>调用工具API,完成操作<br/>搜索/文件/邮件/代码...]
P3 --> P4[4. 反思 Reflection<br/>评估结果,决定是否重试]
P4 --> Decision{任务完成?}
Decision -->|是| End([结束])
Decision -->|否| P2
style P1 fill:#e3f2fd
style P2 fill:#fff3e0
style P3 fill:#f3e5f5
style P4 fill:#e8f5e9
style Decision fill:#fff9c4
图5: Agent工作循环 - 感知→规划→执行→反思,持续迭代直到完成
| 对比项 | 传统Chatbot | AI Agent |
|---|---|---|
| 能力 | 只能聊天,回答问题 | 能执行任务,调用工具 |
| 自主性 | 被动响应 | 主动规划和执行 |
| 记忆 | 单轮或短期上下文 | 长期记忆(数据库/文件) |
| 工具使用 | 无 | 浏览器、文件系统、API等 |
| 示例 | 客服机器人 | 自动化办公助手 |
理解用户意图和当前环境状态。
# 意图识别示例
def perceive(user_input):
"""分析用户输入,识别任务类型"""
prompt = f"""
分析以下用户请求,识别任务类型:
用户: {user_input}
任务类型(选择一个):
- search: 需要搜索信息
- file_operation: 文件操作
- calculation: 计算任务
- email: 发送邮件
"""
task_type = llm(prompt)
return task_type
将复杂任务分解为可执行的步骤。
# ReAct(Reasoning + Acting)框架
def plan(task):
"""使用思维链(CoT)规划任务步骤"""
prompt = f"""
任务: {task}
请一步步分析如何完成这个任务:
思考(Thought): [分析当前状态]
行动(Action): [选择工具和参数]
观察(Observation): [预期结果]
重复上述过程直到任务完成。
"""
plan = llm(prompt)
return parse_plan(plan)
调用工具API完成具体操作。
# 工具调用框架
class ToolExecutor:
def __init__(self):
self.tools = {
"search": self.search_web,
"read_file": self.read_file,
"write_file": self.write_file,
"send_email": self.send_email,
}
def execute(self, action, params):
"""执行工具调用"""
tool_func = self.tools.get(action)
if tool_func:
result = tool_func(**params)
return result
else:
return {"error": "Unknown tool"}
def search_web(self, query):
"""搜索网络"""
return google_search(query)
def read_file(self, path):
"""读取文件"""
with open(path, 'r') as f:
return f.read()
评估执行结果,修正错误。
def reflect(task, plan, results):
"""反思执行结果,决定是否需要重试"""
prompt = f"""
原始任务: {task}
执行计划: {plan}
执行结果: {results}
评估:
1. 任务是否完成?
2. 结果是否正确?
3. 需要修正吗?
"""
reflection = llm(prompt)
if "需要重试" in reflection:
new_plan = adjust_plan(plan, reflection)
return new_plan
else:
return "完成"
ReAct = Reasoning(推理) + Acting(行动)
循环流程:
1. Thought: 我应该先搜索相关资料
2. Action: search("Python RAG实现")
3. Observation: 找到5篇文章...
4. Thought: 信息足够了,现在可以写代码
5. Action: write_file("rag.py", code)
6. Observation: 文件创建成功
7. Thought: 任务完成!
完整代码示例:
class ReActAgent:
def __init__(self, llm, tools):
self.llm = llm
self.tools = tools
self.max_iterations = 10
def run(self, task):
"""执行任务"""
history = []
for i in range(self.max_iterations):
# 1. 思考(Reasoning)
prompt = self._build_prompt(task, history)
response = self.llm(prompt)
# 2. 解析响应
thought, action, action_input = self._parse_response(response)
# 3. 执行行动(Acting)
if action == "FINISH":
return action_input # 任务完成
observation = self.tools[action](action_input)
# 4. 记录历史
history.append({
"thought": thought,
"action": action,
"action_input": action_input,
"observation": observation
})
return "达到最大迭代次数,任务未完成"
def _build_prompt(self, task, history):
"""构造ReAct Prompt"""
prompt = f"任务: {task}nn"
for step in history:
prompt += f"Thought: {step['thought']}n"
prompt += f"Action: {step['action']}({step['action_input']})n"
prompt += f"Observation: {step['observation']}nn"
prompt += "Thought: " # 触发下一轮思考
return prompt
单个Agent有局限,多Agent协作更强大:
# 多Agent系统示例
class MultiAgentSystem:
def __init__(self):
self.agents = {
"researcher": ResearchAgent(), # 负责搜索和分析
"coder": CodingAgent(), # 负责写代码
"reviewer": ReviewAgent(), # 负责审查代码
"tester": TestingAgent(), # 负责测试
}
def solve(self, task):
"""多Agent协作解决任务"""
# 1. 研究员收集资料
research_result = self.agents["researcher"].run(task)
# 2. 程序员写代码
code = self.agents["coder"].run(research_result)
# 3. 审查员检查代码
review = self.agents["reviewer"].run(code)
if review["status"] == "approved":
# 4. 测试员运行测试
test_result = self.agents["tester"].run(code)
return test_result
else:
# 返回修改意见给程序员
return self.agents["coder"].revise(code, review["feedback"])
graph TD
User[用户需求:<br/>开发一个功能] --> Coordinator[协调器<br/>Coordinator]
Coordinator --> Researcher[研究员Agent]
Coordinator --> Coder[程序员Agent]
Coordinator --> Tester[测试员Agent]
Researcher -->|搜集资料| ResultA[结果A]
Coder -->|编写代码| ResultB[结果B]
Tester -->|运行测试| ResultC[结果C]
ResultA --> Reviewer
ResultB --> Reviewer
ResultC --> Reviewer[审查员Agent<br/>Code Review]
Reviewer --> Decision{审查结果}
Decision -->|通过| Done[完成!]
Decision -->|不通过| Feedback[反馈给程序员]
Feedback --> Coder
style User fill:#e3f2fd
style Coordinator fill:#fff9c4
style Researcher fill:#e8f5e9
style Coder fill:#f3e5f5
style Tester fill:#ffe0b2
style Reviewer fill:#ffccbc
style Done fill:#c8e6c9
图6: Multi-Agent系统 - 专业分工,协同完成复杂任务
OpenClaw(原名Clawdbot/Moltbot)是一个开源的本地优先AI代理平台,将上述所有技术整合在一起:
graph TB
subgraph UI[用户交互层]
Web[Web界面]
WhatsApp[WhatsApp]
T@elegrimm[T@elegrimm]
WeChat[微信]
end
subgraph Core[OpenClaw核心引擎]
Agent[Agent<br/>规划系统]
RAG[RAG<br/>检索引擎]
Memory[长期记忆<br/>Markdown]
end
subgraph MCP[MCP协议层 - 统一工具调用]
Protocol[MCP标准协议<br/>JSON-RPC 2.0]
end
subgraph Skills[工具集 Skills]
File[文件系统]
DB[数据库]
Email[邮件]
Browser[浏览器]
GitHub[GitHub]
More[...]
end
subgraph LLMs[大模型支持]
GPT[GPT-4]
Claude[Claude]
Qwen[Qwen]
DeepSeek[DeepSeek]
MoreLLM[...]
end
UI --> Core
Core --> Protocol
Protocol --> Skills
Core --> LLMs
style Core fill:#fff9c4
style MCP fill:#e1f5fe
style Skills fill:#e8f5e9
style LLMs fill:#f3e5f5
图7: OpenClaw架构 - 集MCP、RAG、Agent于一体的完整AI助手
GitHub 10万+ Star的秘密:
你: "整理桌面上的文件,把PDF放到Documents,图片放到Pictures"
OpenClaw执行:
1. 扫描桌面文件列表
2. 识别文件类型
3. 批量移动文件
4. 创建整理报告
你: "查看今天的邮件,标记重要的,草拟回复"
OpenClaw执行:
1. 连接Gmail API
2. 筛选今天的邮件
3. 用AI分析重要性
4. 生成回复草稿
5. 发送通知
你: "在GitHub上找一个Python爬虫项目,下载并运行示例"
OpenClaw执行:
1. 搜索GitHub API
2. 克隆仓库
3. 安装依赖
4. 运行示例
5. 报告结果
# 1. 安装Node.js(需要v18+)
# 2. 全局安装OpenClaw
npm i -g clawdbot
# 3. 初始化配置
clawdbot onboard
# 按提示配置:
# - 选择AI模型(Claude/GPT/Qwen等)
# - 输入API Key
# - 设置工作目录
# 4. 启动服务
clawdbot start
# 5. 访问Web界面
#
# 使用阿里云轻量服务器一键部署
# 1. 购买服务器(选择OpenClaw镜像)
# 2. 获取百炼API-Key
# 3. 在控制台粘贴API-Key
# 4. 生成访问Token
# 5. 通过浏览器访问
# 优势:
# 7×24小时在线
# 无需本地配置
# 远程访问(WhatsApp/T@elegrimm/微信)
{
"models": {
"default": "claude-3.5-sonnet",
"providers": {
"anthropic": {
"apiKey": "sk-ant-xxx",
"models": ["claude-3.5-sonnet", "claude-3-opus"]
},
"openai": {
"apiKey": "sk-xxx",
"baseURL": "https://api.openai.com/v1",
"models": ["gpt-4", "gpt-3.5-turbo"]
},
"aliyun": {
"apiKey": "sk-xxx",
"baseURL": "https://dashscope.aliyuncs.com/compatible-mode/v1",
"models": ["qwen-max", "qwen-turbo"]
}
}
},
"tools": {
"mcp_servers": [
{
"name": "filesystem",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-filesystem", "/Users/you"]
},
{
"name": "brave-search",
"command": "npx",
"args": ["-y", "@modelcontextprotocol/server-brave-search"],
"env": {
"BRAVE_API_KEY": "your-api-key"
}
}
]
},
"memory": {
"type": "markdown",
"path": "~/.clawdbot/memory"
},
"integrations": {
"whatsapp": {
"enabled": true,
"qr_code_path": "~/.clawdbot/whatsapp_qr.png"
},
"telegram": {
"enabled": true,
"bot_token": "your-telegram-bot-token"
}
}
}
内存结构:
~/.clawdbot/memory/
├── user_preferences.md # 用户偏好
├── task_history/ # 任务历史
│ ├── 2026-02-01.md
│ └── 2026-02-12.md
├── knowledge/ # 知识库
│ ├── work_projects.md
│ └── personal_notes.md
└── context/ # 上下文缓存
└── current_session.md
每次对话都会更新记忆:
# user_preferences.md
## 工作习惯
- 偏好使用Python
- 喜欢详细的代码注释
- 每天9:00查看邮件
## 常用联系人
- 张三(同事): zhangsan@example.com
- 李四(客户): lisi@client.com
## 重要日期
- 项目截止: 2026-02-28
- 团队会议: 每周三14:00
// 自然语言定时任务
"每天早上8点检查GitHub热榜,如果有AI相关的高星项目就通知我"
// OpenClaw内部解析为:
{
"schedule": "0 8 * * *", // Cron表达式
"task": {
"action": "github_trending",
"filter": { "topic": "AI", "min_stars": 100 },
"notification": {
"channel": "telegram",
"priority": "high"
}
}
}
OpenClaw通过MCP协议无缝对接:
| 应用类型 | 支持工具 | 能力 |
|---|---|---|
| 聊天 | WhatsApp, T@elegrimm, Discord, 微信, 飞书 | 远程控制、接收通知 |
| 文件 | 本地文件系统, Google Drive, Dropbox | 读写、搜索、整理 |
| 邮件 | Gmail, Outlook | 收发、筛选、自动回复 |
| 代码 | GitHub, GitLab, VS Code | 克隆、提交、代码审查 |
| 浏览器 | Puppeteer, Playwright | 自动化操作、截图 |
| 数据库 | PostgreSQL, MongoDB | 查询、更新 |
用户
↓
┌────────────────┐
│ OpenClaw │ ← 完整的AI助手产品
│ (集成平台) │
└────────────────┘
↓
┌────────────┼────────────┐
↓ ↓ ↓
┌──────┐ ┌──────┐ ┌────────┐
│ Agent│ │ RAG │ │ MCP │
│ 系统 │ │ 引擎 │ │ 协议层 │
└──────┘ └──────┘ └────────┘
↓ ↓ ↓
规划执行 知识检索 工具调用
↓ ↓ ↓
┌──────────────────────────────┐
│ Skill工具集 │
│ (文件/邮件/浏览器/数据库) │
└──────────────────────────────┘
示例任务: "帮我写一篇关于RAG的技术博客"
1. Agent规划:
Thought: 这是一个写作任务,需要先搜集资料
Action: 使用RAG检索相关知识
2. RAG检索:
Query: "RAG技术原理和应用"
↓ (通过MCP调用搜索工具)
Retrieve: 找到5篇相关文章
↓
Context: 整理成结构化知识
3. MCP工具调用:
Tool: brave_search("RAG最新论文")
Tool: read_file("/docs/rag_notes.md")
↓
Result: 最新技术动态
4. Agent执行:
Action: 生成博客大纲
Action: 逐段撰写内容
Action: write_file("rag_blog.md", content)
↓
Complete: 博客写作完成!
5. Skill支撑:
- 搜索Skill: Brave Search API
- 文件Skill: 文件系统读写
- 写作Skill: Markdown生成
| 技术 | 核心作用 | 比喻 | 是否必需 |
|---|---|---|---|
| MCP | 标准化工具调用接口 | 操作系统的API | ️ 可选(提高互操作性) |
| RAG | 实时知识增强 | 图书馆 | ️ 可选(解决知识过时) |
| Agent | 自主规划和执行 | 秘书助手 | 必需(智能化的核心) |
| Skill | 具体能力模块 | 工具箱 | 必需(实际功能) |
| OpenClaw | 完整产品平台 | 智能办公室 | ️ 可选(集成方案) |
需求: 让AI能回答公司内部文档的问题。
# 完整的RAG系统搭建
import os
from langchain.document_loaders import DirectoryLoader, PDFLoader
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.embeddings import HuggingFaceEmbeddings
from langchain.vectorstores import FAISS
from langchain.chains import RetrievalQA
from langchain.llms import OpenAI
# 1. 加载企业文档
loader = DirectoryLoader(
"/company_docs/",
glob="**/*.pdf",
loader_cls=PDFLoader
)
documents = loader.load()
# 2. 文本分块
text_splitter = RecursiveCharacterTextSplitter(
chunk_size=1000,
chunk_overlap=200
)
texts = text_splitter.split_documents(documents)
# 3. 使用开源嵌入模型(节省成本)
embeddings = HuggingFaceEmbeddings(
model_name="BAAI/bge-large-zh-v1.5" # 中文效果好
)
# 4. 构建向量库
vectorstore = FAISS.from_documents(texts, embeddings)
vectorstore.save_local("company_kb")
# 5. 创建问答链
qa_chain = RetrievalQA.from_chain_type(
llm=OpenAI(temperature=0),
chain_type="stuff",
retriever=vectorstore.as_retriever(search_kwargs={"k": 3}),
return_source_documents=True
)
# 6. 测试问答
query = "公司的年假正策是什么?"
result = qa_chain({"query": query})
print(f"答案: {result['result']}")
print(f"n引用来源:")
for doc in result['source_documents']:
print(f"- {doc.metadata['source']}, 页码{doc.metadata.get('page', '未知')}")
输出示例:
答案: 根据公司员工手册第23页,正式员工享有以下年假:
- 工作满1年: 5天
- 工作满3年: 10天
- 工作满5年: 15天
年假必须提前2周申请,特殊情况可临时调整。
引用来源:
- /company_docs/员工手册2026.pdf, 页码23
- /company_docs/HR制度汇编.pdf, 页码45
需求: 输入数据文件,自动生成分析报告。
# ReAct Agent实现
class DataAnalysisAgent:
def __init__(self):
self.tools = {
"load_csv": self.load_csv,
"describe": self.describe_data,
"plot": self.plot_chart,
"generate_report": self.generate_report
}
def analyze(self, file_path):
"""主流程"""
task = f"分析{file_path}中的数据并生成报告"
# ReAct循环
for step in range(10):
# 思考
thought = self.llm_think(task, self.history)
if "完成" in thought:
break
# 行动
action, params = self.parse_action(thought)
result = self.tools[action](**params)
# 观察
self.history.append({
"thought": thought,
"action": action,
"result": result
})
return self.history[-1]["result"]
def llm_think(self, task, history):
"""LLM推理"""
prompt = f"""
任务: {task}
已执行步骤:
{self.format_history(history)}
下一步应该做什么?请按以下格式回答:
思考: [分析当前状态]
行动: [工具名称]
参数: [JSON格式]
可用工具:
- load_csv(path): 加载CSV文件
- describe(df): 统计描述
- plot(df, x, y, type): 绘图
- generate_report(insights): 生成报告
"""
response = call_llm(prompt)
return response
def load_csv(self, path):
import pandas as pd
df = pd.read_csv(path)
return {"data": df, "shape": df.shape}
def describe_data(self, df):
stats = df.describe()
return stats.to_dict()
def plot_chart(self, df, x, y, chart_type):
import matplotlib.pyplot as plt
if chart_type == "bar":
df.plot.bar(x=x, y=y)
elif chart_type == "line":
df.plot.line(x=x, y=y)
plt.savefig("analysis_chart.png")
return "图表已保存: analysis_chart.png"
def generate_report(self, insights):
"""生成Markdown报告"""
report = f"""
# 数据分析报告
## 数据概览
{insights['overview']}
## 关键发现
{insights['findings']}
## 可视化图表
## 建议
{insights['recommendations']}
"""
with open("report.md", "w") as f:
f.write(report)
return "报告已生成: report.md"
# 使用示例
agent = DataAnalysisAgent()
result = agent.analyze("sales_data.csv")
执行过程:
步骤1:
Thought: 需要先加载数据
Action: load_csv("sales_data.csv")
Observation: 加载成功,数据形状(1000, 5)
步骤2:
Thought: 了解数据的基本统计信息
Action: describe(df)
Observation: 平均销售额¥5000,最高¥20000
步骤3:
Thought: 绘制销售趋势图
Action: plot(df, x="date", y="sales", type="line")
Observation: 图表已保存
步骤4:
Thought: 生成报告
Action: generate_report({...})
Observation: 报告已生成!
需求: 让AI能调用Notion API管理知识库。
# MCP Server端(Notion工具)
from mcp import MCPServer
app = MCPServer()
@app.tool("notion_create_page")
def create_page(title: str, content: str, parent_id: str):
"""在Notion中创建页面"""
import requests
url = "https://api.notion.com/v1/pages"
headers = {
"Authorization": f"Bearer {NOTION_TOKEN}",
"Content-Type": "application/json",
"Notion-Version": "2022-06-28"
}
data = {
"parent": {"page_id": parent_id},
"properties": {
"title": {
"title": [{"text": {"content": title}}]
}
},
"children": [
{
"object": "block",
"type": "paragraph",
"paragraph": {
"rich_text": [{"text": {"content": content}}]
}
}
]
}
response = requests.post(url, json=data, headers=headers)
return response.json()
@app.tool("notion_search")
def search_pages(query: str):
"""搜索Notion页面"""
# 实现搜索逻辑...
pass
# 启动MCP服务器
if __name__ == "__main__":
app.run(host="0.0.0.0", port=8080)
# Client端(AI Agent)
from mcp_client import MCPClient
client = MCPClient("http://localhost:8080")
# AI通过MCP调用Notion工具
def ai_assistant(user_request):
if "记录到Notion" in user_request:
# 1. 提取信息
title, content = extract_info(user_request)
# 2. 调用MCP工具
result = client.call_tool("notion_create_page", {
"title": title,
"content": content,
"parent_id": "workspace_root"
})
return f"已创建页面: {result['url']}"
# 使用
response = ai_assistant("把今天的会议纪要记录到Notion")
# 输出: "已创建页面: x"
需求: 每天早上自动整理邮件并生成简报。
// OpenClaw配置文件
{
"workflows": [
{
"name": "daily_email_digest",
"schedule": "0 8 * * *", // 每天8点
"steps": [
{
"action": "gmail.list_messages",
"params": {
"query": "is:unread after:yesterday",
"maxResults": 50
},
"output": "unread_emails"
},
{
"action": "ai.analyze",
"params": {
"input": "{{ unread_emails }}",
"prompt": "分析这些邮件,按重要性分类:n1. 紧急待办n2. 需要回复n3. 仅供参考"
},
"output": "categorized"
},
{
"action": "markdown.generate",
"params": {
"template": "email_digest.template",
"data": "{{ categorized }}"
},
"output": "digest_content"
},
{
"action": "telegram.send_message",
"params": {
"ch@t_id": "@me",
"text": "{{ digest_content }}",
"parse_mode": "Markdown"
}
}
]
}
]
}
生成的每日简报:
# 邮件日报 - 2026年2月12日
## 紧急待办(3封)
1. **客户投诉** - support@client.com
- 优先级: 高
- 需在今天12:00前回复
2. **合同审批** - legal@company.com
- 需要签字确认
3. **服务器故障** - ops@company.com
- 生产环境异常,需立即处理
## 需要回复(5封)
- 项目进度询问 - pm@partner.com
- 技术方案讨论 - dev@team.com
- ...
## 仅供参考(12封)
- 行业动态周报
- 内部公告
- ...
需求: 智能客服机器人,能查询知识库并执行操作。
# Agentic RAG完整实现
class SmartCustomerService:
def __init__(self):
# RAG组件
self.vectorstore = self.init_knowledge_base()
# Agent工具
self.tools = {
"search_kb": self.search_knowledge,
"query_order": self.query_order_status,
"create_ticket": self.create_support_ticket,
"transfer_human": self.transfer_to_human
}
def handle_request(self, user_query):
"""处理用户请求"""
# 1. 意图识别
intent = self.classify_intent(user_query)
if intent == "knowledge_query":
# 简单问答 → 直接RAG
return self.simple_rag(user_query)
elif intent == "complex_task":
# 复杂任务 → Agentic RAG
return self.agentic_rag(user_query)
else:
return "抱歉,我无法理解您的问题"
def simple_rag(self, query):
"""简单RAG: 直接检索+生成"""
# 检索相关文档
docs = self.vectorstore.similarity_search(query, k=3)
# 生成答案
context = "n".join([d.page_content for d in docs])
prompt = f"根据以下资料回答问题:n{context}nn问题:{query}"
answer = call_llm(prompt)
return answer
def agentic_rag(self, task):
"""智能代理RAG: 规划+迭代检索+执行"""
plan = []
context = []
# 循环规划和执行
for step in range(5):
# 1. 规划下一步
thought = self.plan_next_step(task, plan, context)
if "完成" in thought:
break
# 2. 解析行动
action, params = self.parse_thought(thought)
# 3. 执行工具
if action == "search_kb":
# RAG检索
docs = self.vectorstore.similarity_search(params["query"])
result = [d.page_content for d in docs]
else:
# 其他工具调用
result = self.tools[action](**params)
# 4. 更新上下文
plan.append(thought)
context.append(result)
# 生成最终答案
final_answer = self.synthesize_answer(task, context)
return final_answer
def plan_next_step(self, task, history, context):
"""规划下一步行动"""
prompt = f"""
用户任务: {task}
已执行步骤:
{self.format_history(history)}
已获取信息:
{self.format_context(context)}
分析:
1. 任务是否完成?
2. 还需要什么信息?
3. 下一步应该调用哪个工具?
可用工具:
- search_kb(query): 搜索知识库
- query_order(order_id): 查询订单状态
- create_ticket(issue): 创建工单
- transfer_human(): 转人工客服
回答格式:
思考: [分析]
行动: [工具名称]
参数: {"key": "value"}
"""
response = call_llm(prompt)
return response
# 使用示例
service = SmartCustomerService()
# 场景1: 简单问答
user1 = "你们的退货正策是什么?"
answer1 = service.handle_request(user1)
# → 直接RAG检索知识库
# 场景2: 复杂任务
user2 = "我的订单ORD123456还没发货,能帮我催一下吗?"
answer2 = service.handle_request(user2)
# → Agentic RAG流程:
# 1. search_kb("订单查询") → 了解查询方法
# 2. query_order("ORD123456") → 查询状态
# 3. 发现未发货 → create_ticket("催单")
# 4. 生成回复
真相: RAG只是检索,不是理解。
问题:
解决方案:
# 混合策略
def hybrid_answer(query):
# 1. 先检索
docs = vectorstore.search(query)
if len(docs) == 0:
# 2. 检索失败 → 尝试Web搜索
web_results = brave_search(query)
docs = web_results
if len(docs) == 0:
# 3. 仍然失败 → 承认不知道
return "抱歉,我在知识库和网络上都没有找到相关信息"
# 4. 生成答案
return generate_answer(query, docs)
真相: Agent会犯错,会进入死循环,会浪费token。
常见问题:
防御措施:
class SafeAgent:
def __init__(self):
self.max_iterations = 10 # 最大迭代次数
self.max_tokens = 50000 # Token预算
self.used_tokens = 0
self.action_history = [] # 防止重复
def run(self, task):
for i in range(self.max_iterations):
# 1. 检查Token预算
if self.used_tokens > self.max_tokens:
return "错误: Token预算耗尽"
# 2. 规划行动
action = self.plan(task)
self.used_tokens += count_tokens(action)
# 3. 检测死循环
if self.is_repeating(action):
return "错误: 检测到重复操作,可能陷入死循环"
# 4. 执行并验证
result = self.execute(action)
if not self.validate_result(result):
# 执行失败 → 重新规划
continue
# 5. 判断是否完成
if self.is_complete(task, result):
return result
return "错误: 达到最大迭代次数"
def is_repeating(self, action):
"""检测是否重复执行相同操作"""
recent = self.action_history[-3:] # 检查最近3次
return recent.count(action) >= 2
真相: MCP只是一种可选的标准化方案,不用也能实现功能。
选择建议:
# 不用MCP的简单实现
def direct_api_call():
"""直接调用API"""
import requests
response = requests.get("https://api.example.com/data")
return response.json()
# 使用MCP的标准化实现
def mcp_call():
"""通过MCP协议"""
client = MCPClient("http://localhost:8080")
result = client.call_tool("example_api", {})
return result
# 如果只调用一两个API,直接调用更简单!
真相: OpenClaw很强,但有限制:
| 能做 | 不能做 |
|---|---|
| 文件操作、邮件处理 | 复杂的视觉任务(如图像生成) |
| 网络搜索、数据分析 | 实时音视频处理 |
| 跨应用自动化 | 需要GUI交互的复杂软件 |
| 代码辅助编写 | 替代专业程序员的深度开发 |
最佳实践:
真相: 大部分Skill只是封装好的函数调用,不是AI的智能。
# 所谓的"天气查询Skill"
def weather_skill(city):
return requests.get(f"https://api.weather.com/{city}").json()
# 所谓的"计算器Skill"
def calculator_skill(expression):
return eval(expression) # 危险!仅示例
# 这和传统编程有什么区别?没有!
真正的AI能力:
Skill只是工具,智能来自如何选择和组合使用这些工具。
| 技术 | 一句话总结 | 最大价值 | 最大陷阱 |
|---|---|---|---|
| Skill | 功能模块的新名字 | 标准化能力接口 | 过度包装,名不副实 |
| MCP | AI与外部系统的标准协议 | 统一生态,降低开发成本 | 协议开销,生态不成熟 |
| RAG | 给AI装上实时查资料能力 | 解决知识过时和幻觉 | 检索质量决定上限 |
| Agent | 能自主规划和执行的AI | 真正实现任务自动化 | 成本高,易出错 |
| OpenClaw | 集成上述所有技术的平台 | 开箱即用的完整方案 | 学习曲线,配置复杂 |
选择决策树:
你的需求是?
├─ 只需要问答 → 简单RAG
├─ 需要调用外部工具
│ ├─ 工具少(<5个) → 直接API调用
│ └─ 工具多(>5个) → 考虑MCP协议
├─ 需要自动化任务 → Agent系统
└─ 需要完整产品 → OpenClaw
预算评估:
- RAG: $0.01/query(Embedding) + $0.02/answer(LLM)
- Agent: $0.1-1/task(取决于复杂度)
- OpenClaw: $50-200/月(云端部署) 或 免费(本地)
不要被花哨的名词迷惑,理解底层逻辑才是关键!
如果这篇文章帮助你看穿了AI领域的"名词扎片",请点攒、收藏、分享!
