Reactance0083/pydantic-ai-email-linear-auto-triage

## 当前状态 本仓库是一个预览版/未经验证的入门项目。它不是活跃的旗舰产品，目前也未经过商业验证。在进行全新的商业就绪检查并移除此通知之前，请勿将其视为可直接购买的生产包。 # Email→Linear Issue 自动分拣 使用 AI 驱动的分拣自动将收到的电子邮件转换为已确定优先级的 Linear issue。从原始邮件内容中提取客户信息、优先级和问题类型，然后通过 Slack 针对紧急事项发送通知，并在您的 Linear 工作区中即时创建结构化的工单。 ## 概述 此模板提供了一个预览版 FastAPI webhook 服务，它可以： - 通过 SMTP 转发或 Gmail API 集成接收电子邮件 - 使用 Claude AI 提取优先级、客户和问题分类 - 创建包含丰富元数据和适当链接的 Linear issue - 为高优先级工单发送 Slack 警报 - 存储分拣决策以供审计和完善 **为什么使用它？** - **节约成本：** 消除每月 20-40 美元的 Zapier 费用及人工分拣开销 - **速度：** 30 秒的邮件到工单流水线，而人工路由需要 5 分钟 - **一致性：** AI 驱动的分类减少了优先级分配中的人为错误 - **可扩展性：** 基于 Pydantic AI 构建，易于自定义分拣逻辑 ## 它的功能 ### 邮件摄取 - 接收原始邮件 POST payload（SMTP webhook 格式或解析后的 JSON） - 提取发件人、主题、正文和附件元数据 - 支持纯文本和 HTML 邮件正文 ### AI 驱动的分拣 使用 Claude 从邮件内容中提取： - **优先级**（urgent/high/normal/low） - **客户标识符**（邮件域、姓名、账户 ID） - **问题类型**（bug/feature-request/support/billing） - **摘要**（根据主题 + 正文上下文自动生成） - **建议的指派人**（基于问题类型模式，可选） ### Linear 集成 - 在您的 Linear 工作区中创建 issue - 将原始邮件作为评论附加到 issue 中 - 根据分拣输出设置优先级和状态 - 链接到客户/团队项目（可配置） - 支持用于邮件元数据的自定义字段 ### Slack 通知 - 将紧急/高优先级的工单发布到指定频道 - 包含客户信息、issue 链接和优先级徽章 - 可选的主题回复以供后续更新 ### 审计与历史记录 - 将所有分拣决策存储在 SQLite（或配置的 DB）中 - 支持性能监控和模型完善 - 支持手动覆盖和反馈循环 ## 前置条件 - **Python 3.11+** - **Linear API token**（在 [Settings > API > Personal API Keys](https://linear.app/settings/api) 中创建） - **Claude API key**（来自 [Anthropic Console](https://console.anthropic.com/)） - **Slack webhook URL**（可选，来自 [Slack Apps](https://api.slack.com/apps)） - **Gmail API 凭据**或 SMTP 中继服务（可选，用于邮件摄取） ### 可选项 - Docker + Docker Compose（用于容器化部署） - PostgreSQL（用于生产数据库，默认为 SQLite） ## 设置 ### 1. 克隆并安装依赖 ``` git clone https://github.com/yourusername/email-linear-triage.git cd email-linear-triage python -m venv venv source venv/bin/activate # On Windows: venv\Scripts\activate pip install -r requirements.txt ``` ### 2. 创建环境文件 在项目根目录创建 `.env`： ``` # API Keys ANTHROPIC_API_KEY=sk-ant-... LINEAR_API_KEY=lin_api_... SLACK_WEBHOOK_URL=https://hooks.slack.com/services/YOUR/WEBHOOK/URL # Optional # Linear 配置 LINEAR_TEAM_ID=acme # Your Linear team slug (e.g., 'acme' from linear.app/acme) LINEAR_PROJECT_ID=INB # Project key for incoming emails (default: 'INB') LINEAR_DEFAULT_STATUS=backlog # Initial status for new issues # Email 配置 SMTP_SECRET_TOKEN=your-secret-token-here # For webhook authentication EMAIL_DOMAIN=yourdomain.com # Database（可选） DATABASE_URL=sqlite:///./triage.db # Or: postgresql://user:pass@localhost/triage # Feature Flags ENABLE_SLACK_NOTIFICATIONS=true ENABLE_AUTO_ASSIGN=false TRIAGE_MODEL=claude-3-5-sonnet-20241022 # Claude model to use ``` ### 3. 初始化数据库 ``` python -m alembic upgrade head ``` 或对于 SQLite（自动创建）： ``` python -c "from app.db import init_db; init_db()" ``` ### 4. 运行服务器 ``` uvicorn app.main:app --host 0.0.0.0 --port 8000 --reload ``` 服务器运行在 `http://localhost:8000` ### 5. 配置邮件路由 **选项 A：SMTP 转发**（推荐） - 在您的邮件提供商处设置转发规则： - **From：** `tickets@yourdomain.com` - **To：** `{your-server}/webhook/email`（带身份验证） **选项 B：Gmail API** - 在 Google Cloud Console 中启用 [Gmail API](https://developers.google.com/gmail/api/quickstart/python) - 将凭据 JSON 下载到 `./credentials.json` - 应用会定期自动获取带标签的邮件 **选项 C：手动测试** ``` curl -X POST http://localhost:8000/webhook/email \ -H "X-Webhook-Token: your-secret-token-here" \ -H "Content-Type: application/json" \ -d '{ "from": "customer@example.com", "subject": "Payment processing is broken", "body": "Hi, our recurring invoices havent charged for 2 days. This is urgent!", "timestamp": "2024-01-15T14:30:00Z" }' ``` ### 6. 设置 Linear Webhook（可选，用于未来的集成） 在 Linear Settings > Integrations > Webhooks 中，添加： - **URL：** `{your-server}/webhook/linear-event` - **Events：** Issue created、issue updated - 可用于通过邮件回复关闭 issue ## 用法 ### 基础的邮件到 Issue 流程 1. **邮件到达** `tickets@yourdomain.com`（通过 SMTP 转发） 2. **Webhook 处理程序** 接收 POST 请求并验证 token 3. **Claude 分拣** 对邮件进行分类（2-5 秒） 4. **创建 Linear issue** 并附带提取的元数据 5. **发送 Slack 通知**（如果为 urgent/high） 6. **返回响应**，包含 issue URL ### 示例：发送邮件 ``` curl -X POST http://localhost:8000/webhook/email \ -H "X-Webhook-Token: your-secret-token-here" \ -H "Content-Type: application/json" \ -d '{ "from": "sarah@acmecorp.com", "subject": "[BUG] Dashboard crashes on mobile", "body": "When I open the dashboard on iPhone, it instantly crashes. Happens every time. Our team cant work.", "timestamp": "2024-01-15T09:30:00Z" }' ``` **响应（201 Created）：** ``` { "status": "success", "linear_issue_id": "INB-234", "linear_issue_url": "https://linear.app/acme/issue/INB-234", "triage_result": { "priority": "urgent", "issue_type": "bug", "customer_domain": "acmecorp.com", "summary": "Dashboard mobile app crashes on iOS", "suggested_assignee": "eng-mobile" }, "slack_notification_sent": true, "processing_time_ms": 3200 } ``` ## API 端点 ### POST `/webhook/email` **摄取原始邮件并创建 Linear issue** **请求头：** ``` X-Webhook-Token: {SMTP_SECRET_TOKEN} Content-Type: application/json ``` **请求体：** ``` { "from": "customer@example.com", "subject": "Issue title", "body": "Email body text", "html_body": "

