HermesGateway 微信集成：24 小时在线编程助手

┌─────────────┐     ┌──────────────────┐     ┌─────────────────┐
│   微信端     │────▶│  HermesGateway    │────▶│   Claude Code    │
│  接收/发送   │◀────│  (智能路由中枢)    │◀────│   (代码执行引擎)  │
└─────────────┘     └──────────────────┘     └─────────────────┘
                           │
                    ┌──────┴──────┐
                    ▼             ▼
              ┌──────────┐  ┌──────────┐
              │ 记忆存储  │  │ 工具注册  │
              │ (Redis)  │  │ 中心     │
              └──────────┘  └──────────┘

用户微信消息
  → 微信Hook接收
  → HermesGateway解析意图
  → 路由决策（编码任务? 查询? 闲聊?）
  → Claude Code执行代码
  → 结果格式化
  → 微信回复

# config/gateway.yaml
gateway:
  name: "WeChat-Coding-Assistant"
  version: "1.0.0"

  # 入口配置 - 微信消息接收
  entrypoints:
    wechat:
      type: "webhook"
      port: 8080
      auth_token: "${WECHAT_WEBHOOK_TOKEN}"
      media_timeout: 30s

  # 路由规则
  routing:
    default_provider: "hermes-llama"
    rules:
      - pattern: "^(帮我写|写一个|实现|修复bug|重构)"
        route_to: "claude-coder"
        priority: "high"

      - pattern: "^(解释|是什么|怎么用)"
        route_to: "hermes-llama"
        priority: "medium"

      - pattern: "^(查|搜索|文档)"
        route_to: "search-agent"
        priority: "low"

      - pattern: ".*"
        route_to: "hermes-llama"
        priority: "fallback"

  # Agent 注册
  agents:
    claude-coder:
      type: "claude-code"
      model: "claude-sonnet-4-20250514"
      max_tokens: 8192
      sandbox: true
      timeout: 300s

    hermes-llama:
      type: "hermes-native"
      model: "hermes-3-llama-3.1-70b"
      endpoint: "http://localhost:8000/v1"

    search-agent:
      type: "tool-agent"
      tools: ["web_search", "doc_lookup"]

# src/wechat_bridge.py
from wechaty import Wechaty, Message, Room
from hermes_gateway import GatewayClient, RouteResult
import asyncio
import logging

logger = logging.getLogger("wechat-bridge")

class WeChatBridge:
    """微信与 HermesGateway 的桥接层"""

    def __init__(self, gateway_url: str = "http://localhost:8080"):
        self.gateway = GatewayClient(gateway_url)
        self.pending_tasks = {}  # 跟踪异步任务
        self.session_context = {}  # 用户会话上下文

    async def on_message(self, msg: Message):
        """处理接收到的微信消息"""
        if msg.is_self():
            return  # 忽略自己发送的消息

        sender = msg.talker()
        content = msg.text()
        room = msg.room()

        logger.info(f"收到消息 from {sender.name}: {content}")

        # 构建请求
        request = {
            "user_id": sender.contact_id,
            "user_name": sender.name,
            "content": content,
            "message_type": "text",
            "room_id": room.room_id if room else None,
            "session_key": self._build_session_key(sender, room),
        }

        # 发送到 Gateway 处理
        task = asyncio.create_task(self._process_message(request, msg))
        self.pending_tasks[sender.contact_id] = task

    async def _process_message(self, request: dict, original_msg: Message):
        """通过 Gateway 处理消息并回复"""
        try:
            # 1. 获取会话历史
            history = self._get_session_history(request["session_key"])
            request["history"] = history

            # 2. 发送到 HermesGateway
            response = await self.gateway.route(request)

            # 3. 根据响应类型发送不同格式的消息
            await self._send_response(original_msg, response)

            # 4. 更新会话历史
            self._update_session_history(
                request["session_key"],
                request["content"],
                response["reply"]
            )

        except Exception as e:
            logger.error(f"处理消息失败: {e}")
            await original_msg.talker().say(f"⚠️ 处理出错: {str(e)}")

    async def _send_response(self, msg: Message, response: dict):
        """智能发送响应 - 根据内容类型选择消息格式"""
        reply = response.get("reply", "")
        response_type = response.get("type", "text")

        if response_type == "code_result":
            # 代码结果分块发送
            chunks = self._split_code_response(reply)
            for chunk in chunks:
                await msg.talker().say(chunk)
                await asyncio.sleep(0.5)  # 避免频率限制

        elif response_type == "file":
            # 发送文件
            file_path = response.get("file_path")
            await msg.talker().say(FileBox.from_file(file_path))

        else:
            # 普通文本
            for chunk in self._split_long_text(reply, 1800):
                await msg.talker().say(chunk)

    def _build_session_key(self, sender, room) -> str:
        """构建会话标识"""
        if room:
            return f"room:{room.room_id}:{sender.contact_id}"
        return f"private:{sender.contact_id}"

