我们把「拉取消息」和「回复」这两件事塞进同一个 while 循环，让机器人保持“在线”。下面这段代码几乎Ke以直接拷贝跑通：

def run_monitor -> None:
    """长轮询监听循环"""
    buf = BUF_FILE.read_text.strip if BUF_FILE.exists else ""
    if buf:
        print
    print
    fail_count = 0
    while True:
        try:
            resp = getUpdates
        except Exception as e:
            fail_count += 1
            print: {e}", flush=True)
            if fail_count>= 5:
                print
                time.sleep
                fail_count = 0
            else:
                time.sleep
            continue
        fail_count = 0
        if resp.get != 0 or resp.get != 0:
            print
            time.sleep
            continue
        new_buf = resp.get
        if new_buf:
            buf = new_buf
            BUF_FILE.write_text
        for msg in resp.get or :
            # 忽略机器人自己发出的回执消息，只处理用户发来的文本
            if msg.get != 100:   # 假设100是用户发来的普通文本
                continue
            from_user     = msg.get
            context_token = msg.get
            # 提取文本内容
            text = ""
            for item in msg.get or :
                if item.get == 100:       # 同样假设100代表文本类型
                    text =  or {}).get
                    break
            if not text.strip:
                continue     # 空白信息直接跳过
           # ---------- 到这里我们Yi经拿到了用户说的话 ---------- 
           # 接下来交给本地 AI Agent ----------------------------------------------------
           reply = askAgent          # ← 核心业务函数
           try:
               sendMessage
               print
           except Exception as e:
               print

每个用户dou应该拥有独立的对话历史，否则大家会抢着kan别人的记录。下面这段代码兼顾了多用户隔离与异常捕获：

# ── 导入本地 Agent ──
from agent import agent_loop, SYSTEM as AGENT_SYSTEM
_sessions: dict] = {}
def askAgent -> str:
    """
    把用户的新消息喂给本地 LLM，返回生成的回答。
      • 第一次对话时自动加上系统提示词。
      • 每次调用dou会把 Assistant 的回复追加到历史中，实现连续对话。
      • 若模型抛异常，会返回友好的错误提示。
    """
    if user_id not in _sessions:
        _sessions = 
    _sessions.append
    try:
        answer = agent_loop
        return answer or ""
    except Exception as exc:
        return f" {exc}"

💡 小技巧：Ru果你的模型支持流式输出，Ke以 成异步生成，把每一段文字即时推送回微信，实现“打字机”效果。

def sendMessage(token: str,
                to_user_id: str,
                text: str,
                context_token: str) -> None:
    """向指定用户发送纯文本消息，context_token 必须原样回传"""
    _post(
        "ilink/bot/sendmessage",
        token=token,
        body={
            "msg":{
                "from_user_id":"",                 # bot 发出留空即可
                "to_user_id":to_user_id,
                "client_id":f"mini-{secrets.token_hex}",
                "message_type":100,                  # 文本类型编号
                "message_state":200,                 # Yi完成，不走流式模式
                "item_list":,
                "context_token":context_token,
            },
            "base_info":{"channel_version":"mini-bridge-"},
        },
        timeout=15,
    )

这里我们硬编码了 message_type 与 message_state 的数值，只要保持一致即可；若以后官方改动，只需geng新这两个常量。

def _headers -> dict:
    """构造所有请求必须携带的 Header"""
    h={"Content-Type":"application/json",
         "AuthorizationType":"ilink_bot_token",
         "X-WECHAT-UIN":base64.b64encode(
             str)).encode
         ).decode}
    if token:
         h=f"Bearer {token}"
    return h
def _get->dict:
    req=urllib.request.Request(url,
                           method='GET',
                           headers=headers or {})
    with urllib.request.urlopen as rsp:
         return json.loads.decode)
def _post(path:str,
          body:dict,
          token:Optional=None,
          timeout:int=15)->dict:
    data=json.dumps.encode
    req=urllib.request.Request,
                           data=data,
                           headers={**_headers,
                                 'Content-Length':str)},
                           method='POST')
    with urllib.request.urlopen as rsp:
         return json.loads.decode)
def _url->str:
    base=BASE_URL.rstrip+'/'
    return base+path

def main:
    """程序入口，仅Zuo两件事：
         · 登录
         · 启动长轮询监听"""
    creds=load_token
    if not creds:                         # 没有缓存则走扫码流程
        print
        creds=login
        save_token
        print
    token      = creds
    account_id  = creds
    base_url   = creds.get
    print
    run_monitor
if __name__=="__main__":
    main

kan完上述代码，你可Neng会惊讶：仅仅几百行，就完成了一个Neng够在真实微信群里聊天、读取文件甚至调用自定义插件的智Neng体。相比市面上那些需要第三方 SaaS 平台搭桥的方案，这套方案Zui大亮点在于：

完全自托管：所有请求dou直达腾讯官方接口，没有额外代理层。

随时可调：只要换掉 agent_loop 的实现，就Neng切换不同的大模型。

持久化安全：Token 与游标分别存文件，一次登录后即可长期运行，无需频繁扫码。

Poor‑Man‑Scalability：即使只有单核机器，也Neng保持每分钟数十条对话不掉线，因为核心dou是阻塞 I/O + 简单重连逻辑。

💬 小结一句话：只要懂得调用 HTTP 接口和维护上下文，你就Ke以让任何「AI 大脑」跑进微信群，而不必受限于公众号或小程序的审查。