🧊 冻冻智能数据分析助手基于 LangChain + LM Studio 的 语义层 NL2SQL 数据分析系统，支持用自然语言查询业务数据，自动生成 SQL、可视化图表和分析结论。git地址https://github.com/zxliucn/Datas_Agent目录项目概述技术架构项目结构核心模块4.1 主入口 (app.py)4.2 意图解析 (core/intent_parser.py)4.3 SQL 构建 (core/sql_builder.py)4.4 SQL 执行 (core/sql_executor.py)4.5 图表生成 (core/chart_generator.py)4.6 洞察生成 (core/insight_generator.py)查询模板系统 (templates)Web 界面 (web)工具模块 (utils)配置说明数据库表结构快速开始API 接口项目预览1. 项目概述冻冻智能数据分析助手是一个基于语义层 NL2SQL 架构的数据分析系统，采用 预定义模板 + LLM 意图解析 的安全方案，避免 LLM 直接生成 SQL 带来的安全风险。核心特性功能说明🔍 自然语言查询输入自然语言问题，自动解析意图并生成 SQL📊 可视化图表根据查询类型自动选择折线图、柱状图、饼图等💡 智能洞察从查询结果中提取关键结论和业务建议🔒 安全可控预定义 SQL 模板，LLM 仅负责意图解析，不直接生成 SQL🌐 Web 界面基于 Gradio 的可视化交互界面支持的查询类型查询类型示例问题商品销量虾滑卖了多少？上个月牛筋的销售额商品排行这个月哪个商品卖得最好？销量前10订单趋势最近7天的订单趋势订单列表待发货的订单有哪些？客户排行哪个客户花钱最多？库存预警库存不足的商品售后排行退款最多的商品经营概况本月经营情况2. 技术架构┌─────────────────────────────────────────────────────────┐ │ Gradio Web UI │ │ 输入自然语言问题 → 展示图表和洞察结论 │ └──────────────────────┬──────────────────────────────────┘ │ ▼ ┌─────────────────────────────────────────────────────────┐ │ NL2SQL 主控制器 │ │ 意图解析 → SQL 构建 → 执行 → 可视化 │ └──────────────────────┬──────────────────────────────────┘ │ ┌──────────────┼──────────────┐ ▼ ▼ ▼ ┌──────────────┐ ┌────────────┐ ┌──────────────┐ │ IntentParser │ │ SQLBuilder │ │ SQLExecutor │ │ LLM 意图解析 │ │ 模板拼装 │ │ MySQL 查询 │ └──────────────┘ └────────────┘ └──────────────┘ │ ┌──────────────┼──────────────┐ ▼ ▼ ▼ ┌──────────────┐ ┌────────────┐ ┌──────────────┐ │ChartGenerator│ │ InsightGen │ │ QueryTemplates│ │ 图表可视化 │ │ 洞察分析 │ │ SQL 模板库 │ └──────────────┘ └────────────┘ └──────────────┘设计思路采用 语义层 NL2SQL 方案，而非让 LLM 直接生成 SQL：LLM 只做意图解析：理解用户自然语言，输出结构化的 {intent, params} JSONSQL 由模板拼装：每种查询类型对应预定义的 SQL 模板，安全可控参数映射处理：通过 param_mapping 将自然语言参数（如"上个月"）转换为 SQL 条件优势：✅ 安全：杜绝 SQL 注入风险✅ 可控：SQL 逻辑可审查、可维护✅ 稳定：不依赖 LLM 生成正确 SQL 的能力✅ 可扩展：新增查询类型只需添加模板3. 项目结构Datas_agent/ ├── app.py # 主入口，启动服务 ├── config.py # 配置文件（数据库、LLM） ├── .env # 环境变量 ├── requirements.txt # 依赖清单 │ ├── core/ # 核心业务模块 │ ├── __init__.py │ ├── intent_parser.py # LLM 意图解析 │ ├── sql_builder.py # SQL 模板拼装 │ ├── sql_executor.py # SQL 执行器 │ ├── chart_generator.py # 图表生成（Plotly） │ └── insight_generator.py # 洞察分析生成 │ ├── templates/ # SQL 查询模板 │ ├── __init__.py │ └── query_templates.py # 8 种查询模板定义 │ ├── web/ # Web 界面 │ ├── __init__.py │ └── gradio_app.py # Gradio 界面 │ ├── utils/ # 工具模块 │ ├── __init__.py │ └── logger.py # 日志与计时 │ └── preview/ # 预览截图 ├── p1.png └── p2.png4. 核心模块4.1 主入口 (app.py)NL2SQL 主控制器，串联整个处理流程：class NL2SQL: def __init__(self, llm): self.intent_parser = IntentParser(llm) # LLM 意图解析 self.sql_builder = SQLBuilder() # SQL 模板拼装 def convert(self, question): # 1. 解析意图 → {intent, params} result = self.intent_parser.parse_with_defaults(question) # 2. 构建 SQL sql = self.sql_builder.build(intent, params) return intent, sql, params启动流程：初始化 LLM → 创建各模块 → 测试数据库连接 → 启动 Gradio Web 界面（端口 7861）4.2 意图解析 (core/intent_parser.py)使用 LLM 将自然语言问题解析为结构化的意图和参数。处理流程：将用户问题 + 预定义的意图列表作为 Prompt 发送给 LLMLLM 返回 JSON 格式的 {intent, params}提取 JSON 并解析，失败则返回 unknown调用 parse_with_defaults() 自动填充默认参数Prompt 策略：提供 8 种查询类型的触发词、参数说明和示例强制 JSON 输出格式使用 %s 占位符避免 format 冲突输出示例：// 输入："上个月虾滑卖了多少" {"intent": "product_sales", "params": {"product": "虾滑", "time_range": "last_month"}} // 输入："销量前10的商品" {"intent": "product_ranking", "params": {"top_n": 10}}4.3 SQL 构建 (core/sql_builder.py)根据意图和参数，从预定义模板拼装 SQL。构建流程：从模板库获取 SQL 模板应用默认值（如 limit=10, time_range=this_month）通过 param_mapping 映射参数（如将 time_range 转换为 AND DATE(o.order_time) = ...）填充模板占位符清理 SQL（修复空 WHERE、多余 AND 等边界情况）4.4 SQL 执行 (core/sql_executor.py)基于 SQLAlchemy + PyMySQL 执行 SQL 查询。仅允许 SELECT 查询，拒绝写操作返回 pandas DataFrame启动时自动测试数据库连接4.5 图表生成 (core/chart_generator.py)基于 Plotly 根据查询意图自动选择图表类型：意图类型图表类型说明order_trend折线图时间趋势，支持多指标product_ranking柱状图排行榜，限显前 15 条customer_ranking柱状图（绿色）客户消费排行stock_warning柱状图（红色）库存预警after_sale_ranking柱状图（橙色）售后排行product_salesMarkdown 指标卡单商品销量详情business_overviewMarkdown 表格经营概况其他自动判断饼图/柱状图/折线图4.6 洞察生成 (core/insight_generator.py)从查询结果中提取关键结论和业务建议，输出 Markdown 格式。分析维度：商品排行：Top3、总销量、集中度分析（Top3 占比）订单趋势：日均数据、升降趋势判断、峰值日期客户排行：VIP 客户识别、人均消费库存预警：最紧急商品、按分类统计、补货建议售后排行：退款率分析、改进建议经营概况：核心指标表格、退款率评级5. 查询模板系统 (templates)query_templates.py 定义了 8 种查询模板，每种模板包含：字段说明templateSQL 模板字符串，{placeholder} 占位required_params必填参数列表optional_params可选参数列表param_mapping参数到 SQL 条件的映射函数defaults默认值字典时间范围转换：支持 today、yesterday、this_week、this_month、last_month、last_7_days、last_30_days，自动转换为 MySQL 日期函数。参数映射函数示例：# 时间范围 → WHERE 条件 "time_range": lambda x: f"AND {get_time_condition(x)}" # 商品分类 → 分类过滤 "category": lambda x: f"AND p.category = '{x}'" # 经营概况 → 同时返回时间条件和中文时间名（tuple） "time_range": lambda x: (get_time_condition(x), get_time_range_name(x))6. Web 界面 (web)基于 Gradio 的交互式 Web 界面。界面布局：输入区：文本框 + 开始分析 / 清空按钮SQL 展示：代码块展示生成的 SQL图表区：Plotly 交互式图表洞察区：Markdown 格式的分析结论内置示例问题：提供各查询类型的示例问题，引导用户使用。7. 工具模块 (utils)日志模块 (logger.py)info / debug / error / warn 四级日志@timing 装饰器：记录函数执行耗时通过 APP_CONFIG['debug'] 控制是否输出 DEBUG 级别日志8. 配置说明环境变量 (.env)# MySQL 数据库 MYSQL_HOST=localhost MYSQL_PORT=3306 MYSQL_USER=your_username MYSQL_PASSWORD=your_password MYSQL_DATABASE=your_database # LM Studio 本地模型 # base_url: LM Studio API 地址 # model: 使用的模型（如 qwen/qwen3.5-9b） # temperature: 0（意图解析需要确定性输出）应用配置 (config.py)LM_STUDIO_CONFIG = { 'base_url': 'http://127.0.0.1:1234/v1', 'model': 'qwen/qwen3.5-9b', 'temperature': 0, 'max_tokens': 512, } APP_CONFIG = { 'default_limit': 100, 'max_limit': 1000, 'debug': True, }9. 数据库表结构系统依赖以下 MySQL 表：表名说明关键字段products商品表product_id, name, category, price, stockorders订单表order_id, user_id, order_time, total_amount, order_statusorder_items订单明细表order_id, product_id, product_name, quantity, total_priceusers用户表id, username, nicknameafter_sale_item售后表order_id, product_name, refund_num, refund_amount10. 快速开始安装依赖pip install -r requirements.txt主要依赖：langchain-openai — LLM 调用（兼容 OpenAI API 格式）gradio — Web 界面plotly — 图表可视化pandas — 数据处理sqlalchemy + pymysql — 数据库连接python-dotenv — 环境变量管理配置启动 LM Studio 并加载模型（如 Qwen3.5-9B），确保 API 服务运行在 http://127.0.0.1:1234/v1修改 .env 文件中的 MySQL 连接信息启动服务python app.py启动成功后访问：http://localhost:7861试试这些问题虾滑卖了多少？ 这个月哪个商品卖得最好？ 最近7天的订单趋势 待发货的订单有哪些？ 哪个客户花钱最多？ 库存不足的商品 退款最多的商品 本月经营情况11. API 接口系统通过 Gradio 提供交互式 Web 接口（端口 7861），无需额外 API 调用。直接在浏览器中访问即可使用。如需集成到其他系统，可参考 NL2SQL.convert() 方法封装为 REST API：from app import NL2SQL from core import SQLExecutor nl2sql = NL2SQL(llm) executor = SQLExecutor() intent, sql, params = nl2sql.convert("虾滑卖了多少") df = executor.execute(sql)12. 项目预览系统提供基于 Gradio 的 Web 数据分析界面，输入自然语言即可获得 SQL、图表和洞察结论。界面预览![数据分析界面-主视图]图 1：智能数据分析助手主界面 — 自然语言输入与结果展示![数据分析界面-功能展示]图 2：查询结果展示 — SQL 语句、可视化图表与分析结论总结特性实现方式🔒 安全可控语义层模板方案，LLM 仅解析意图，不直接生成 SQL🧠 智能解析LangChain + LM Studio 本地大模型📊 自动可视化Plotly 按意图自动匹配图表类型💡 业务洞察基于查询结果生成结构化分析结论🚀 快速扩展新增查询类型只需添加模板，无需修改 LLM 逻辑
冻冻网络智能客服系统基于 LangGraph + Ollama 的企业级智能客服系统，支持商品查询、订单查询、售后进度查询、烹饪指南等功能。git地址https://github.com/zxliucn/Customer_Services目录项目概述技术架构项目结构核心模块4.1 入口文件4.2 Agent 图结构4.3 状态管理4.4 LLM 配置工具模块 (tools)5.1 商品工具5.2 订单工具5.3 售后工具5.4 信息提取器5.5 用户与会话Agent 节点 (agent/node)6.1 意图识别6.2 路由6.3 工具调用6.4 RAG 检索6.5 回答生成6.6 直接回答6.7 帮助信息6.8 转人工RAG 系统配置说明数据库表结构快速开始项目预览API 接口1. 项目概述冻冻网络智能客服是一个基于 LangGraph 构建的企业级对话系统，采用多节点协作架构，支持：功能说明📦 商品查询价格、规格、库存🚚 订单查询订单状态、物流进度💰 售后进度退款/退货/换货进度🍳 烹饪指南食材做法食谱👨‍💼 转人工投诉等复杂问题转人工核心特性：🎯 意图识别 + 规则匹配混合决策🔧 工具调用：订单/商品/售后数据库查询📚 RAG 检索：用户手册 + 烹饪指南💾 多轮对话记忆 (MemorySaver)🏷️ 游客/登录用户区分2. 技术架构┌─────────────────────────────────────────────────────────────┐ │ FastAPI 入口 │ │ /chat, /health │ └─────────────────────────┬───────────────────────────────────┘ │ ┌─────────────────────────▼───────────────────────────────────┐ │ LangGraph Agent │ │ ┌──────────┐ ┌──────────┐ ┌──────────┐ │ │ │ Intent │───▶│ Router │───▶│ Nodes │ │ │ │ 意图识别 │ │ 路由 │ │ 执行 │ │ │ └──────────┘ └──────────┘ └──────────┘ │ │ │ │ │ ┌─────────┬─────────┬─────────┬──┘ │ │ ▼ ▼ ▼ ▼ │ │ ┌────────┐┌────────┐┌────────┐┌────────┐ │ │ │ Tool ││ RAG ││ Direct ││ Help │ │ │ │工具调用 ││知识检索 ││直接回答 ││帮助信息 │ │ │ └────────┘└────────┘└────────┘└────────┘ │ │ │ │ │ │ └────┬────┘ │ │ ▼ │ │ ┌────────────┐ │ │ │ Generate │ │ │ │ 回答生成 │ │ │ └────────────┘ │ └─────────────────────────┬───────────────────────────────────┘ │ ┌─────────────────────┼─────────────────────┐ ▼ ▼ ▼ ┌────────┐ ┌────────────┐ ┌──────────┐ │ MySQL │ │ Chroma │ │ Ollama │ │ 数据库 │ │ 向量数据库 │ │ LLM │ └────────┘ └────────────┘ └──────────┘技术栈：层级技术选型对话框架LangGraphLLMOllama (dd_model)EmbeddingQwen3-Embedding-0.6B向量库Chroma数据库MySQLWeb 框架FastAPI3. 项目结构Customer_Services/ ├── main.py # FastAPI 入口 ├── requirements.txt # 依赖 ├── .env # 环境变量 │ ├── config/ # 配置模块 │ ├── settings.py # 全局配置 │ ├── database.py # 数据库连接 │ └── logging_config.py # 日志配置 │ ├── agent/ # Agent 核心 │ ├── graph.py # LangGraph 图定义 │ ├── state.py # 状态定义 │ ├── llm.py # LLM 配置 │ └── node/ # 节点实现 │ ├── intent.py # 意图识别 │ ├── router.py # 路由决策 │ ├── tool.py # 工具调用 │ ├── rag.py # RAG 检索 │ ├── generate.py # 回答生成 │ ├── direct.py # 直接回答 │ ├── help.py # 帮助信息 │ └── human.py # 转人工 │ ├── tools/ # 工具函数 │ ├── product_tools.py # 商品查询 │ ├── order_tools.py # 订单查询 │ ├── after_sale_tools.py # 售后查询 │ ├── extractors.py # 信息提取 │ ├── user_tools.py # 用户管理 │ ├── guest_tools.py # 游客管理 │ ├── conversation_tools.py # 会话管理 │ └── message_tools.py # 消息管理 │ ├── rag/ # RAG 系统 │ ├── retriever.py # 检索器 │ └── data/ # 向量数据 │ ├── chroma_db/ # Chroma 向量库 │ ├── 冻品商城用户使用手册.md │ └── 食材烹饪方式指南.md │ ├── data/ # SQL 脚本 │ ├── dongdong_products.sql │ ├── dongdong_orders.sql │ ├── dongdong_after_sale.sql │ └── other.sql │ └── logs/ # 日志目录 ├── app_*.log └── error_*.log4. 核心模块4.1 入口文件 (main.py)# 启动服务 uvicorn main:app --host 0.0.0.0 --port 8000核心接口：接口方法说明/healthGET健康检查/chatPOST对话接口/GET根路径信息请求示例：curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{"message": "虾滑多少钱", "session_id": "test123"}'4.2 Agent 图结构 (agent/graph.py)def build_graph(): graph = StateGraph(AgentState) # 添加节点 graph.add_node("intent", intent_node) # 意图识别 graph.add_node("help", help_node) # 帮助信息 graph.add_node("tool", tool_node) # 工具调用 graph.add_node("rag", rag_node) # RAG检索 graph.add_node("generate", generate_node) # 生成回答 graph.add_node("direct", direct_node) # 直接回答 graph.add_node("human", human_node) # 转人工 # 条件路由 graph.add_conditional_edges("intent", route_by_intent, {...}) return graph.compile(checkpointer=MemorySaver())4.3 状态管理 (agent/state.py)class AgentState(TypedDict): messages: Annotated[List[dict], add_messages] # 对话历史 intent: str # 当前意图 user_id: Optional[int] # 用户ID is_guest: bool # 是否游客 extracted_order_id: Optional[str] # 提取的订单号 extracted_product_name: Optional[str] # 提取的商品 extracted_after_sale_id: Optional[str] # 提取的售后单号 tool_result: Optional[str] # 工具返回结果 rag_context: Optional[str] # RAG检索结果 final_response: str # 最终回答 need_human: bool # 是否需要转人工4.4 LLM 配置 (agent/llm.py)def get_llm(temperature: float = 0.3): return ChatOllama( base_url="http://localhost:11434", model="dd_model", temperature=temperature, )5. 工具模块 (tools)5.1 商品工具 (product_tools.py)函数功能query_product(user_input)查询单个商品信息search_products(keyword)关键词搜索商品check_stock(product_name, quantity)检查库存返回示例：【虾滑】 商品ID: P001 价格: 29.9元/袋 库存: 100件 (有货) 保质期: 12个月 储存方式: -18°C冷冻5.2 订单工具 (order_tools.py)函数功能query_order(user_input, user_id, is_guest)查询订单详情get_order_status(order_id)获取订单状态get_user_orders(user_id, limit)获取订单列表5.3 售后工具 (after_sale_tools.py)函数功能query_after_sale_progress(user_input, user_id, is_guest)查询售后进度query_refund_progress(order_id)查询退款进度5.4 信息提取器 (extractors.py)函数功能extract_order_id(text)提取订单号extract_after_sale_id(text)提取售后单号extract_product_name(text)提取商品名extract_phone(text)提取手机号extract_all(text)提取所有信息正则模式：# 订单号: O2025010006, DD202412340001 r'(?<![A-Za-z0-9])O\d{6,12}(?![A-Za-z0-9])' r'(?<![A-Za-z0-9])DD\d{8,12}(?![A-Za-z0-9])' # 售后单号: AS202412340001 r'AS\d{8,}' # 手机号: 1[3-9]\d{9} r'1[3-9]\d{9}'5.5 用户与会话模块函数功能user_toolsget_or_create_user()获取/创建用户guest_toolsgenerate_guest_id()生成游客IDguest_toolscreate_guest_session()创建游客会话conversation_toolsget_or_create_conversation()获取/创建会话conversation_toolsupdate_conversation_count()更新消息计数message_toolssave_message()保存消息message_toolsget_conversation_messages()获取历史消息6. Agent 节点 (agent/node)6.1 意图识别 (intent.py)两层意图识别：用户输入 → 规则匹配(rule_match) → LLM补充(llm_intent)意图类型：意图关键词路由节点help你好、在吗、帮助helpproduct商品、价格、库存toolorder订单、快递、物流toolafter_sale退款、售后toolcooking怎么做、怎么吃ragmanual怎么操作、流程ragcomplaint投诉、生气humanchat闲聊direct6.2 路由 (router.py)def route_by_intent(state: dict) -> str: routing = { "help": "help", "product": "tool", "order": "tool", "after_sale": "tool", "cooking": "rag", "manual": "rag", "complaint": "human", "chat": "direct", } return routing.get(intent, "direct")6.3 工具调用 (tool.py)def tool_node(state: dict) -> dict: intent = state.get("intent") if intent == "product": result = query_product(user_input) elif intent == "order": result = query_order(user_input, user_id, is_guest) elif intent == "after_sale": result = query_after_sale_progress(user_input, user_id, is_guest) return {"tool_result": result, ...}6.4 RAG 检索 (rag.py)def rag_node(state: dict) -> dict: if intent == "cooking": context = retriever.search_cooking(query, top_k=2) elif intent == "manual": context = retriever.search_manual(query, top_k=2) return {"rag_context": context}向量集合：集合名用途cooking_guide烹饪指南user_manual用户手册6.5 回答生成 (generate.py)def generate_node(state: dict) -> dict: system_prompt = f"""你是{COMPANY_NAME}的智能客服... {full_context} 用户问：{user_input} 请根据参考资料直接回答：""" response = llm.invoke([SystemMessage(content=system_prompt), ...]) return {"final_response": response.content}6.6 直接回答 (direct.py)用于闲聊/问候，保持对话上下文记忆。6.7 帮助信息 (help.py)def get_help_message() -> str: return f"""📖 **您好** 我是{COMPANY_NAME}智能助手**冻冻**，可以帮你： 📦 **查询商品** - 价格、规格、库存 🚚 **查询订单** - 订单状态、物流 💰 **售后进度** - 退款/退货进度 🍳 **烹饪指南** - 食材做法 👨‍💼 **转人工** - 需要人工服务 --- 客服热线：{HOTLINE}"""6.8 转人工 (human.py)def human_node(state: dict) -> dict: return { "final_response": "🙋 好的，正在为您转接人工客服...", "need_human": True }7. RAG 系统检索器 (rag/retriever.py)class RAGRetriever: def __init__(self, persist_dir="./rag/data/chroma_db"): self.embed_model = HuggingFaceEmbedding( model_name="Qwen/Qwen3-Embedding-0.6B", trust_remote_code=True ) self.chroma_client = chromadb.PersistentClient(path=persist_dir) def search_cooking(self, query, top_k=2): """检索烹饪指南""" index = self._get_index("cooking_guide") return index.as_retriever(similarity_top_k=top_k).retrieve(query) def search_manual(self, query, top_k=2): """检索用户手册""" index = self._get_index("user_manual") return index.as_retriever(similarity_top_k=top_k).retrieve(query)8. 配置说明环境变量 (.env)# Ollama 配置 OLLAMA_BASE_URL=http://localhost:11434 MODEL_NAME=dd_model # Embedding 模型 EMBEDDING_MODEL_PATH=/Users/t-mac/.cache/.../Qwen3-Embedding-0.6B # MySQL 数据库 MYSQL_HOST=localhost MYSQL_PORT=3306 MYSQL_USER=root MYSQL_PASSWORD= MYSQL_DATABASE=frozen_shop # API 配置 API_HOST=0.0.0.0 API_PORT=8000 # 日志 LOG_LEVEL=INFO9. 数据库表结构核心表表名用途users用户信息guest_sessions游客会话conversations会话记录messages消息历史products商品信息orders订单信息order_items订单明细after_sale售后记录after_sale_item售后商品明细10. 快速开始安装依赖pip install -r requirements.txt启动 Ollamaollama serve ollama pull dd_model启动服务python main.py # 或 uvicorn main:app --host 0.0.0.0 --port 8000启动成功后，访问 Web 界面在浏览器中打开项目目录下的 talk.html：# Mac 系统 open talk.html # 或者直接用浏览器打开 # file:///Users/t-mac/workspace/AI/Customer_Services/talk.html界面功能：👤 支持切换用户（游客、用户2、用户3、用户4）💬 快捷按钮：查订单、查售后、烹饪方法、转人工🔄 新会话按钮：重置对话历史测试对话curl -X POST http://localhost:8000/chat \ -H "Content-Type: application/json" \ -d '{"message": "虾滑多少钱", "session_id": "test123"}'12. 项目预览系统提供 Web 聊天界面，可直接在浏览器中与客服对话、查询商品和订单信息。对话界面预览图 1：智能客服主界面 — 支持自然语言商品查询与订单管理图 2：多轮对话与转人工功能 — 展示完整对话流程与售后处理13. API 接口健康检查GET /health响应：{ "status": "healthy", "service": "冻冻网络", "version": "1.0.0" }对话接口POST /chat Content-Type: application/json { "message": "虾滑多少钱", "session_id": "user_123" }响应：{ "response": "【虾滑】价格：29.9元/袋，库存：100件...", "session_id": "user_123", "intent": "product" }总结本系统采用 LangGraph 状态图架构，通过意图识别 → 路由 → 节点执行 → 回答生成的流程，实现了一个功能完整的智能客服系统。核心亮点：混合决策：规则匹配 + LLM 意图识别工具驱动：数据库查询 (订单/商品/售后)RAG 增强：烹饪指南、用户手册知识库可扩展：节点化设计，易于添加新功能
项目结构llamaindex-rag-example/├── data/ # 存放待处理的原始文档（如PDF/TXT/Markdown等）├── storage_exp/ # 持久化存储向量索引的目录（自动生成）├── 01_build.py # 构建并持久化向量索引的脚本└── 02_query.py # 加载索引并进行问答查询的脚本PDF文件转Markdown#如果直接用 pdf版解析会很差，可以用mineru转化为md文件，但是测试 api 速度也不快， 可以直接在官网将文档转化下载，时间最快，还可以进行调整 官网：https://mineru.net/ 01_build.py 构建索引import os import time import torch from pathlib import Path from llama_index.core import VectorStoreIndex, Settings, Document from llama_index.embeddings.huggingface import HuggingFaceEmbedding from llama_index.core.node_parser import SimpleNodeParser # 配置 device = "cpu" #如果直接用 pdf版解析会很差，可以用mineru转化为md文件，但是测试 api 速度也不快， 可以直接在官网将文档转化下载，时间最快，还可以进行调整https://mineru.net/ markdown_path = "./data/test_manual.md" persist_dir = "./storage_exp" # 创建目录 os.makedirs(persist_dir, exist_ok=True) os.makedirs("./models", exist_ok=True) # 2. 配置全局 Embedding（本地 Qwen3 模型） EMBEDDING_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" Settings.embed_model = HuggingFaceEmbedding( model_name=EMBEDDING_PATH, trust_remote_code=True ) # 读取 Markdown 文件 print(f"读取 Markdown: {markdown_path}") with open(markdown_path, "r", encoding="utf-8") as f: markdown_content = f.read() print(f"文档长度: {len(markdown_content)} 字符") # 创建 Document doc = Document( text=markdown_content, metadata={ "source": Path(markdown_path).name, "file_path": str(Path(markdown_path).absolute()), } ) # 切分文档 print("切分文档...") parser = SimpleNodeParser(chunk_size=512, chunk_overlap=50) nodes = parser.get_nodes_from_documents([doc]) print(f"生成 {len(nodes)} 个节点") # 构建索引 print("构建索引...") start_time = time.time() index = VectorStoreIndex(nodes) build_time = time.time() - start_time print(f"索引构建完成，耗时 {build_time:.2f} 秒") # 保存索引 print(f"保存索引到 {persist_dir}...") index.storage_context.persist(persist_dir=persist_dir) # 统计 index_size = 0 for root, dirs, files in os.walk(persist_dir): for file in files: index_size += os.path.getsize(os.path.join(root, file)) print("\n" + "="*50) print("构建完成!") print("="*50) print(f"文档长度: {len(markdown_content):,} 字符") print(f"节点数量: {len(nodes)}") print(f"索引大小: {index_size / 1024:.1f} KB") print(f"总耗时: {build_time:.2f} 秒") print("="*50) 02_query.pyimport os import sys import time from llama_index.core import StorageContext, load_index_from_storage, Settings from llama_index.llms.openai import OpenAI from llama_index.llms.openai_like import OpenAILike from llama_index.core.prompts import PromptTemplate from llama_index.core.postprocessor import SentenceTransformerRerank from llama_index.retrievers.bm25 import BM25Retriever # 导入依赖 from llama_index.core import VectorStoreIndex, SimpleDirectoryReader, Settings from llama_index.embeddings.huggingface import HuggingFaceEmbedding import os from dotenv import load_dotenv # 加载环境变量 load_dotenv() QIANWEN_KEY = os.getenv("QIANWEN_API_KEY") os.environ["TOKENIZERS_PARALLELISM"] = "false" # 配置 device = "mps" persist_dir = "./storage_exp" lm_studio_url = "http://localhost:1234/v1" # 2. 配置全局 Embedding（本地 Qwen3 模型） EMBEDDING_PATH = "/Users/t-mac/.cache/huggingface/hub/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" Settings.embed_model = HuggingFaceEmbedding( model_name=EMBEDDING_PATH, trust_remote_code=True ) # 1. 配置全局 LLM（通义千问） Settings.llm = OpenAILike( model="qwen3.5-flash", api_base="https://dashscope.aliyuncs.com/compatible-mode/v1", api_key=QIANWEN_KEY, is_chat_model=True, temperature=0.1 ) # 重排序器 # 中文专用重排模型 # reranker = SentenceTransformerRerank( # model="BAAI/bge-reranker-base", # top_n=5 # ) # 提示词模板 PROMPT = PromptTemplate( """你是一个专业的 HR 助手，请基于以下员工手册内容回答问题。 {context_str} 问题：{query_str} 回答：""" ) # 加载索引 print("加载索引...") storage_context = StorageContext.from_defaults(persist_dir=persist_dir) index = load_index_from_storage(storage_context) print(f"索引加载成功，共 {len(index.docstore.docs)} 个节点") def hybrid_search(query, top_k=5): """混合检索""" # 向量检索 vector_retriever = index.as_retriever(similarity_top_k=10) vector_nodes = vector_retriever.retrieve(query) # BM25 检索 all_nodes = [] for doc_id in index.docstore.docs.keys(): doc = index.docstore.get_document(doc_id) if hasattr(doc, 'nodes'): all_nodes.extend(doc.nodes) else: all_nodes.append(doc) bm25_retriever = BM25Retriever.from_defaults( nodes=all_nodes, similarity_top_k=10, ) bm25_nodes = bm25_retriever.retrieve(query) # 合并去重 seen = set() merged = [] for node in vector_nodes + bm25_nodes: if node.node.node_id not in seen: seen.add(node.node.node_id) merged.append(node) return merged[:top_k] # # 重排序 # reranked = reranker.postprocess_nodes(merged, query_str=query) # return reranked[:top_k] def generate_answer(query, nodes): """生成回答""" from llama_index.core import get_response_synthesizer synthesizer = get_response_synthesizer(text_qa_template=PROMPT) return synthesizer.synthesize(query, nodes=nodes) def query(question): """执行查询""" nodes = hybrid_search(question) response = generate_answer(question, nodes) return response.response if __name__ == "__main__": question = "员工违约的条件有哪些？" print(f"\n问题: {question}") print("-" * 40) start = time.time() answer = query(question) elapsed = time.time() - start print(f"回答: {answer}") print("-" * 40) print(f"耗时: {elapsed:.2f} 秒") 测试结果/opt/miniconda3/envs/langchain-dev/bin/python /Users/t-mac/workspace/Python/python_code/24-LlamaIndex/projects/llamaindex-rag-example/02_query.py W0522 23:40:21.752000 33319 site-packages/torch/distributed/elastic/multiprocessing/redirects.py:35] NOTE: Redirects are currently not supported in MacOs. W0522 23:40:21.774000 33319 site-packages/torch/utils/_pytree.py:630] <enum 'KernelPreference'> is an Enum subclass and is now natively supported by torch.compile as an opaque value type. Calling register_constant() on Enum subclasses is deprecated and will be an error in a future release. W0522 23:40:21.793000 33319 site-packages/torch/utils/_pytree.py:630] <enum 'ScaleCalculationMode'> is an Enum subclass and is now natively supported by torch.compile as an opaque value type. Calling register_constant() on Enum subclasses is deprecated and will be an error in a future release. Loading weights: 100%|██████████| 310/310 [00:00<00:00, 8048.99it/s] 加载索引... 索引加载成功，共 98 个节点 问题: 员工违约的条件有哪些？ ---------------------------------------- 回答: 你好！我是你的专业 HR 助手。根据提供的《员工手册》内容，虽然手册中主要使用“违纪”、“严重违纪”及“解除劳动合同”等术语，但以下情况均属于员工违反公司规章制度或劳动合同约定的条件（即通常理解的“违约”情形）。 以下是基于文档整理的员工违纪及违规的具体条件分类： ### 1. 考勤与出勤违规 * **迟到早退：** 连续 30 天内迟到或早退三次及以上（10.1.1）；试用期内迟到早退累计达三次或以上（4.5）。 * **旷工：** * 未经请假或请假未获批准擅自不到岗（3.1）。 * 请假期满未续假或续假未批逾期不归（3.3.2）。 * 特殊原因未在当日上班 1 小时内请假，或事后 3 个工作日内未补办手续（3.3.3）。 * 不服从工作调动、工时制度调整拒不执行导致未正常出勤（3.3.4）。 * **后果阈值：** 连续旷工 2 天及以上，或一年内累计旷工 3 天及以上（10.1.2）；连续旷工 3 天及以上，或全年累计旷工 7 天及以上（3.2）。 * **打卡异常：** 委托他人打卡或替他人打卡（10.1.3）；连续 30 天累计 6 次以上未打卡且无法说明原因或提供证明（1.4 / 10.1.4）。 * **试用期出勤率：** 试用期实际出勤日/应出勤日低于 85%（4.5）。 ### 2. 诚信与入职信息违规 * **虚假信息：** 伪造学历、工作经历、证书，简历、入职登记信息、健康状况等与真实情况不符（4.6）。 * **隐瞒欺骗：** 有欺骗、隐瞒行为（4.6）。 * **价值观问题：** 诚信等价值观存在问题，无论服务年限长短，均可能导致立即解除劳动合同（1.3）。 * **竞业限制：** 入职本公司将违反竞业限制义务（4.15）。 ### 3. 行为规范与安全违规 * **扰乱秩序：** 发起、参与离岗、怠工、罢工、示威，或鼓动他人参与（10.2.1）。 * **骚扰行为：** 侮辱、诽谤、辱骂、造谣、威胁、恐吓他人（10.2.2）。 * **现场管理：** 将无关人员带入工作现场，情节严重或造成损害（10.2.3）。 * **安全隐患：** 在库房等高风险区域内吸烟，情节严重或造成损害（10.2.4）。 * **资产损害：** 损坏或遗失公司财物，浪费公司资产，情节严重或造成损害（10.2.5 / 10.2.6）。 ### 4. 法律、健康与个人资质违规 * **刑事犯罪：** 被判处有期徒刑尚在服刑（4.8）；通缉在案（4.10）；实施违反国家法律法规的行为（如加入邪教、恐怖组织、酗酒、吸毒、赌博、嫖娼等）（4.14）。 * **权利状态：** 被剥夺公民权利者（4.9）。 * **年龄与健康：** 未满 16 周岁（4.11）；传染性疾病病原携带者或患有其他严重疾病（4.12）；患有精神病（4.13）。 * **债务问题：** 拖欠他人重大债务或进入法院执行程序未偿还债务者（4.16）。 * **过往记录：** 曾经被公司辞退或未经批准擅自离职者（4.7）。 ### 5. 试用期特殊解除条件 * **处分记录：** 试用期期间受记过及以上处分者（4）。 * **综合表现：** 上述考勤、诚信、法律等条款在试用期内触犯者（4.5 - 4.16）。 ### 6. 管理与连带责任 * **监管不力：** 管理者对员工违纪行为存在监管不力、失察、包庇、纵容或教唆等情形的，主管也将受到处罚（9.6）。 * **兜底条款：** 其他任何与前述违纪行为严重性相当的行为，或违反部门规定的行为，情节较重或造成一般损害的（5）。 **总结：** 根据手册第 10 条，若员工出现上述**严重违纪行为**，公司将对其进行**辞退（解除劳动合同）**。其中涉及诚信价值观问题、刑事犯罪、严重旷工等行为，公司有权立即解除合同且不支付经济补偿。 希望以上整理能帮助您准确理解员工手册中的相关规定。如有具体案例需要分析，欢迎随时提问。 ---------------------------------------- 耗时: 37.51 秒
import re import hashlib import xml.etree.ElementTree as ET from datetime import datetime from fastapi import FastAPI, Request, Response # RAG from langchain_huggingface import HuggingFaceEmbeddings from langchain_community.vectorstores import FAISS # LLM from langchain_ollama import OllamaLLM # LangGraph from langgraph.checkpoint.memory import InMemorySaver from langgraph.graph import StateGraph, END from typing import TypedDict from langchain_core.messages import HumanMessage, AIMessage import requests import time app = FastAPI(title="微信公众号 RAG Agent") # ===================== 配置 ===================== WECHAT_TOKEN = "wechat_rag_agent_2026" WECHAT_ORIGIN_ID = "gh_cfbff1593cc5" EMBEDDING_PATH = "./rag/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" FAISS_INDEX_DIR = "./rag/faiss_index" LLM_MODEL = "my-qwen" OLLAMA_HOST = "http://localhost:11434" CACHE_EXPIRE_SECONDS = 30 SESSION_TIMEOUT = 60 # ================================================== # ===================== 优化版欢迎语 ===================== WELCOME_TEXT = """✨ 你好，欢迎使用智能励志助手 在这里，你可以随时获取正能量与成长感悟～ 📌 使用指南 1️⃣ 回复【今天】或【每日一句】 获取每日一句精美图文，中英双语励志短句 2️⃣ 发送关键词触发知识库 支持：励志、奋斗、努力、学习、知识、坚持 成长、名言、失败、成功、气馁、失望、沮丧、希望 3️⃣ 直接发送文字聊天 AI 会为你简洁回复，陪你聊天解惑 愿你步履不停，始终向阳 🌱 """ # 会话记录 USER_SESSION = {} MESSAGE_CACHE = {} LAST_MSG_ID = None def clean_expired_cache(): now = time.time() expired = [k for k, v in MESSAGE_CACHE.items() if v["expire_at"] < now] for key in expired: del MESSAGE_CACHE[key] # ---------------------- RAG ---------------------- device = "cpu" embeddings = HuggingFaceEmbeddings( model_name=EMBEDDING_PATH, model_kwargs={"device": device} ) rag_db = FAISS.load_local( FAISS_INDEX_DIR, embeddings, allow_dangerous_deserialization=True ) retriever = rag_db.as_retriever(k=2) # ---------------------- LLM ---------------------- llm = OllamaLLM(model=LLM_MODEL, base_url=OLLAMA_HOST) # ---------------------- 工具函数 ---------------------- def is_today_sentence(text): t = text.strip().lower() return t in ["今天", "每日一句", "今日一句", "句子"] def get_iciba(): try: url = "https://open.iciba.com/dsapi/" resp = requests.get(url, timeout=3) data = resp.json() return { "title": f"📅 每日一句 | {data['dateline']}", "desc": f"{data['note']}", "pic": data["fenxiang_img"] } except Exception as e: print(f"金山接口异常: {e}") return None def is_motivation(text): keywords = ["励志", "奋斗", "努力", "学习", "知识", "坚持", "成长", "名言", "失败","成功","气馁","失望","沮丧","希望"] return any(key in text for key in keywords) # ---------------------- LangGraph ---------------------- class AgentState(TypedDict): user_input: str intent: str reply: str messages: list def classify_intent(state: AgentState): msg = state["user_input"] if is_today_sentence(msg): return {"intent": "sentence"} elif is_motivation(msg): return {"intent": "rag"} else: return {"intent": "chat"} # ===================== 最终美化版 RAG 展示 ===================== # def rag_node(state: AgentState): # import re # docs = retriever.invoke(state["user_input"]) # if not docs: # return {"reply": "💡 未找到相关内容，换个关键词试试～"} # # lines = ["🌿 精选励志名言\n"] # for d in docs: # text = d.page_content.strip() # eng = re.sub(r'[\u4e00-\u9fff]', '', text).strip() # chn = re.sub(r'[a-zA-Z\',.;!?\"\- ]', '', text).strip() # # lines.append(f"❝ {eng} ❞") # lines.append(f" {chn}\n") # # lines.append("📚 来源：本地 RAG 知识库") # return {"reply": "\n".join(lines)} def rag_node(state: AgentState): import re docs = retriever.invoke(state["user_input"]) if not docs: return {"reply": "未找到相关内容，换个关键词试试～"} lines = ["🌿 精选励志名言"] lines.append("") for idx, d in enumerate(docs, 1): text = d.page_content.strip() # 清理英文：删除中文+中文符号，只保留标准英文 eng = re.sub(r'[\u4e00-\u9fff]', '', text) eng = re.sub(r'[，。、；：！？…""''（）]', '', eng) eng = re.sub(r'\s+', ' ', eng).strip() eng = eng.rstrip('.!?') + '.' # 清理中文：删除英文+符号，补全句号 chn = re.sub(r'[a-zA-Z\'",;:!.?\-()\[\]]', '', text) chn = re.sub(r'\s+', ' ', chn).strip() chn = chn.rstrip('，。；！？') + '。' # 拼接内容（无多余空行） lines.append(f"{idx}. {eng}") lines.append(f"\t\t{chn}") lines.append("---") # 删除最后一条分隔线 if lines and lines[-1] == "---": lines.pop() lines.append("") lines.append("📍 来源：本地 RAG 知识库") return {"reply": "\n".join(lines)} def chat_node(state: AgentState): reply = llm.invoke(state["user_input"]) return {"reply": reply, "messages": [AIMessage(content=reply)]} memory = InMemorySaver() wf = StateGraph(AgentState) wf.add_node("cls", classify_intent) wf.add_node("rag", rag_node) wf.add_node("chat", chat_node) wf.set_entry_point("cls") wf.add_conditional_edges("cls", lambda s: s["intent"], { "sentence": "chat", "rag": "rag", "chat": "chat" }) wf.add_edge("rag", END) wf.add_edge("chat", END) agent = wf.compile(checkpointer=memory) # ---------------------- 微信接口 ---------------------- @app.get("/wechat") def wechat_verify(signature: str, timestamp: str, nonce: str, echostr: str): print(f"[GET] 微信接口验证") tmp = sorted([WECHAT_TOKEN, timestamp, nonce]) sha1 = hashlib.sha1("".join(tmp).encode()).hexdigest() return Response(content=echostr if sha1 == signature else "err") @app.post("/wechat") async def wechat_msg(request: Request): global LAST_MSG_ID try: xml_body = await request.body() print(f"\n========== 收到微信请求 ==========") root = ET.fromstring(xml_body) msg_type = root.find("MsgType").text openid = root.find("FromUserName").text to_user = root.find("ToUserName").text print(f"消息类型: {msg_type}, 用户OPENID: {openid}") if msg_type != "text": print("非文本消息，直接返回") return Response("") content = root.find("Content").text.strip() msg_id = root.find("MsgId").text.strip() now = time.time() clean_expired_cache() print(f"用户消息: {content}, 消息ID: {msg_id}") first_message = False if openid not in USER_SESSION or now - USER_SESSION[openid] > SESSION_TIMEOUT: first_message = True print("✅ 新会话，标记为首次消息") USER_SESSION[openid] = now # 每日一句 if is_today_sentence(content): data = get_iciba() if data: print(f"✅ 调用金山接口成功，返回图文") reply_xml = f"""<xml> <ToUserName> {openid} </ToUserName> <FromUserName> {to_user} </FromUserName> <CreateTime>{int(now)}</CreateTime> <MsgType> news </MsgType> <ArticleCount>1</ArticleCount> <Articles> <item> <Title> {data['title']} <Description> {data['desc']} </Description> <PicUrl> {data['pic']} </PicUrl> <Url> </Url> </item> </Articles> </xml>""" return Response(content=reply_xml, media_type="application/xml") # 正常 AI 逻辑 if msg_id != LAST_MSG_ID: MESSAGE_CACHE.clear() LAST_MSG_ID = msg_id print(f"✅ 新消息ID，清空缓存") if msg_id in MESSAGE_CACHE and MESSAGE_CACHE[msg_id]["expire_at"] > now: reply = MESSAGE_CACHE[msg_id]["reply"] print(f"✅ 命中缓存，直接回复: {reply}") else: config = {"configurable": {"thread_id": openid}} result = agent.invoke({ "user_input": content, "messages": [HumanMessage(content=content)] }, config=config) reply = result["reply"] MESSAGE_CACHE[msg_id] = { "reply": reply, "expire_at": now + CACHE_EXPIRE_SECONDS } print(f"✅ AI生成回复: {reply}") if first_message: print(f"✅ 首次消息，拼接欢迎语") reply = WELCOME_TEXT + reply reply_xml = f"""<xml> <ToUserName> {openid} </ToUserName> <FromUserName> {to_user} </FromUserName> <CreateTime>{int(now)}</CreateTime> <MsgType> text </MsgType> <Content> {reply} </Content> </xml>""" print(f"✅ 最终回复XML: {reply_xml}") return Response(content=reply_xml, media_type="application/xml") except Exception as e: print(f"❌ 全局异常: {str(e)}") return Response("") @app.get("/") def index(): return {"status": "running"}
微信公众号 RAG Agent 最终版 GitHub 文档（已补全：项目目录、requirements.txt、Modelfile、Nginx反向代理、微信配置全流程、完整启动命令）🚀 项目介绍本项目是一个专为微信公众号设计的轻量智能对话Agent，基于 FastAPI + LangGraph + Ollama + FAISS 构建，支持：大模型闲聊（一句话短回答，适配微信场景）RAG 知识库检索日期类接口调用（每日一句）微信重试缓存机制：彻底解决公众号5秒超时问题，无需客服消息，100%保证消息送达Nginx 反向代理，适配公网部署核心优势：2核4G服务器即可稳定运行，纯CPU推理，无GPU依赖，零客服消息次数占用。📁 项目目录结构wechat_agent_ai/ ├── rag/ # RAG 知识库模块 │ ├── faiss_index/ # FAISS 向量索引文件（向量化后生成） │ ├── models--Qwen--Qwen3-Embedding-0.6B/ # 通义千问嵌入模型 │ ├── data_vectorization.py # 知识库向量化脚本（生成FAISS索引） │ ├── iciba_data_multi.json # 每日一句备用数据 │ └── rag_data_retrieval.py # RAG 检索测试脚本 ├── main.py # 主程序（微信接口+Agent+重试缓存核心逻辑） ├── main_1.py # 历史版本/备用代码 ├── test_local.py # 本地测试脚本（无需微信，直接测试Agent功能） ├── requirements.txt # 项目依赖清单（一键安装） ├── Modelfile # Ollama 自定义模型配置文件 ├── dockerfile # Docker 构建文件（可选） ├── start.sh # 启动脚本 ├── .dockerignore # Docker 忽略文件 └── README.md # 本说明文档目录说明文件/目录作用rag/RAG 知识库相关代码、模型、向量索引，用于知识库问答main.py项目核心主程序，包含微信公众号接口、LangGraph 工作流、重试缓存逻辑test_local.py本地测试脚本，可直接运行验证Agent功能，无需配置微信公众号requirements.txt项目所有Python依赖包，一键安装环境ModelfileOllama 自定义模型配置，用于构建适配微信的短回答模型 my-qwen📦 环境安装（通过 requirements.txt 一键部署）1. 前置依赖系统：Ubuntu 20.04+（推荐）Python：3.8+服务器：2核4G及以上，开放80端口（Nginx反向代理用）2. 安装 Ollama（大模型运行环境）# 一键安装 Ollama curl https://ollama.com/install.sh | sh # 拉取基础模型（qwen2:0.5b 超轻量，适配2核4G） ollama pull qwen2:0.5b-instruct-q2_K3. 构建自定义 Ollama 模型（Modelfile）# 创建 Modelfile（适配微信的短回答模型） tee Modelfile <<EOF FROM qwen2:0.5b-instruct-q2_K PARAMETER num_ctx 512 PARAMETER num_predict 64 PARAMETER temperature 0.1 PARAMETER top_p 0.7 PARAMETER num_thread 2 SYSTEM "直接回答，一句话。" EOF # 构建自定义模型 my-qwen ollama create my-qwen -f Modelfile # 测试模型（验证是否正常） ollama run my-qwen "你好"4. 一键安装 Python 依赖（requirements.txt）# 1. 创建并激活虚拟环境（推荐） python3 -m venv venv source venv/bin/activate # 2. 一键安装所有依赖 pip install -r requirements.txtrequirements.txt 完整内容fastapi==0.110.0 uvicorn==0.28.0 langchain==0.2.0 langgraph==0.0.56 langchain-ollama==0.1.0 langchain-huggingface==0.1.0 langchain-community==0.2.0 faiss-cpu==1.8.0 requests==2.31.0 python-dotenv==1.0.1 huggingface-hub==0.23.0⚙️ 核心配置说明1. main.py 核心配置修改 main.py 中的「线上配置」部分，替换为自己的信息：# ===================== 【线上配置】 ===================== WECHAT_TOKEN = "wechat_rag_agent_2026" # 与微信公众号配置的Token完全一致 WECHAT_ORIGIN_ID = "gh_cfbff1593cc5" # 公众号原始ID（后台「设置与开发」→「基本配置」查看） EMBEDDING_PATH = "./rag/models--Qwen--Qwen3-Embedding-0.6B/snapshots/c54f2e6e80b2d7b7de06f51cec4959f6b3e03418" FAISS_INDEX_DIR = "./rag/faiss_index" # FAISS 向量索引目录 LLM_MODEL = "my-qwen" # 自定义Ollama模型名称（与Modelfile构建的一致） OLLAMA_HOST = "http://localhost:11434" # 缓存过期时间：30秒（仅覆盖微信重试窗口期，超时自动清空） CACHE_EXPIRE_SECONDS = 30 # ========================================================2. Nginx 反向代理配置（关键！适配微信公网访问）1. 安装 Nginxsudo apt update && sudo apt install nginx -y2. 创建反向代理配置文件sudo tee /etc/nginx/sites-available/wechat_agent <<EOF server { listen 80; server_name 你的服务器公网IP; # 替换为你的服务器IP（与微信配置的URL一致） # 👇 微信公众号接口反向代理（核心配置） location /wechat { proxy_pass http://127.0.0.1:8080/wechat; proxy_set_header Host \$host; proxy_set_header X-Real-IP \$remote_addr; proxy_set_header X-Forwarded-For \$proxy_add_x_forwarded_for; proxy_read_timeout 60s; # 适配微信超时，避免代理断开 } } EOF3. 启用配置并重启 Nginx# 启用配置 sudo ln -s /etc/nginx/sites-available/wechat_agent /etc/nginx/sites-enabled/ # 校验配置是否正确 sudo nginx -t # 重启 Nginx 生效 sudo systemctl restart nginx🚀 完整服务启动命令（直接复制运行）以下命令包含「关闭旧进程→启动Ollama→预加载模型→启动FastAPI」，一键完成服务部署，确保运行稳定：source venv/bin/activate pkill -9 ollama pkill -9 uvicorn fuser -k 8080/tcp OLLAMA_CPU_THREADS=2 OLLAMA_LOAD_MODEL_AT_START=true nohup ollama serve > ollama.log 2>&1 & ollama run my-qwen "/bye" nohup uvicorn main:app --host 0.0.0.0 --port 8080 --workers 1 > wechat.log 2>&1 命令说明：前3行：关闭旧的Ollama、uvicorn进程，释放8080端口，避免端口占用冲突第4行：2核满负载启动Ollama，启动时预加载模型，彻底消除冷启动耗时第5行：预加载my-qwen模型，进一步加速第一次请求响应第6行：启动FastAPI服务，监听所有IP（0.0.0.0），适配Nginx反向代理，1个worker避免资源竞争查看服务状态（确认启动成功）：ps -ef | grep ollama ps -ef | grep uvicorn📱 微信公众号配置（与截图完全对应）1. 配置信息（后台「微信开发者平台」→「公众号」->「消息」）配置项填写内容URLhttp://你的服务器公网IP/wechat（与Nginx配置一致，示例：http://101.42.xxx.xxx/wechat）Tokenwechat_rag_agent_2026（与main.py配置完全一致）EncodingAESKey随机生成（无需修改，明文模式下不影响）消息加密方式明文模式（推荐，避免加解密问题）数据格式XML（与代码适配）2. 配置校验提交配置后，微信会自动校验URL有效性，校验通过后即可启用，公众号即可正常接收用户消息并自动回复。🧠 核心逻辑说明1. 重试缓存机制（解决微信5秒超时）微信公众号被动回复官方建议超时5秒，超时会自动重试同一条消息（MsgId 不变）第一次请求：调用AI生成回复，写入内存缓存（有效期30秒，仅绑定当前MsgId）微信重试：第二次请求命中缓存，0.2秒内直接返回，确保用户100%收到新消息触发：MsgId 变化时，自动清空所有缓存，避免重复回答过期清理：每次请求自动清理30秒以上的过期缓存，无内存泄漏2. 意图识别路由（LangGraph 工作流）用户输入类型路由节点处理逻辑日期格式（YYYY-MM-DD）date调用金山词霸接口，返回每日一句含「励志/奋斗/名言」等关键词rag检索FAISS知识库，返回相关内容其他闲聊内容chat调用my-qwen大模型，一句话短回答📝 日志查看与问题排查1. 日志查看命令# 查看FastAPI/微信消息日志（核心日志） tail -f wechat.log # 查看Ollama模型日志 tail -f ollama.log # 查看Nginx代理日志 tail -f /var/log/nginx/access.log2. 常见问题排查（1）微信消息收不到检查服务器安全组，确认80端口已开放（Nginx监听80，微信仅支持80/443端口）确认Nginx配置正确，proxy_pass 指向 127.0.0.1:8080/wechat查看wechat.log，确认AI调用、缓存逻辑正常检查微信Token与代码配置完全一致（2）回复速度慢确认Ollama启动时添加OLLAMA_CPU_THREADS=2，2核满负载运行确认Modelfile中num_thread=2，充分利用CPU执行预加载模型命令（启动命令第5行），消除冷启动耗时（3）重复回答查看日志中「缓存清理」记录，确认新消息MsgId变化时缓存已清空缓存有效期30秒，超时自动清理，重启FastAPI可手动清空所有缓存🛑 服务停止命令# 停止Ollama和FastAPI服务 pkill -9 ollama uvicorn # 释放8080端口 fuser -k 8080/tcp # 停止Nginx（如需） sudo systemctl stop nginx📌 注意事项服务器重启后，需重新执行「完整服务启动命令」（复制即可运行），可配置开机自启避免手动操作Ollama模型建议使用0.5B级别的量化模型（如qwen2:0.5b-instruct-q2_K），避免CPU负载过高导致卡顿微信公众号仅支持80/443端口，必须通过Nginx反向代理8080端口，否则无法完成URL校验缓存仅用于接住微信重试，有效期30秒，无内存占用风险，无需手动清理正式环境建议配置HTTPS（Let's Encrypt免费证书），使用443端口，提升安全性并避免微信接口限制启动命令中--host 0.0.0.0表示监听所有公网IP，配合Nginx反向代理可实现公网访问，无需修改其他配置若出现端口占用，可再次执行fuser -k 8080/tcp释放端口，重新启动服务