# src/middleware/wechat_middleware.py
from typing import Dict, Any, Callable
import time
import re

class WeChatMiddleware:
    """微信消息处理中间件"""

    def __init__(self):
        self.rate_limits: Dict[str, list] = {}
        self.max_requests_per_minute = 30
        self.max_message_length = 4000

    def process_incoming(self, message: Dict[str, Any]) -> Dict[str, Any]:
        """预处理入站消息"""
        # 1. 内容清洗
        message["content"] = self._clean_content(message["content"])

        # 2. 长度限制
        if len(message["content"]) > self.max_message_length:
            message["content"] = message["content"][:self.max_message_length]
            message["truncated"] = True

        # 3. 类型检测
        message["intent_hint"] = self._detect_intent(message["content"])

        return message

    def check_rate_limit(self, user_id: str) -> bool:
        """检查用户请求频率"""
        now = time.time()
        if user_id not in self.rate_limits:
            self.rate_limits[user_id] = []

        # 清理60秒前的记录
        self.rate_limits[user_id] = [
            t for t in self.rate_limits[user_id]
            if now - t < 60
        ]

        if len(self.rate_limits[user_id]) >= self.max_requests_per_minute:
            return False

        self.rate_limits[user_id].append(now)
        return True

    def _clean_content(self, content: str) -> str:
        """清洗消息内容"""
        # 去除多余空白
        content = re.sub(r'\s+', ' ', content.strip())
        # 去除微信特有表情编码
        content = re.sub(r'\[.*?\]', '', content)
        return content

    def _detect_intent(self, content: str) -> str:
        """粗略意图检测，辅助路由"""
        coding_keywords = ["写", "代码", "函数", "类", "bug", "报错", "异常",
                          "实现", "修复", "优化", "重构", "测试"]
        query_keywords = ["什么", "怎么", "为什么", "解释", "文档", "用法"]

        if any(kw in content for kw in coding_keywords):
            return "coding"
        elif any(kw in content for kw in query_keywords):
            return "query"
        return "chat"

# config/agents/claude-coder.yaml
agent:
  name: "claude-coder"
  type: "claude-code"

  # 模型配置
  model:
    name: "claude-sonnet-4-20250514"
    max_tokens: 8192
    temperature: 0.1  # 编码任务使用低温度保证确定性

  # 沙箱配置
  sandbox:
    enabled: true
    allowed_dirs:
      - "/tmp/workspace"
      - "/opt/data/projects"
    blocked_commands:
      - "rm -rf /"
      - "mkfs"
      - "dd"
      - "curl.*\\|.*sh"
    network_access: false  # 默认禁止外网访问

  # 工具权限
  tools:
    - "file_read"
    - "file_write"
    - "file_edit"
    - "bash_execute"
    - "grep_search"
    - "glob_search"

  # 系统提示词
  system_prompt: |
    你是一个专业的编程助手，通过微信与用户交互。

    规则：
    1. 用中文回复，代码注释也用中文
    2. 先理解需求，再给出解决方案
    3. 代码要完整可运行，包含必要注释
    4. 修改代码时说明改了什么、为什么改
    5. 运行测试验证结果
    6. 回复简洁，代码外的解释不超过3句话

    输出格式：
    - 先给出简要说明
    - 然后给出代码（用代码块）
    - 最后说明如何使用/测试

# src/agents/claude_executor.py
import asyncio
import json
import tempfile
from pathlib import Path
from typing import Optional, Dict, Any

