核心内容摘要
91在线看：开启你的无限视界，精彩触手可及
大多数公司都面临同样的现实数据无处不在。
您的邮箱里有邮件Slack里有对话GitHub里有代码Google Drive里有报表甚至可能还有一些没人记得密码的随机分析仪表板。
如果您想构建一个检索增强生成RAG系统那简直是一场噩梦。
您不能只连接一个数据源就万事大吉除非您对半生不熟的答案感到满意。
各大厂商早已深谙此道微软将RAG集成到M365产品中谷歌将其融入Vertex AI SearchAWS在Amazon Q Business中推出了自己的解决方案如果您完全投入某个生态系统这确实很美好。
但如果您想按自己的方式来做在本地运行完全不依赖云服务呢那么……让我们构建一个吧。
通过AI生成的粗略手绘风格工作流程图我们要构建什么我们将构建一个本地AI设置可以通过单一的聊天界面查询200多个不同的数据源使用以下技术•mcp-use来自YC S25的便捷工具用于本地MCP客户端•MindsDB用于连接所有这些数据源•Ollama用于在本地运行开源大语言模型一旦您理解了整个流程就相当简单您在聊天框中输入问题MCP客户端将其传递给MindsDB MCP服务器MindsDB确定您实际要查询的数据源它对数据源进行查询将结果发送给您的LLM然后——瞧——您得到了一个有实际上下文支撑的答案如果您更喜欢图片而非文字本质上就是这样您 → MCP客户端 → MindsDB MCP服务器 → 您的数据源 → LLM → 答案步骤1——本地运行MindsDB我们将使用Docker运行MindsDB。
打开终端并运行docker run -it -p 47334:47334 mindsdb/mindsdb就是这样无需注册无需云服务。
您现在正在自己的机器上运行MindsDB。
步骤2——获取MindsDB图形界面打开浏览器并访问http://localhost:47334您应该能看到MindsDB编辑器——有点像SQL工作台但内置了魔法。
从这里您可以连接数百个不同的数据源Slack、Gmail、Salesforce、MySQL应有尽有。
步骤3——连接一些数据这里是我连接的几个示例向您展示这有多简单-- SlackCREATE DATABASE slack_dbWITH ENGINE slack,PARAMETERS { token: your-slack-api-token};-- GmailCREATE DATABASE gmail_dbWITH ENGINE gmail,PARAMETERS { credentials: gmail_credentials.json};-- GitHubCREATE DATABASE github_dbWITH ENGINE github,PARAMETERS { token: your-github-token};-- Hacker NewsCREATE DATABASE hn_dbWITH ENGINE hacker_news;之后您可以运行常规的SQL查询例如SELECT * FROM gmail_db.inbox WHERE is_unread true;相当疯狂吧。
步骤4——MCP服务器配置好了MindsDB可以与我们的数据通信但现在我们希望它支持MCP协议以便我们的AI智能体可以使用它。
我们为此创建一个小的JSON文件{ servers: { mindsdb: { url: http://localhost:47334, tools: [list_databases, query] } }}现在我们的MCP客户端知道了MindsDB服务器的位置及其提供的工具。
步骤5——本地MCP客户端Ollama这部分实际上很有趣——只需要几行Python代码。
from mcp_use import MCPClientfrom ollama import ChatModel#
加载MCP客户端配置client MCPClient.from_config(mcp_config.json)#
连接本地LLMllm ChatModel(llama
#
创建智能体agent llm.bind(client)#
运行查询response agent.ask(Show me all unread Slack messages from this week.)print(response)这简直太简单了。
您只需将llama3改为mistral或gemma2就可以轻松切换LLM。
步骤6——Streamlit中的快速界面如果您想要一个不错的基于浏览器的聊天框谁不想要呢import streamlit as stst.title(MCP驱动的RAG)query st.text_input(问我任何问题)if st.button(运行): answer agent.ask(query) st.write(answer)运行它streamlit run app.py输入列出分配给我的所有GitHub问题然后见证魔法的发生。
这是一个单一的Python文件app.py它将整个本地RAG工作流程整合在一个地方——从MCP客户端通过MindsDB工具到本地LLMOllama全部封装在Streamlit界面中。
安装一次性# 如果您愿意创建一个虚拟环境然后pip install streamlit requests# 以下两个是按原样提供的如果您有它们就安装pip install mcp_use ollama# 如果这些包在您的pip环境中不可用# 请遵循mcp_use / ollama的安装文档并调整下面的导入。
app.py单文件RAG集成在一个框架中app.py单文件Streamlit应用演示本地RAG流程您界面→ MCP客户端 → MindsDBlist_databases / query→ Ollama LLM → 答案注意- 需要在http://localhost:47334运行MindsDBMindsDB Docker。
- 需要本地Ollama / LLM可通过ollama Python绑定访问。
- 假设有mcp_use Python客户端如果您的环境不同 请调整下面的小包装函数以调用正确的方法。
运行streamlit run app.pyimport jsonimport timeimport tracebackfrom typing import Any, Dict, List, Optionalimport streamlit as stimport requests# 尝试导入mcp_use和ollama。
如果缺失应用仍会启动并显示说明。
try: from mcp_use import MCPClient # 根据前面的示例期望在您的环境中 _HAS_MCP_USE Trueexcept Exception: MCPClient None _HAS_MCP_USE Falsetry: # Ollama python客户端的形状因版本而异这反映了早期的用法ChatModel(...) from ollama import ChatModel _HAS_OLLAMA Trueexcept Exception: ChatModel None _HAS_OLLAMA False# ---------------------------# 工具函数 / 包装器# ---------------------------DEFAULT_MCP_CONFIG { servers: { mindsdb: { url: http://localhost:47334, tools: [list_databases, query] } }}def ensure_mcp_config_file(path: str mcp_config.json): 如果文件不存在写入默认的MCP配置。
try: with open(path, x) as f: json.dump(DEFAULT_MCP_CONFIG, f, indent
print(f在{path}创建了默认MCP配置) except FileExistsError: passclass SimpleMCPWrapper: 对mcp_use.MCPClient的小型兼容包装器以提供 list_databases()和query()调用采用容错方式。
def __init__(self, config_path: str mcp_config.json): self.config_path config_path self.client None if _HAS_MCP_USE and MCPClient is not None: try: self.client MCPClient.from_config(config_path) except Exception as e: # 某些版本可能有不同的构造函数——尝试备选方案 try: self.client MCPClient(config_path) except Exception as e2: print(从库初始化MCPClient失败。
将在可能的情况下回退到HTTP。
) print(mcp_use错误, e, e
# 还存储MindsDB基本URL用于HTTP回退 with open(config_path, r) as f: cfg json.load(f) try: self.mindsdb_url cfg[servers][mindsdb][url].rstrip(/) except Exception: self.mindsdb_url http://localhost:47334 def list_databases(self) - List[str]: 返回数据库/数据源列表。
尝试客户端然后回退到HTTP。
# 尝试MCP客户端调用 if self.client: # 尝试常见的函数名 for name in (list_databases, list_databases_tool, list_tools, tools): fn getattr(self.client, name, None) if callable(fn): try: out fn() # 如果返回的是带有databases或类似的字典进行标准化 if isinstance(out, dict): if databases in out: return out[databases] # 尝试解析其他形状 return list(out.keys()) if isinstance(out, (list, tuple)): return list(out) except Exception: pass # 如果存在尝试通用的call或invoke接口 for name in (call_tool, invoke_tool, call): fn getattr(self.client, name, None) if callable(fn): try: # 某些MCP客户端期望服务器/工具名称 try: out fn(mindsdb, list_databases, {}) except TypeError: out fn(list_databases) if isinstance(out, (list, tuple)): return list(out) if isinstance(out, dict) and databases in out: return out[databases] except Exception: pass # HTTP回退到MindsDB REST尽力而为 # MindsDB在不同版本中没有单一的标准化公共列出数据库端点 # 但我们可以尝试调用模式或连接器端点。
try: # 尝试MindsDB的常见API端点 resp requests.get(f{self.mindsdb_url}/api/databases, timeout
if resp.ok: payload resp.json() # 尝试常见的形状 if isinstance(payload, dict): if data in payload and isinstance(payload[data], list): return [d.get(name) or d.get(database) or str(d) for d in payload[data]] return list(payload.keys()) if isinstance(payload, list): return [p.get(name) if isinstance(p, dict) else str(p) for p in payload] except Exception: pass # 如果到达这里返回一条有帮助的消息 return [无法枚举——检查MCP客户端或MindsDB] def query(self, query_text: str, max_rows: int
- Dict[str, Any]: 让MindsDB运行联合查询尽力而为。
此包装器首先尝试MCP客户端否则尝试MindsDB SQL端点。
# 尝试MCP客户端 if self.client: for name in (query, run_query, invoke_query, call_tool): fn getattr(self.client, name, None) if callable(fn): try: # 尝试几种参数形状 try: out fn(mindsdb, query, {query: query_text, max_rows: max_rows}) except TypeError: out fn(query_text) return {source: mcp_client, result: out} except Exception: # 继续尝试其他函数名 pass # 回退尝试MindsDB查询HTTP端点尽力而为使用/api/sql try: # 某些MindsDB版本公开/api/sql或/api查询尝试两者 for endpoint in (/api/sql, /api/query, /api/query/execute, /api/sql/execute): try: resp requests.post(self.mindsdb_url endpoint, json{query: query_text}, timeout
except Exception: continue if not resp.ok: continue try: data resp.json() except Exception: data resp.text return {source: mindsdb_http, result: data} except Exception: pass # 如果全部失败 return {source: error, error: 无法运行查询。
检查MCP客户端或MindsDB HTTP API。
}class OllamaWrapper: 向Ollama发送提示的最小包装器本地LLM。
def __init__(self, model_name: str llama
: self.model_name model_name self.model None if _HAS_OLLAMA and ChatModel is not None: try: # API各不相同——有些库是ChatModel(name...)其他使用client.chat(...) try: self.model ChatModel(model_name) except Exception: # 尝试不同的构造方式 self.model ChatModel(modelmodel_name) except Exception as e: print(初始化ollama.ChatModel失败, e) self.model None def chat(self, prompt: str, max_tokens: int
- str: # 尝试python绑定 if self.model: for fnname in (chat, generate, complete, __call__): fn getattr(self.model, fnname, None) if callable(fn): try: out fn(prompt) # 许多包装器返回复杂对象标准化为字符串 if isinstance(out, (str,)): return out if isinstance(out, dict) and text in out: return out[text] # 尝试字符串化 return str(out) except Exception: continue # 回退尝试本地Ollama HTTP如果Ollama守护进程在11434上监听 try: ollama_url http://localhost:11434/api/generate payload {model: self.model_name, prompt: prompt} resp requests.post(ollama_url, jsonpayload, timeout
if resp.ok: data resp.json() # 形状会不同尝试提取常见字段 if isinstance(data, dict): if text in data: return data[text] if response in data: return str(data[response]) return str(data) except Exception: pass return LLM不可用无法与Ollama Python客户端或HTTP端点通信。
# ---------------------------# 构建Streamlit界面# ---------------------------st.set_page_config(page_title本地RAGMCP MindsDB Ollama, layoutwide)st.title(本地RAG——单帧演示MCP → MindsDB → Ollama)st.markdown( 此演示期望- 在本地运行MindsDBDocker地址为http://localhost:47334- 本地Ollama LLM或Ollama HTTP端点地址为http://localhost:11434- 如果您想要直接绑定则需要mcp_use和ollama python库。
如果缺少任何部分界面仍将在可能的情况下通过HTTP回退运行查询。
)# 确保默认配置存在ensure_mcp_config_file(mcp_config.json)col1, col2 st.columns([1, 2])with col1: st.subheader(MCP / MindsDB) st.write(编辑MCP配置JSON) mcp_config_text st.text_area(mcp_config.json, valuejson.dumps(DEFAULT_MCP_CONFIG, indent
, height
if st.button(保存MCP配置): with open(mcp_config.json, w) as f: f.write(mcp_config_text) st.success(已保存mcp_config.json) time.sleep(
0.
if st.button(列出数据源): wrapper SimpleMCPWrapper(mcp_config.json) try: dbs wrapper.list_databases() st.write(数据源 / 数据库) st.json(dbs) except Exception as e: st.error(列出数据库失败 str(e)) st.text(traceback.format_exc()) st.markdown(---) st.write(针对MindsDB运行原始联合SQL查询尽力而为) raw_query st.text_area(原始SQL示例, valueSELECT * FROM gmail_db.inbox WHERE is_unread true LIMIT 20;, height
if st.button(运行原始查询): wrapper SimpleMCPWrapper(mcp_config.json) res wrapper.query(raw_query) if res.get(source) error: st.error(res.get(error)) else: st.write(结果来源 %s % res.get(source)) st.json(res.get(result))with col2: st.subheader(聊天 RAG查询 → MindsDB工具 → Ollama) st.write(高级查询纯英文。
应用将通过MCP包装器识别数据源运行查询并使用本地LLM进行
总结。
) user_q st.text_input(询问例如显示过去7天未读的Slack消息) model_name st.text_input(Ollama模型名称, valuellama
max_rows st.slider(要获取的最大行数联合查询, 1, 500,
if st.button(运行RAG查询) and user_q.strip(): st.info(正在运行RAG流程...) #
调用MCP包装器运行mindsdb query工具尽力而为 wrapper SimpleMCPWrapper(mcp_config.json) # 非常简单的启发式——在实践中您会调用MCP query工具来将 # 用户文本翻译成SQL或询问MindsDB的SQL转换器。
这里我们只是转发用户文本。
try: mindsdb_resp wrapper.query(user_q, max_rowsmax_rows) except Exception as e: st.error(MindsDB查询步骤失败 str(e)) mindsdb_resp {source: error, error: str(e)} if mindsdb_resp.get(source) error: st.error(MindsDB步骤无法运行。
请参阅下面的详细信息。
) st.json(mindsdb_resp) else: st.success(MindsDB返回了结果尽力而为。
) st.subheader(来自MindsDB的原始数据) st.json(mindsdb_resp.get(result)) #
使用本地LLM进行
总结/生成答案 st.subheader(LLM
总结 / 答案) ollama OllamaWrapper(model_namemodel_name) # 制作
总结提示包含用户问题 原始输出字符串化 # 保持提示简短以避免令牌爆炸——如果结果很大则进行采样或
总结模式 raw_text mindsdb_resp.get(result) # 将复杂负载转换为紧凑字符串 try: raw_preview json.dumps(raw_text, indent
[:4000] except Exception: raw_preview str(raw_text)[:4000] prompt ( 您是一个有用的助手。
用户询问\n\n f\{user_q}\\n\n 以下是从联合数据源检索的结果截断\n f{raw_preview}\n\n 请提供简洁、清晰且可操作的答案引用数据中的相关事实。
) # 调用聊天模型 llm_response ollama.chat(prompt) st.write(llm_response)st.markdown(---)st.caption(此示例故意采取防御性策略首先尝试Python绑定然后尝试HTTP回退。
\ 调整小包装器以匹配您安装的确切mcp_use / ollama API。
)。
## 学AI大模型的正确顺序千万不要搞错了2026年AI风口已来各行各业的AI渗透肉眼可见超多公司要么转型做AI相关产品要么高薪挖AI技术人才机遇直接摆在眼前有往AI方向发展或者本身有后端编程基础的朋友直接冲AI大模型应用开发转岗超合适就算暂时不打算转岗了解大模型、RAG、Prompt、Agent这些热门概念能上手做简单项目也绝对是求职加分王给大家整理了超全最新的AI大模型应用开发学习清单和资料手把手帮你快速入门学习路线:✅大模型基础认知—大模型核心原理、发展历程、主流模型GPT、文心一言等特点解析✅核心技术模块—RAG检索增强生成、Prompt工程实战、Agent智能体开发逻辑✅开发基础能力—Python进阶、API接口调用、大模型开发框架LangChain等实操✅应用场景开发—智能问答系统、企业知识库、AIGC内容生成工具、行业定制化大模型应用✅项目落地流程—需求拆解、技术选型、模型调优、测试上线、运维迭代✅面试求职冲刺—岗位JD解析、简历AI项目包装、高频面试题汇总、模拟面经以上6大模块看似清晰好上手实则每个部分都有扎实的核心内容需要吃透我把大模型的学习全流程已经整理好了抓住AI时代风口轻松解锁职业新可能希望大家都能把握机遇实现薪资/职业跃迁这份完整版的大模型 AI 学习资料已经上传CSDN朋友们如果需要可以微信扫描下方CSDN官方认证二维码免费领取【保证100%免费】