LangChain / LlamaIndex / Vercel AI SDK：三大框架集成实战与性能对比

第六十二章：与 Slack/Teams 集成：企业通信平台中的 AI 助手

62.1 企业通信平台集成的价值

Slack 和 Microsoft Teams 是现代企业最主要的沟通协作平台。将 Claude 集成进这两个平台，意味着 AI 能力直接融入员工的日常工作流：

• 即时响应：员工无需切换到其他工具，直接在聊天界面获得 AI 帮助
• 上下文感知：Bot 可以读取频道历史、理解团队语境
• 权限管理：基于工作区、频道、用户角色控制 AI 能力边界
• 审计追踪：所有 AI 交互记录在平台日志中，满足合规要求

本章重点介绍使用 Slack Bolt 框架（Python 版本）集成 Claude，以及 Microsoft Teams Bot Framework 的集成方案。

62.2 Slack Bolt + Claude：完整集成方案

62.2.1 环境配置

创建 Slack App：

1. 访问 https://api.slack.com/apps，点击 Create New App
2. 选择 From scratch，填写 App Name 和工作区
3. 在 OAuth & Permissions 中添加 Bot Token Scopes：
   • app_mentions:read（响应 @提及）
   • channels:history（读取公开频道历史）
   • chat:write（发送消息）
   • im:history（DM 历史）
   • im:write（DM 回复）
   • commands（Slash Commands）
4. 在 Event Subscriptions 中启用并添加事件：
   • app_mention
   • message.im
5. 安装 App 到工作区，获取 Bot Token（xoxb-...）
6. 在 Basic Information 获取 Signing Secret

pip install slack-bolt anthropic

# .env 文件
SLACK_BOT_TOKEN=xoxb-xxxxxxxxxxxx
SLACK_SIGNING_SECRET=xxxxxxxxxxxxxxxx
ANTHROPIC_API_KEY=sk-ant-xxxxxxxxxxxxxx

62.2.2 基础 Bot：响应 @提及

# slack_claude_bot.py
import os
from slack_bolt import App
from slack_bolt.adapter.socket_mode import SocketModeHandler
import anthropic
from dotenv import load_dotenv

load_dotenv()

# 初始化 Slack App
app = App(
    token=os.environ["SLACK_BOT_TOKEN"],
    signing_secret=os.environ["SLACK_SIGNING_SECRET"]
)

# 初始化 Claude 客户端
claude = anthropic.Anthropic()

# 对话历史存储（生产环境应使用 Redis/数据库）
conversation_history = {}

@app.event("app_mention")
def handle_mention(event, say, client):
    """响应频道中的 @Bot 提及"""
    channel_id = event["channel"]
    user_id = event["user"]
    thread_ts = event.get("thread_ts", event["ts"])  # 优先使用线程 ts
    
    # 从消息中去除 Bot 提及标记（<@BOT_ID>）
    text = event["text"]
    bot_id = client.auth_test()["user_id"]
    text = text.replace(f"<@{bot_id}>", "").strip()
    
    if not text:
        say("你好！有什么我可以帮助你的？", thread_ts=thread_ts)
        return
    
    # 获取线程历史（如果在线程中）
    session_key = f"{channel_id}:{thread_ts}"
    if session_key not in conversation_history:
        conversation_history[session_key] = []
    
    history = conversation_history[session_key]
    history.append({"role": "user", "content": text})
    
    # 显示"正在输入"状态
    client.reactions_add(
        channel=channel_id,
        name="hourglass_flowing_sand",
        timestamp=event["ts"]
    )
    
    try:
        response = claude.messages.create(
            model="claude-opus-4-5",
            max_tokens=1024,
            system="""你是公司的 AI 助手 Aria，部署在 Slack 中。
你的职责是帮助员工回答问题、分析数据、起草文档、解释技术概念。
回答时使用 Slack 的 Markdown 格式（*粗体*、`代码`、```代码块```）。
对于需要隐私信息的请求，请建议用户通过私信（DM）联系你。""",
            messages=history
        )
        
        reply_text = response.content[0].text
        history.append({"role": "assistant", "content": reply_text})
        
        # 限制历史长度（避免超出 token 限制）
        if len(history) > 20:
            conversation_history[session_key] = history[-20:]
        
        say(reply_text, thread_ts=thread_ts)
        
    except Exception as e:
        say(f":warning: 处理请求时出错：{str(e)}", thread_ts=thread_ts)
    finally:
        # 移除"正在输入"状态
        client.reactions_remove(
            channel=channel_id,
            name="hourglass_flowing_sand",
            timestamp=event["ts"]
        )