class ClaudeCodeExecutor:
    """Claude Code 任务执行器"""

    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.workspace = Path(config.get("workspace", "/tmp/workspace"))
        self.workspace.mkdir(parents=True, exist_ok=True)

    async def execute_coding_task(self, task: str, context: str = "") -> Dict[str, Any]:
        """执行编码任务"""

        # 1. 创建工作空间
        task_dir = self.workspace / f"task_{id(task)}"
        task_dir.mkdir()

        # 2. 构建 Claude Code 提示
        prompt = self._build_prompt(task, context)

        # 3. 调用 Claude Code
        result = await self._run_claude_code(prompt, task_dir)

        # 4. 验证结果
        validation = await self._validate_result(result, task_dir)

        # 5. 格式化回复
        reply = self._format_reply(result, validation)

        # 6. 清理
        self._cleanup(task_dir)

        return {
            "reply": reply,
            "type": "code_result" if result.get("code") else "text",
            "status": "success" if validation.get("passed") else "partial",
            "files": result.get("files", []),
        }

    def _build_prompt(self, task: str, context: str) -> str:
        """构建 Claude Code 提示词"""
        prompt = f"""请完成以下编程任务：

任务描述：
{task}

"""
        if context:
            prompt += f"""
相关上下文：
{context}

"""
        prompt += """
请按以下步骤执行：
1. 分析需求
2. 编写代码（保存到当前目录）
3. 运行测试验证
4. 输出结果总结

开始执行。"""
        return prompt

    async def _run_claude_code(self, prompt: str, work_dir: Path) -> Dict[str, Any]:
        """调用 Claude Code CLI 执行任务"""
        cmd = [
            "claude", "-p", prompt,
            "--allowedTools", "Bash,Read,Write,Edit,Grep,Glob",
            "--outputFormat", "json"
        ]

        proc = await asyncio.create_subprocess_exec(
            *cmd,
            cwd=str(work_dir),
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.PIPE,
        )

        stdout, stderr = await proc.communicate()

        return {
            "stdout": stdout.decode("utf-8"),
            "stderr": stderr.decode("utf-8"),
            "returncode": proc.returncode,
            "code": True if proc.returncode == 0 else False,
            "files": [f.name for f in work_dir.iterdir() if f.is_file()],
        }

    async def _validate_result(self, result: Dict, work_dir: Path) -> Dict:
        """验证执行结果"""
        # 检查是否有测试文件
        test_files = list(work_dir.glob("test_*")) + list(work_dir.glob("*_test.*"))

        if test_files:
            # 运行测试
            proc = await asyncio.create_subprocess_exec(
                "python", "-m", "pytest", str(work_dir), "-v",
                stdout=asyncio.subprocess.PIPE,
                stderr=asyncio.subprocess.PIPE,
            )
            stdout, _ = await proc.communicate()

            return {
                "passed": proc.returncode == 0,
                "test_output": stdout.decode("utf-8")[-500:],  # 最近500字符
            }

        return {"passed": None, "note": "无测试文件"}

    def _format_reply(self, result: Dict, validation: Dict) -> str:
        """格式化微信回复内容"""
        reply_parts = []

        # 提取关键信息
        stdout = result.get("stdout", "")

        # 查找代码块
        import re
        code_blocks = re.findall(r'```[\w]*\n(.*?)```', stdout, re.DOTALL)

        if code_blocks:
            reply_parts.append("✅ 代码已完成：\n")
            for i, code in enumerate(code_blocks[:3]):  # 最多3个代码块
                lang = "python"
                reply_parts.append(f"```{lang}\n{code.strip()[:800]}\n```")
        else:
            # 纯文本回复
            reply_parts.append(stdout[:1500])

        if validation.get("passed") is True:
            reply_parts.append("\n✅ 测试全部通过")
        elif validation.get("passed") is False:
            reply_parts.append("\n❌ 部分测试未通过")

        return "\n".join(reply_parts)

#!/bin/bash
# scripts/start-wechat-assistant.sh

echo "🚀 启动微信编程助手..."

# 1. 启动 HermesGateway
hermes-gateway start --config config/gateway.yaml &
GATEWAY_PID=$!

# 2. 等待 Gateway 就绪
echo "⏳ 等待 Gateway 启动..."
while ! curl -sf http://localhost:8080/health > /dev/null; do
    sleep 1
