--- title: "LLM Slack集成" description: "全面介绍大语言模型与Slack平台的集成方案，涵盖Bot开发、消息处理和自动化工作流构建" tags: ["Slack", "LLM集成", "Bot开发", "消息处理", "自动化"] category: "llm" icon: "🧠"

LLM Slack集成

Slack作为全球最流行的企业协作平台之一，拥有丰富的API和应用生态。将大语言模型（LLM）与Slack集成，可以打造智能助手Bot，实现消息自动处理、内容生成和团队协作增强。本文将从Bot开发、消息处理和自动化工作流三个维度，详细介绍LLM与Slack的集成方案。

Slack Bot开发基础

Slack Bot是基于Slack API的自动化程序，可以监听频道消息、发送回复、执行命令。开发LLM驱动的Slack Bot，需要先搭建基础框架。

import os
from slack_sdk import WebClient
from slack_sdk.socket_mode import SocketModeClient
import openai

SLACK_BOT_TOKEN = os.environ["SLACK_BOT_TOKEN"]
SLACK_APP_TOKEN = os.environ["SLACK_APP_TOKEN"]

client = WebClient(token=SLACK_BOT_TOKEN)
socket_client = SocketModeClient(app_token=SLACK_APP_TOKEN, web_client=client)

def ask_llm(question):
    response = openai.ChatCompletion.create(
        model="gpt-4",
        messages=[{"role": "user", "content": question}],
        max_tokens=1000
    )
    return response.choices[0].message.content

@socket_client.socket_mode_request_listener
def process_message(req):
    if req.type == "events_api" and req.payload["event"]["type"] == "message":
        event = req.payload["event"]
        text = event.get("text", "")
        if "<@BOT_ID>" in text:
            clean_text = text.replace("<@BOT_ID>", "").strip()
            answer = ask_llm(clean_text)
            client.chat_postMessage(channel=event["channel"], text=answer)

这段代码展示了如何使用Slack SDK的Socket Mode接收消息，当检测到@Bot时提取问题内容，调用LLM生成回答并回复到频道。Socket Mode无需公网IP，非常适合本地开发和内部部署场景。

消息处理与智能回复

在实际应用中，消息处理需要考虑多种情况：不同频道的上下文、消息线程的维护、以及消息格式的转换。构建一个健壮的消息处理系统是LLM Slack集成的核心。

import json
from collections import defaultdict

class SlackMessageHandler:
    def __init__(self, llm_client, slack_client):
        self.llm = llm_client
        self.slack = slack_client
        self.conversations = defaultdict(list)
    
    def handle_message(self, channel, user, text, thread_ts=None):
        context = self.get_thread_context(channel, thread_ts)
        
        messages = [{"role": "system", "content": "你是一个专业的团队助手，根据上下文提供准确的回答。"}]
        for msg in context:
            messages.append({"role": "user", "content": msg["text"]})
        messages.append({"role": "user", "content": text})
        
        response = self.llm.chat.completions.create(
            model="gpt-4",
            messages=messages,
            max_tokens=800
        )
        answer = response.choices[0].message.content
        
        self.slack.chat_postMessage(
            channel=channel,
            text=answer,
            thread_ts=thread_ts or None
        )
        return answer
    
    def get_thread_context(self, channel, thread_ts):
        if not thread_ts:
            return []
        history = self.slack.conversations_replies(channel=channel, ts=thread_ts)
        return [{"text": m["text"], "user": m["user"]} for m in history["messages"][:-1]]

这个消息处理器支持线程上下文维护，能够将对话历史传递给LLM，实现多轮对话。通过维护conversations字典，可以追踪每个频道的对话状态，提供连贯的交互体验。

自动化工作流构建

LLM与Slack集成的最高价值在于构建自动化工作流。典型场景包括：每日站会摘要生成、周报自动整理、会议纪要总结、跨团队信息同步等。

from datetime import datetime, timedelta
import schedule

def daily_standup_summary():
    channels = ["#engineering", "#product", "#design"]
    for ch in channels:
        history = client.conversations_history(
            channel=ch,
            oldest=(datetime.now() - timedelta(days=1)).timestamp()
        )
        messages = [m["text"] for m in history["messages"] if "BOT_ID" not in m.get("user", "")]
        
        if messages:
            prompt = f"将以下频道消息整理为简洁的每日站会摘要：\n\n" + "\n".join(messages[:20])
            summary = ask_llm(prompt)
            client.chat_postMessage(channel="#daily-standup", text=f"📋 {ch} 每日摘要:\n{summary}")

def auto_review_request(payload):
    pr_url = payload["pr_url"]
    diff = fetch_pr_diff(pr_url)
    review_prompt = f"审查以下代码变更并给出改进建议：\n\n{diff[:3000]}"
    review = ask_llm(review_prompt)
    client.chat_postMessage(channel="#code-review", text=f"🤖 AI代码审查 {pr_url}:\n{review}")

schedule.every().day.at("09:00").do(daily_standup_summary)

工作流调度使用schedule库实现定时任务，自动从多个频道收集消息并生成摘要。代码审查工作流通过监听GitHub Webhook触发，自动拉取PR差异并调用LLM进行审查。

部署与最佳实践

生产环境部署LLM Slack Bot需要注意以下几点。首先使用环境变量管理所有密钥，避免硬编码。其次实施速率限制，防止LLM API调用超限。最后建立监控告警机制，确保Bot稳定性。

import time
from functools import wraps

def rate_limit(calls_per_minute=20):
    calls = []
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            now = time.time()
            calls[:] = [c for c in calls if now - c < 60]
            if len(calls) >= calls_per_minute:
                raise Exception("Rate limit exceeded")
            calls.append(now)
            return func(*args, **kwargs)
        return wrapper
    return decorator

将速率限制装饰器应用于LLM调用函数，可以有效控制API使用量，避免因突发流量导致的成本激增和服务中断。

总结

LLM与Slack的集成为企业协作带来了智能化升级。从基础的消息收发到复杂的自动化工作流，LLM能力的注入让团队沟通更加高效。开发者可以根据实际需求，灵活组合消息处理、上下文管理和工作流调度模块，构建适合自身团队的智能协作工具。