HTML version (optional)

", "timestamp": "2024-01-15T10:00:00Z", "attachments": [ { "filename": "screenshot.png", "content_base64": "iVBORw0KGgoAAAANS...", "mime_type": "image/png" } ] } ``` **响应（201 Created）：** ``` { "status": "success|error", "linear_issue_id": "INB-123", "linear_issue_url": "string", "triage_result": { "priority": "urgent|high|normal|low", "issue_type": "bug|feature|support|billing", "customer_domain": "string", "customer_name": "string (optional)", "summary": "string", "suggested_assignee": "string (optional)" }, "slack_notification_sent": boolean, "error": "string (if status='error')" } ``` ### POST `/api/triage/override/{issue_id}` **手动覆盖 AI 分拣决策** **请求头：** ``` X-API-Key: {LINEAR_API_KEY} Content-Type: application/json ``` **请求体：** ``` { "priority": "high", "issue_type": "bug", "notes": "Manually corrected from 'low' due to context" } ``` **响应（200 OK）：** ``` { "status": "updated", "triage_record_id": "uuid", "changes": { "priority": {"old": "normal", "new": "high"} } } ``` ### GET `/api/triage/history` **检索分拣历史记录和指标** **查询参数：** - `limit=50`（默认） - `offset=0` - `priority_filter=urgent|high|normal|low`（可选） - `date_from=2024-01-01`（可选） - `date_to=2024-01-31`（可选） **响应（200 OK）：** ``` { "total_processed": 342, "results": [ { "id": "uuid", "email_from": "customer@example.com", "linear_issue_id": "INB-234", "priority": "high", "issue_type": "bug", "created_at": "2024-01-15T09:30:00Z", "processing_time_ms": 3200, "model_confidence": 0.94 } ], "statistics": { "avg_processing_time_ms": 2800, "priority_distribution": { "urgent": 15, "high": 87, "normal": 198, "low": 42 }, "issue_type_distribution": { "bug": 124, "feature": 56, "support": 142, "billing": 20 } } } ``` ### GET `/health` **服务健康检查** **响应（200 OK）：** ``` { "status": "healthy", "timestamp": "2024-01-15T10:00:00Z", "dependencies": { "anthropic": "ok", "linear": "ok", "slack": "ok", "database": "ok" } } ``` ## 配置 ### 环境变量 | 变量 | 必需 | 默认值 | 描述 | |----------|----------|---------|-------------| | `ANTHROPIC_API_KEY` | ✓ | — | 来自 Anthropic Console 的 Claude API key | | `LINEAR_API_KEY` | ✓ | — | 来自 Settings > API 的 Linear API token | | `LINEAR_TEAM_ID` | ✓ | — | Linear team slug（例如，'acme'） | | `LINEAR_PROJECT_ID` | | `INB` | 用于新 issue 的 Linear 项目 key | | `LINEAR_DEFAULT_STATUS` | | `backlog` | 初始 issue 状态（backlog/todo/in_progress） | | `SMTP_SECRET_TOKEN` | ✓ | — | 用于 webhook 身份验证的 secret token | | `SLACK_WEBHOOK_URL` | | — | Slack webhook URL（留空则禁用） | | `ENABLE_SLACK_NOTIFICATIONS` | | `true` | 为 urgent/high 发布通知 | | `ENABLE_AUTO_ASSIGN` | | `false` | 根据问题类型自动指派 | | `TRIAGE_MODEL` | | `claude-3-5-sonnet-20241022` | Claude 模型（使用 3-opus-20250219 获得最高准确度） | | `DATABASE_URL` | | `sqlite:///./triage.db` | PostgreSQL 或 SQLite 连接字符串 | | `GMAIL_CREDENTIALS_PATH` | | `./credentials.json` | Gmail API 凭据路径（如果使用 Gmail） | | `EMAIL_DOMAIN` | | — | 您的邮件域（用于 reply-to 头） | | `LOG_LEVEL` | | `INFO` | 日志记录级别（DEBUG/INFO/WARNING/ERROR） | ### Pydantic AI 配置 编辑 `app/config.py` 进行自定义： ``` # Claude model 设置 TRIAGE_MODEL = "claude-3-5-sonnet-20241022" # Change to claude-3-opus-20250219 for higher accuracy # Triage 分类阈值 PRIORITY_KEYWORDS = { "urgent": ["critical", "down", "broken", "asap", "emergency"], "high": ["bug", "broken", "failing", "urgent"], "normal": ["feature", "improve"], "low": ["typo", "minor", "nice-to-have"] } # Linear 字段映射 LINEAR_PRIORITY_MAP = { "urgent": 4, # Urgent in Linear "high": 3, "normal": 2, "low": 1 } ``` ## 自定义 ### 更改分拣提示词 编辑 `app/agents/triage_agent.py`： ``` TRIAGE_SYSTEM_PROMPT = """You are an expert customer support triage system... Analyze the email and extract: 1. Priority (urgent/high/normal/low) - consider customer tone, service impact, frequency 2. Issue type (bug/feature/support/billing) - classify by nature 3. Customer identifier - extract domain or company name 4. Concise summary - max 10 words 5. Suggested team - based on issue type """ ``` ### 添加自定义 Issue 字段 在 `app/models/triage.py` 中，扩展 `TriageResult`： ``` class TriageResult(BaseModel): priority: str issue_type: str customer_domain: str summary: str custom_field_1: str | None = None # Add your field ``` 然后更新 Claude 提示词以提取它，并在 `app/integrations/linear.py` 中更新 Linear 创建逻辑： ``` custom_field_id = "LIN_CUSTOM_1" issue_data["fieldValues"].append({ "fieldId": custom_field_id, "value": triage_result.custom_field_1 }) ``` ### 将 Issue 路由到不同项目 在 `app/integrations/linear.py` 中，修改项目选择： ``` def get_target_project(triage_result: TriageResult) -> str: if triage_result.issue_type == "billing": return "BIL" # Billing project elif triage_result.customer_domain == "enterprise.com": return "ENT" # Enterprise project return settings.LINEAR_PROJECT_ID ``` ### 自定义 Slack 消息 在 `app/integrations/slack.py` 中，编辑 Slack payload： ``` blocks = [ { "type": "section", "text": { "type": "mrkdwn", "text": f"🔴 *URGENT: {triage_result.summary}*\nCustomer: {triage_result.customer_domain}\n<{issue_url}|View in Linear>" } } ] ``` ### 使用不同的 Claude 模型 获得**更高的准确度**（更慢 + 更贵）： ``` TRIAGE_MODEL=claude-3-opus-20250219 python -m uvicorn app.main:app ``` **更低成本**（更快）： ``` TRIAGE_MODEL=claude-3-5-haiku-20241022 python -m uvicorn app.main:app ``` ### 添加数据库持久化 从 SQLite 切换到 PostgreSQL： ``` pip install psycopg2-binary export DATABASE_URL=postgresql://user:password@localhost:5432/triage python -m alembic upgrade head ``` ## 测试 ### 运行单元测试 ``` pytest tests/ -v ``` ### 本地测试分拣 Agent ``` python -m app.agents.triage_agent --email-from "customer@example.com" --subject "Payment failed" --body "We can't process payments today" ``` ### 模拟邮件 Webhook ``` python scripts/test_email_webhook.py ``` ## 部署 ### Docker Compose ``` docker-compose up -d ``` 查看 `docker-compose.yml` 了解生产环境配置（PostgreSQL、环境变量）。 ### Heroku ``` git push heroku main heroku config:set ANTHROPIC_API_KEY=sk-ant-... heroku config:set LINEAR_API_KEY=lin_api_... ``` ### AWS Lambda ``` pip install aws-wsgi # 有关 container image 设置，请参见 Dockerfile.lambda ``` ## 故障排除 **"Invalid Linear API Key"** - 在 [Settings > API > Personal API Keys](https://linear.app/settings/api) 中验证 token - 确保 token 具有 `read` 和 `write` 权限 **"Claude rate limit exceeded"** - 升级 Anthropic 套餐或实现请求队列 - 在高峰时段批量处理邮件 **"Slack notification not sent"** - 验证 `SLACK_WEBHOOK_URL` 已设置且有效 - 检查 Slack 工作区 webhook 权限 - 设置 `ENABLE_SLACK_NOTIFICATIONS=false` 以跳过错误 **"Database connection error"** - 对于 SQLite：确保 `triage.db` 目录可写 - 对于 PostgreSQL：验证主机/端口/凭据 - 运行 `python -c "from app.db import init_db; init_db()"` 以重新初始化 ## 性能指标 在 Sonnet 3.5 上的典型性能： - **邮件解析：** 50ms - **Claude 分拣：** 2-4s（网络 + 推理） - **Linear issue 创建：** 300-800ms - **Slack 通知：** 200-500ms - **端到端总计：** 2.5-6s ## 许可证 MIT 许可证 — 有关详细信息，请参阅 LICENSE 文件。 基于 [Pydantic AI](https://github.com/pydantic/pydantic-ai)、[FastAPI](https://fastapi.tiangolo.com/) 和 [Linear API](https://linear.app/docs) 构建。

标签：AI分发, AV绕过, FastAPI, Linear, Slack机器人, Webhook, 力导向图, 工单系统, 测试用例, 网络研究, 请求拦截, 逆向工具