done
echo "✅ Gateway 已就绪"

# 3. 启动微信桥接服务
python src/main.py --gateway-url http://localhost:8080 &
BRIDGE_PID=$!

echo "✅ 微信编程助手已启动"
echo "   Gateway PID: $GATEWAY_PID"
echo "   Bridge PID:  $BRIDGE_PID"

# 4. 注册信号处理
trap "kill $GATEWAY_PID $BRIDGE_PID; exit" SIGINT SIGTERM

# 5. 保持运行
wait

用户: 帮我写一个 Python 函数，实现斐波那契数列，要求用 memoization 优化

→ 微信桥接收消息
→ 中间件清洗 + 意图检测 → intent: "coding"
→ Gateway 匹配路由规则 → route_to: "claude-coder"
→ Claude Code 执行:
   - 分析需求
   - 编写带 memoization 的 fib 函数
   - 编写测试用例
   - 运行 pytest 验证
→ 结果格式化 → 微信回复:

助手: ✅ 代码已完成：

```python
from functools import lru_cache

@lru_cache(maxsize=None)
def fibonacci(n: int) -> int:
    """
    计算斐波那契数列第 n 项（使用 memoization 优化）
    时间复杂度: O(n)，空间复杂度: O(n)
    """
    if n < 0:
        raise ValueError("n 必须为非负整数")
    if n <= 1:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

## 五、高级功能

### 5.1 代码文件管理

对于复杂的代码任务，助手可以生成文件并通过微信发送：

```python
async def send_code_files(self, work_dir: Path, chat_id: str):
    """将生成的代码文件打包发送到微信"""
    import zipfile

    zip_path = work_dir / "result.zip"
    with zipfile.ZipFile(zip_path, 'w') as zf:
        for file in work_dir.rglob('*'):
            if file.is_file() and file.suffix in {'.py', '.js', '.ts', '.html', '.css'}:
                zf.write(file, file.relative_to(work_dir))

    # 通过微信发送文件
    await self.send_file(chat_id, str(zip_path), "代码文件.zip")

# src/session_manager.py
import redis
import json

class SessionManager:
    def __init__(self, redis_url: str = "redis://localhost:6379/0"):
        self.redis = redis.from_url(redis_url)
        self.ttl = 3600  # 会话过期时间：1小时

    def get_history(self, session_key: str, max_messages: int = 20) -> list:
        """获取会话历史"""
        history = self.redis.lrange(f"session:{session_key}", 0, max_messages - 1)
        return [json.loads(msg) for msg in history]

    def add_message(self, session_key: str, role: str, content: str):
        """添加消息到会话历史"""
        message = {"role": role, "content": content, "timestamp": time.time()}
        self.redis.lpush(f"session:{session_key}", json.dumps(message))
        self.redis.expire(f"session:{session_key}", self.ttl)

# config/scheduler.yaml
cron_jobs:
  - name: "daily-code-review"
    schedule: "0 22 * * *"  # 每晚 10 点
    action:
      type: "claude-code"
      prompt: |
        请回顾今天所有通过微信提交的代码任务，生成一份日报：
        1. 完成了哪些任务
        2. 生成了哪些文件
        3. 待改进的建议
    notify:
      type: "wechat"
      target: "owner"

security:
  # 消息内容安全
  content_filter:
    enabled: true
    block_patterns:
      - "删除.*数据库"
      - "格式化.*磁盘"
      - "DROP TABLE"

  # 代码执行安全
  code_sandbox:
    enabled: true
    max_execution_time: 60s
    max_memory: 512MB
    no_network: true  # 隔离网络

  # 用户权限
  access_control:
    admin_users: ["wechat_id_1", "wechat_id_2"]
    readonly_users: ["wechat_id_3"]
    coding_allowed: true  # 是否允许代码执行

# src/monitoring.py
class HealthMonitor:
    def __init__(self):
        self.metrics = {
            "messages_received": 0,
            "messages_processed": 0,
            "errors": 0,
            "avg_response_time": 0,
        }

    async def check_health(self) -> dict:
        return {
            "gateway": await self._check_gateway(),
            "claude_api": await self._check_claude_api(),
            "redis": await self._check_redis(),
            "wechat_connection": await self._check_wechat(),
        }