@app.event("message")
def handle_dm(event, say):
    """响应私信（DM）"""
    # 忽略 Bot 自己的消息
    if event.get("bot_id"):
        return
    # 只处理 DM（channel type 为 im）
    if event.get("channel_type") != "im":
        return
    
    user_id = event["user"]
    text = event.get("text", "").strip()
    
    if not text:
        return
    
    # DM 使用用户 ID 作为会话 key
    session_key = f"dm:{user_id}"
    if session_key not in conversation_history:
        conversation_history[session_key] = []
    
    history = conversation_history[session_key]
    history.append({"role": "user", "content": text})
    
    response = claude.messages.create(
        model="claude-opus-4-5",
        max_tokens=2048,
        system="你是员工的私人 AI 助手。在私信中可以处理更敏感的内容，如绩效反馈起草、个人发展规划等。",
        messages=history
    )
    
    reply = response.content[0].text
    history.append({"role": "assistant", "content": reply})
    say(reply)

if __name__ == "__main__":
    handler = SocketModeHandler(app, os.environ["SLACK_APP_TOKEN"])
    handler.start()

62.2.3 Slash Command：结构化功能入口

Slash Command 允许用户通过 /command 触发特定的 AI 功能，适合高频、标准化的使用场景。

# 在 Slack App 配置中添加 Slash Command：
# Command: /summarize
# Request URL: https://your-server.com/slack/events
# Description: 总结当前频道的讨论内容

@app.command("/summarize")
def handle_summarize(ack, body, client, respond):
    """总结频道或线程的历史消息"""
    ack()  # 必须在 3 秒内 ack，否则 Slack 显示超时
    
    channel_id = body["channel_id"]
    user_id = body["user_id"]
    text = body.get("text", "").strip()
    
    # 获取频道最近 50 条消息
    result = client.conversations_history(
        channel=channel_id,
        limit=50
    )
    
    messages = result["messages"]
    if not messages:
        respond("该频道没有可用的消息历史。")
        return
    
    # 格式化消息（排除 Bot 消息和系统消息）
    formatted_messages = []
    for msg in reversed(messages):  # 时间正序
        if msg.get("bot_id") or msg.get("subtype"):
            continue
        user_info = client.users_info(user=msg.get("user", "unknown"))
        username = user_info["user"]["real_name"] if user_info["ok"] else "Unknown"
        formatted_messages.append(f"{username}: {msg.get('text', '')}")
    
    conversation_text = "\n".join(formatted_messages)
    
    # 调用 Claude 生成摘要
    scope = text if text else "最近 50 条消息"
    
    response = claude.messages.create(
        model="claude-opus-4-5",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"请总结以下 Slack 频道对话（{scope}）的主要内容，包括：\n1. 主要讨论话题\n2. 达成的决定（如有）\n3. 待解决的问题（如有）\n\n对话内容：\n{conversation_text}"
        }]
    )
    
    summary = response.content[0].text
    
    # 使用 Block Kit 格式化输出
    respond({
        "blocks": [
            {
                "type": "header",
                "text": {"type": "plain_text", "text": "频道讨论摘要"}
            },
            {
                "type": "section",
                "text": {"type": "mrkdwn", "text": summary}
            },
            {
                "type": "context",
                "elements": [
                    {
                        "type": "mrkdwn",
                        "text": f"由 <@{user_id}> 触发 · 基于最近 {len(formatted_messages)} 条消息"
                    }
                ]
            }
        ]
    })

