使用dify构建微信自动回复Agent-开发者社区

前言

使用dify、ollama构建微信自动回复。

一、环境介绍

1、微信版本：3.9

2、腾讯服务器4核、8G：安装dify

3、AutoDL：安装ollama、安装qwen2.5:1.5b

4、AutoDL：安装Xinference、bge-large-zh-v1.5、bge-reranker-base

二、dify连接AutoDL安装的大模型

点击右上角我的-设置

在模型供应商-安装大模型供应商下面找到Ollama-点击安装

Ollama配置大模型qwen2.5:1.5b：在ollama右边点击添加模型

三、dify配置bge-large-zh-v1.5、bge-reranker-base

在模型供应商-安装大模型供应商下面找到Xorbits Inference-点击安装

配置：bge-large-zh-v1.5

配置：bge-reranker-base

四、dify创建知识库

1、在AUTODL服务器安装：bge-large-zh-v1.5、bge-reranker-base

2、DIFY创建知识库：知识库-创建知识库-创建空白知识库

3、上次知识内容后，选择：bge-large-zh-v1.5、bge-reranker-base

4、等待知识库创建成功(所有文档右边都打上绿色的勾)

五、dify创建微信回复Agent

1、创建Agent

2、配置Agent

六、微信信息读取与调用Agent回复

1、获取difyAPI

2、配置KEY

3、新建配置文件：.env.wechat

DIFY_BASE_URL = "" DIFY_API_KEY = "" TARGET_CHAT_NAME = "" SELF_SENDER = ""

4、新建header:chat_dify_agent_headers.py

from dotenv import load_dotenv import os load_dotenv(".env.wechat") dify_API_KEY = os.getenv("DIFY_API_KEY") headers = { "Authorization": f"Bearer {dify_API_KEY}", "Content-Type": "application/json" }

5、新建去掉think内容：chat_dify_agent_tags.py

import re def remove_think_tags(text): """ 去除文本中的 <think>...</think> 标签及内容 """ # .*? 表示非贪婪匹配，防止跨越多个标签匹配 pattern = r'<think>.*?</think>' return re.sub(pattern, '', text, flags=re.DOTALL)

6、新建调用dify-Agent:chat_dify_agent.py

import requests import json from dotenv import load_dotenv import os from chat_dify_agent_headers import headers from chat_dify_agent_tags import remove_think_tags load_dotenv(".env.wechat") DIFY_BASE_URL = os.getenv("DIFY_BASE_URL") def dify_agent(query: str, conversation_id: str = None): payload = { "inputs": {}, "query": query, "response_mode": "streaming", "user": "python_user001", } if conversation_id: payload["conversation_id"] = conversation_id resp = requests.post( url=f"{DIFY_BASE_URL}/chat-messages", headers=headers, json=payload, stream=True, timeout=180 ) resp.raise_for_status() answer = "" conv_id = None for line in resp.iter_lines(): if not line: continue raw_text = line.decode("utf-8").removeprefix("data: ") if raw_text.strip() == "[DONE]": break try: event_data = json.loads(raw_text) # 获取会话id if not conv_id and event_data.get("conversation_id"): conv_id = event_data["conversation_id"] # Agent‑V2重点：agent_thought事件保存工具执行结果、回答 if event_data.get("event") == "agent_thought": thought = event_data.get("thought", "") tool = event_data.get("tool", "") tool_input = event_data.get("tool_input","") observation = event_data.get("observation","") # print(f"\n[Agent思考] {thought}") if tool: print(f"[调用工具] {tool} 参数={tool_input} 输出={observation}") answer += thought # 普通message事件 if event_data.get("event") == "message": chunk = event_data.get("answer","") answer += chunk print(chunk, end="") except Exception as e: continue clean_answer = remove_think_tags(answer) print(f"\n[Dify输出] {clean_answer}") return clean_answer, conv_id

7、新建调用微信信息读取回复:we_chat.py

from wxauto import WeChat from dotenv import load_dotenv import os import time from wechat.chat_dify_agent import dify_agent load_dotenv(".env.wechat") TARGET_CHAT_NAME = os.getenv("TARGET_CHAT_NAME") def we_chat_agent(): wx = WeChat() # 2. 先打开聊天窗口，这是后续操作的基础 wx.ChatWith(TARGET_CHAT_NAME) time.sleep(1.5) # 等待窗口加载 # 3. 添加监听 wx.AddListenChat(who=TARGET_CHAT_NAME, savepic=False) # --- 核心修改：彻底清空历史消息 --- print("🔄 正在丢弃启动瞬间的历史消息...") # 使用一个循环，持续获取消息直到返回为空，确保历史消息被完全清空 while wx.GetListenMessage(): time.sleep(0.1) print("✅ 历史消息已清空，开始监听新消息...") # ----------------------------------- print(f"✅ 微信机器人启动，监听好友：{TARGET_CHAT_NAME}") try: while True: # 【修复点】：在每次循环开始时，先初始化一个空的 reply_text 变量 reply_text = "" msgs_dict = wx.GetListenMessage() # 没有新消息，直接跳过本次循环 if not msgs_dict: time.sleep(0.8) continue # 处理新消息 for chat in msgs_dict: who = chat.who # 只处理目标联系人的消息 if who != TARGET_CHAT_NAME: continue one_msgs = msgs_dict.get(chat) for msg in one_msgs: content = msg.content sender = getattr(msg, 'sender', '未知发送者') print(f"发送者={sender}") if sender == "Self": continue msg_type = msg.type # 过滤掉时间消息、图片、视频等非文本内容 if msg_type == "time" or content.strip() in {"[图片]", "[视频]", "[位置]", "[语音]", "[以下为新消息]"}: continue # 过滤掉空消息 if not content: continue # 【核心修复】：如果收到的内容就是机器人刚刚发出去的回复，直接跳过，防止死循环 # 注意：这里用 strip() 去除首尾换行和空格后再比较，更加严谨 if content.strip() == reply_text.strip(): print(f"⚠️ 检测到自身发出的消息，已忽略：{content[:20]}...") continue print(f"\n📩 收到新消息【{who}】：{content}") # 调用Dify并回复 reply_text, conversation_id = dify_agent(content, conversation_id) print(f"🤖 Agent回复：{reply_text}") if reply_text: try: chat.SendMsg(reply_text) print("✅ 微信消息发送完成") except Exception as send_err: print(f"❌ 发送微信消息失败：{send_err}") # 短暂休眠，避免CPU占用过高 time.sleep(0.8) except KeyboardInterrupt: print("\n🛑 程序手动停止") except Exception as e: print(f"\n[程序异常] {e}")