零操作让 AI 知道你的全部知识背景——我的 Claude Code Hook 实战

你输入: "Kafka 消费延迟怎么处理？"
     ↓
Hook 拦截，检测到关键词: Kafka、消费延迟
     ↓
从知识库中匹配到相关页面，直接注入到消息中
     ↓
AI 收到的内容 = 你的问题 + 你之前学过的 Kafka 知识
     ↓
AI 的回答已经基于你的经验，不是泛泛而谈

{
  "hooks": {
    "UserPromptSubmit": [{
      "type": "command",
      "command": "~/.claude/scripts/exomind-context.sh"
    }]
  }
}

# stdin = 用户消息
MSG=$(cat)

# 处理匹配逻辑...

# stdout = 注入到上下文的内容
echo "[ExoMind] 以下是相关知识：..."

MSG=$(cat)  # 从 stdin 读取用户消息

# 短消息跳过（减少噪音）
[ ${#MSG} -lt 8 ] && exit 0

# 无关指令跳过
case "$MSG" in
    "好"|"继续"|"是"|"ok"|"yes"|"no")
        exit 0 ;;
esac

import re

msg = sys.stdin.read().strip()

# 英文：精确匹配（3字符以上）
en_words = set(re.findall(r'[A-Za-z][A-Za-z0-9_+-]{2,}', msg))

# 中文：滑动窗口匹配（2-6字符子串）
cn_words = set(re.findall(r'[\u4e00-\u9fff]{2,6}', msg))

# 在知识库索引中查找匹配
hits = []
for w in en_words:
    if w in keyword_index:
        hits.append(w)

for w in cn_words:
    if w in keyword_index:
        hits.append(w)
    else:
        # 滑动窗口：从长到短尝试匹配
        for length in range(min(len(w), 6), 1, -1):
            for i in range(len(w) - length + 1):
                sub = w[i:i+length]
                if sub in keyword_index:
                    hits.append(sub)
                    break

for keyword in hits[:3]:  # 最多注入 3 个关键词
    for subdir in ['entities', 'concepts', 'summaries']:
        filepath = os.path.join(wiki_path, subdir, keyword + '.md')
        if os.path.exists(filepath):
            content = open(filepath, 'r').read()
            body = extract_body(content)  # 去掉 frontmatter
            if len(body) > 800:
                body = body[:800] + '...'
            output_parts.append(f'### {keyword}\n{body}')
            break

print('[ExoMind 知识库上下文] 以下是与当前话题相关的已有知识：')
print('\n'.join(output_parts))

# 会话级去重：30 分钟内不重复注入
dedup_file = '/tmp/.exomind-dedup-' + session_hash + '.json'
dedup = load_dedup(dedup_file)
dedup = {k: v for k, v in dedup.items() if now - v < 1800}  # 清理过期

# 已注入的关键词再次被引用 = 消费（不是重复注入）
consumed = [h for h in hits if h in dedup]
hits = [h for h in hits if h not in dedup]

# 记录消费事件（影响知识质量评分）
for kw in consumed:
    update_quality_score(kw, 'consumptions', +1)

第一次提到 Redis → 注入 Redis 知识（记录 click）
第二次提到 Redis → 检测到消费（记录 consumption，不重复注入）

# 经验/洞察关键词检测
experience_patterns = [
    r'关键经验', r'踩坑', r'最佳实践', r'架构决策',
    r'根因分析', r'优化效果', r'心得', r'避坑',
    r'lesson learned', r'best practice', r'gotcha',
]

# 或者结构化特征：3个以上 bullet point + 技术关键词
bullet_count = len(re.findall(r'^\s*[-*•]\s+', msg, re.MULTILINE))
has_tech_terms = bool(re.search(r'(API|框架|架构|缓存|数据库)', msg))

if insight_count >= 1 or (bullet_count >= 3 and has_tech_terms):
    print('[ExoMind 自动摄入指令] 检测到经验/洞察内容。')
    print('请立即调用 mcp__exomind__ingest 将以上内容摄入知识库。')

05-10: 1 次 █
05-11: 9 次 ████████████████████ （批量开发日，触发最多）
05-12: 7 次 ██████████████
05-13: 2 次 ████
05-14: 2 次 ████
05-15: 1 次 ██
05-16: 1 次 ██

我输入: "Redis 分布式锁怎么实现？"

Hook 自动注入:
  - Redis（实体页面，64条关系）
  - Redis分布式锁（概念页面）
  - 持久化（关联概念）

Claude 的回答:
  直接引用了我之前学过的 RDB/AOF 对锁可靠性的影响，
  而不是泛泛介绍 SETNX + EXPIRE。

我输入: "微信 access_token 过期怎么处理？"

Hook 自动注入:
  - 微信公众号 access_token 机制
  - OAuth 2.0 token 刷新模式
  - 之前项目中踩过的坑

Claude 的回答:
  基于我自己的项目经验回答，提到了我之前遇到过的
  token 刷新竞争条件问题。