@app.command("/draft")
def handle_draft(ack, body, respond):
    """起草指定类型的文档"""
    ack()
    
    text = body.get("text", "").strip()
    if not text:
        respond("请提供起草内容，例如：`/draft 会议纪要 讨论了Q3目标和资源分配`")
        return
    
    # 解析类型和内容（第一个词为类型）
    parts = text.split(" ", 1)
    doc_type = parts[0]
    content = parts[1] if len(parts) > 1 else ""
    
    DOC_TYPE_PROMPTS = {
        "会议纪要": "你是专业的会议纪要撰写助手。基于提供的会议要点，生成标准格式的会议纪要，包含：时间/出席人员占位符、讨论内容、决议事项、后续行动项。",
        "邮件": "你是专业的商务写作助手。根据提供的内容要点，起草一封专业的商务邮件。",
        "公告": "你是公司内部公告起草助手。根据提供的信息，起草一则清晰、正式的内部公告。"
    }
    
    system_prompt = DOC_TYPE_PROMPTS.get(doc_type, f"你是专业的{doc_type}起草助手。")
    
    response = claude.messages.create(
        model="claude-opus-4-5",
        max_tokens=1024,
        system=system_prompt,
        messages=[{"role": "user", "content": f"请起草：{content}"}]
    )
    
    respond(f"*{doc_type}草稿*\n\n{response.content[0].text}")

62.2.4 事件订阅：监听特定触发条件

@app.event("reaction_added")
def handle_reaction(event, client):
    """当用户对消息添加特定 emoji 时触发 Claude 处理"""
    # 当用户添加 :summarize: emoji 时，自动生成消息摘要
    if event["reaction"] != "summarize":
        return
    
    channel_id = event["item"]["channel"]
    message_ts = event["item"]["ts"]
    
    # 获取被 reaction 的消息
    result = client.conversations_history(
        channel=channel_id,
        latest=message_ts,
        limit=1,
        inclusive=True
    )
    
    if not result["messages"]:
        return
    
    message_text = result["messages"][0].get("text", "")
    
    response = claude.messages.create(
        model="claude-haiku-4-5",
        max_tokens=256,
        messages=[{
            "role": "user",
            "content": f"用一句话总结以下内容的核心要点：\n\n{message_text}"
        }]
    )
    
    # 在消息的线程中回复摘要
    client.chat_postMessage(
        channel=channel_id,
        thread_ts=message_ts,
        text=f":robot_face: *一句话摘要*\n{response.content[0].text}"
    )

62.3 Microsoft Teams 集成

62.3.1 Teams Bot Framework 架构

# Microsoft Teams Bot 使用 botbuilder-python SDK
pip install botbuilder-core botbuilder-integration-aiohttp anthropic aiohttp

# teams_claude_bot.py
from botbuilder.core import ActivityHandler, TurnContext, MessageFactory
from botbuilder.schema import Activity, ActivityTypes
import anthropic
import asyncio

class ClaudeBot(ActivityHandler):
    """Microsoft Teams 中的 Claude Bot"""
    
    def __init__(self):
        self.claude = anthropic.Anthropic()
        self.conversation_histories = {}
    
    async def on_message_activity(self, turn_context: TurnContext):
        """处理用户发送的消息"""
        user_id = turn_context.activity.from_property.id
        conversation_id = turn_context.activity.conversation.id
        text = turn_context.activity.text.strip()
        
        # 去除 Teams 中的 HTML 标签（Teams 消息可能包含 HTML）
        import re
        text = re.sub(r'<[^>]+>', '', text).strip()
        
        if not text:
            return
        
        session_key = f"{conversation_id}:{user_id}"
        if session_key not in self.conversation_histories:
            self.conversation_histories[session_key] = []
        
        history = self.conversation_histories[session_key]
        history.append({"role": "user", "content": text})
        
        # 发送"正在输入"指示器
        await turn_context.send_activity(Activity(type=ActivityTypes.typing))
        
        response = self.claude.messages.create(
            model="claude-opus-4-5",
            max_tokens=2048,
            system="""你是部署在 Microsoft Teams 中的企业 AI 助手。
回答格式请使用 Markdown（Teams 支持部分 Markdown）。
对于代码示例，使用代码块格式。""",
            messages=history
        )
        
        reply_text = response.content[0].text
        history.append({"role": "assistant", "content": reply_text})
        
        # 限制历史长度
        if len(history) > 20:
            self.conversation_histories[session_key] = history[-20:]
        
        await turn_context.send_activity(
            MessageFactory.text(reply_text)
        )
    
    async def on_members_added_activity(self, members_added, turn_context: TurnContext):
        """当 Bot 被添加到对话时发送欢迎消息"""
        for member in members_added:
            if member.id != turn_context.activity.recipient.id:
                await turn_context.send_activity(
                    MessageFactory.text(
                        "你好！我是 Claude AI 助手。我可以帮你回答问题、分析文档、起草内容。"
                        "直接向我发消息即可开始！"
                    )
                )

62.3.2 Teams Webhook 配置（Incoming Webhook）

对于简单的单向通知场景，可以使用 Teams 的 Incoming Webhook 将 Claude 分析结果推送到频道：

import anthropic
import requests
import json

claude = anthropic.Anthropic()

def analyze_and_notify_teams(data: dict, teams_webhook_url: str):
    """分析数据并通过 Teams Webhook 发送结果"""
    
    # Claude 分析
    response = claude.messages.create(
        model="claude-haiku-4-5",
        max_tokens=512,
        messages=[{
            "role": "user",
            "content": f"分析以下销售数据，识别异常趋势，给出3条关键洞察：\n{json.dumps(data, ensure_ascii=False, indent=2)}"
        }]
    )
    
    insights = response.content[0].text
    
    # 构建 Teams Adaptive Card 格式
    card = {
        "type": "message",
        "attachments": [
            {
                "contentType": "application/vnd.microsoft.card.adaptive",
                "content": {
                    "$schema": "http://adaptivecards.io/schemas/adaptive-card.json",
                    "type": "AdaptiveCard",
                    "version": "1.4",
                    "body": [
                        {
                            "type": "TextBlock",
                            "text": "AI 销售数据分析报告",
                            "weight": "Bolder",
                            "size": "Large"
                        },
                        {
                            "type": "TextBlock",
                            "text": insights,
                            "wrap": True
                        },
                        {
                            "type": "FactSet",
                            "facts": [
                                {"title": "分析模型", "value": "Claude Haiku"},
                                {"title": "生成时间", "value": "自动生成"}
                            ]
                        }
                    ]
                }
            }
        ]
    }
    
    requests.post(teams_webhook_url, json=card)

62.4 跨平台共用 Claude 逻辑

当同时维护 Slack 和 Teams 集成时，将 Claude 交互逻辑抽象为共用服务层可以避免代码重复：

# claude_service.py - 平台无关的 Claude 服务层
import anthropic
from dataclasses import dataclass
from typing import Optional

@dataclass
class ConversationContext:
    """对话上下文"""
    platform: str  # "slack" | "teams"
    user_id: str
    channel_id: str
    thread_id: Optional[str] = None

class ClaudeService:
    """平台无关的 Claude 服务层"""
    
    def __init__(self):
        self.client = anthropic.Anthropic()
        self._histories: dict[str, list] = {}
    
    def _get_session_key(self, ctx: ConversationContext) -> str:
        return f"{ctx.platform}:{ctx.channel_id}:{ctx.user_id}"
    
    def _get_system_prompt(self, ctx: ConversationContext) -> str:
        base = "你是企业 AI 助手，帮助员工提高工作效率。"
        if ctx.platform == "slack":
            base += "使用 Slack Markdown 格式（*粗体*、`代码`）。"
        elif ctx.platform == "teams":
            base += "使用 Microsoft Teams 支持的 Markdown 格式。"
        return base
    
    def chat(
        self,
        message: str,
        ctx: ConversationContext,
        model: str = "claude-opus-4-5",
        max_tokens: int = 1024
    ) -> str:
        """处理对话消息"""
        key = self._get_session_key(ctx)
        if key not in self._histories:
            self._histories[key] = []
        
        history = self._histories[key]
        history.append({"role": "user", "content": message})
        
        response = self.client.messages.create(
            model=model,
            max_tokens=max_tokens,
            system=self._get_system_prompt(ctx),
            messages=history
        )
        
        reply = response.content[0].text
        history.append({"role": "assistant", "content": reply})
        
        # 保持历史记录在合理范围
        if len(history) > 20:
            self._histories[key] = history[-20:]
        
        return reply
    
    def clear_history(self, ctx: ConversationContext):
        """清除会话历史"""
        key = self._get_session_key(ctx)
        self._histories.pop(key, None)

# 在 Slack Bot 中使用
claude_service = ClaudeService()

@app.event("app_mention")
def handle_slack_mention(event, say, client):
    ctx = ConversationContext(
        platform="slack",
        user_id=event["user"],
        channel_id=event["channel"],
        thread_id=event.get("thread_ts")
    )
    text = extract_mention_text(event, client)
    reply = claude_service.chat(text, ctx)
    say(reply, thread_ts=ctx.thread_id or event["ts"])

62.5 安全与合规考虑

62.5.1 内容过滤

SENSITIVE_KEYWORDS = ["工资", "薪资", "裁员", "竞争对手机密"]

def pre_process_message(text: str, platform: str) -> tuple[str, bool]:
    """消息预处理：检测敏感内容"""
    for keyword in SENSITIVE_KEYWORDS:
        if keyword in text:
            return text, True  # True = 需要额外审查
    return text, False

@app.event("app_mention")  
def handle_mention_with_filter(event, say):
    text = event["text"]
    processed_text, needs_review = pre_process_message(text, "slack")
    
    if needs_review:
        say(":lock: 检测到敏感话题，此对话已被标记供安全团队审查。我仍会尽力帮助你，但请注意企业信息安全政策。")
        log_sensitive_interaction(event)
    
    # 继续正常处理...

62.5.2 速率限制（防止滥用）

from collections import defaultdict
import time

class RateLimiter:
    def __init__(self, max_requests: int, window_seconds: int):
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self._requests: dict[str, list] = defaultdict(list)
    
    def is_allowed(self, user_id: str) -> bool:
        now = time.time()
        user_requests = self._requests[user_id]
        
        # 清除过期记录
        self._requests[user_id] = [t for t in user_requests if now - t < self.window_seconds]
        
        if len(self._requests[user_id]) >= self.max_requests:
            return False
        
        self._requests[user_id].append(now)
        return True

rate_limiter = RateLimiter(max_requests=20, window_seconds=3600)  # 每用户每小时 20 次

@app.event("app_mention")
def handle_mention_with_rate_limit(event, say):
    user_id = event["user"]
    if not rate_limiter.is_allowed(user_id):
        say(f"<@{user_id}> 你已达到本小时的使用限额（20次），请稍后再试。")
        return
    # 正常处理...

小结

Slack Bolt 框架通过事件监听（app_mention、message.im）、Slash Command 和 Reaction 事件三种方式将 Claude 嵌入 Slack 工作流。Microsoft Teams 通过 Bot Framework SDK 提供类似能力，并支持 Adaptive Card 格式化输出。最佳实践是将 Claude 交互逻辑抽象为平台无关的服务层，复用于多个平台；同时配置内容过滤和速率限制保障企业安全合规。生产部署中，对话历史应存储在 Redis 等持久化存储中，而非进程内存，确保服务重启后的会话连续性。