用 Pydantic AI 构建 Slack Lead Qualifier：新成员自动调研、线索打分与每日汇总的完整实战

用 Pydantic AI 构建 Slack Lead Qualifier：新成员自动调研、线索打分与每日汇总的完整实战

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai

本篇技术指南基于 Pydantic AI 官方示例Slack Lead Qualifier（Slack 销售线索筛选器），讲解如何构建一个端到端的 Agentic 应用：当有新成员加入公司公开 Slack 社区时，自动调研其背景与所属组织，评估其对公司商业产品的匹配度，将分析结果发送到私有 Slack 频道，并在每日固定时间发送 Top 5 线索汇总。文章会结合当前仓库中的示例源码（examples/pydantic_ai_examples/slack_lead_qualifier/）逐模块拆解数据模型、Agent 定义、存储、Webhook 与 Modal 部署编排，读完你可以在自己的 Slack 工作区完整复现并改造这套"自动化线索筛选"流水线。

示例概览：从"新成员入群"到"每日线索汇总"

这个示例解决的是一个非常典型的商业场景：公司维护着一个公开的 Slack 社区，每天都有新人加入，其中可能混有潜在付费客户。人工逐个调研成本极高，而本示例用 Pydantic AI 实现了一个三阶段自动流水线：

1. 自动调研：每当新成员加入社区，应用收到 Slack 的team_join事件，自动分析该成员的资料（姓名、邮箱、职位等），结合 DuckDuckGo 搜索其个人与所属组织背景，评估其与公司商业产品（示例中为 Pydantic Logfire）的匹配度；
2. 实时推送：将分析结果以 Slack Block Kit 消息的形式发送到私有频道#new-slack-leads；
3. 每日汇总：通过定时任务，每天将过去 24 小时内匹配度最高的 Top 5 线索汇总发送到另一个频道#daily-slack-leads-summary。

整个应用以 Python 定义在 Modal 上：Web 端点（接收 Slack Webhook）、定时函数（每日汇总）、后台函数（异步执行耗时的 Agent 分析）都由 Modal 负责调度与运维，无需自建基础设施；同时接入 Pydantic Logfire 获得对 Webhook 与定时任务运行过程的完整可观测性。

运行效果如下两图所示：左图为发送到 Slack 的分析消息，右图为 Logfire 中对应的追踪（Trace）视图，每一步（Agent 对话、HTTP 请求/响应）都可点击展开查看详情。

架构总览：五个构建块 + 两类运行时

从源码结构看，这个示例由五个 Python 模块组成，职责单一、层次清晰（目录：examples/pydantic_ai_examples/slack_lead_qualifier/）：

运行时分两类：Web 函数（FastAPI ASGI 应用，常驻容器）与后台/定时函数（modal.Function.spawn异步执行与modal.Cron定时触发），两者共享AnalysisStore实现跨运行的数据读取。

环境准备（Prerequisites）

1. Slack App

需要一个有权限创建 App 的 Slack 工作区，按照官方 Quickstart 创建新 App：

1. 请求 Scope：在"Requesting scopes"步骤中申请以下三个权限：
   • users.read：读取用户基本信息；
   • users.read.email：读取用户邮箱（用于识别组织域名）；
   • users.profile.read：读取用户资料（姓名、职位等）。
2. 安装并授权：安装 App 后记下 Access Token，稍后存入 Modal Secret。
3. 跳过步骤 4、5：订阅team_join事件需要 Webhook URL，此时尚未部署，可以稍后再配置。

创建两个目标频道并把 App 加入其中：

• #new-slack-leads：接收每个新成员的分析结果；
• #daily-slack-leads-summary：接收每日 Top 5 汇总。

这两个频道名在示例中是硬编码的，定义于 functions.py：

NEW_LEAD_CHANNEL = '#new-slack-leads' DAILY_SUMMARY_CHANNEL = '#daily-slack-leads-summary'

想改频道名，clone 仓库后直接修改这两个常量即可。

2. Logfire Write Token

1. 注册 Logfire 账号并创建项目（例如命名为slack-lead-qualifier）；
2. 生成 Write Token 并记下，稍后存入 Modal Secret。

3. OpenAI API Key

在 OpenAI 平台创建 API Key（Agent 使用的模型为openai:gpt-5.2），记下后同样存入 Modal Secret。

4. Modal 账号与 Secrets

注册 Modal 账号后，按照其 Secrets 指南创建 3 个类型为 "Custom" 的 Secret：

这三个 Secret 会在 Modal App 定义中被引用（见下文modal.py的secrets=[...]）。

运行与部署

安装依赖

示例随pydantic-ai一起分发。已通过 pip/uv 安装pydantic-ai时，安装examples可选依赖组即可（详见 示例使用说明）：

pip/uv-add "pydantic-ai[examples]"

若 clone 了仓库，则用uv sync --extra examples同步依赖。从 examples/pyproject.toml 可以看到本示例运行所需的额外依赖，包括modal>=1.0.4、logfire[asyncpg,fastapi,sqlite3,httpx]>=3.14.1、fastapi>=0.117.0、httpx等。

本地热运行（Ephemeral App）

1. 认证 Modal：

   python/uv-run -m modal setup

2. 以临时（ephemeral）Modal App 方式运行，Ctrl+C 即退出：

   python/uv-run -m modal serve -m pydantic_ai_examples.slack_lead_qualifier.modal

3. 记下输出中Created web function web_app =>后的 URL，这就是你的 Webhook 端点地址。

4. 回到 Slack Quickstart 的第 4 步 "Configuring the app for event listening"，订阅team_join事件，并将上面的 URL 填为 Request URL。

之后每当有新成员加入工作区，你就能在运行modal serve的终端与 Logfire Live 视图中看到 Webhook 事件被处理，稍等几秒后结果出现在#new-slack-leads频道。

伪造 Slack 注册事件：可以用任意姓名/邮箱直接向 Webhook 发一个team_join事件来测试：

curl -X POST <webhook endpoint URL> \ -H "Content-Type: application/json" \ -d '{ "type": "event_callback", "event": { "type": "team_join", "user": { "profile": { "email": "samuel@pydantic.dev", "first_name": "Samuel", "last_name": "Colvin", "display_name": "Samuel Colvin" } } } }'

生产部署

需要持久运行时使用 deploy 命令：

python/uv-run -m modal deploy -m pydantic_ai_examples.slack_lead_qualifier.modal

生产部署后记得：把 Slack 事件 Request URL 更新为新的持久 URL；将 Agent 的 instructions 修改为适合自己业务场景的版本（见下文）；如果希望自动化发布，可以把代码放到独立仓库中，用 GitHub Actions 做持续部署（continuous deployment）。

代码拆解：从数据模型到 Agent

数据模型（models.py）

Profile用 Pydantic 定义，字段对应team_join事件里user.profile的内容，前三个字段可空、email必填（来源见 models.py）：

class Profile(BaseModel): first_name: str | None = None last_name: str | None = None display_name: str | None = None email: str def as_prompt(self) -> str: return format_as_xml(self, root_tag='profile')

as_prompt()借助 Pydantic AI 的format_as_xml把 Profile 序列化成 XML 字符串，作为发送给模型的提示词内容，让 LLM 以统一、结构化的形式读取成员资料。

Analysis表示 Agent 的分析结果，字段上通过 docstring 给模型提供语义约束（来源见 models.py）：

class Analysis(BaseModel): profile: Profile organization_name: str organization_domain: str job_title: str relevance: Annotated[int, Ge(1), Le(5)] """Estimated fit for Pydantic Logfire: 1 = low, 5 = high""" summary: str """One-sentence welcome note summarising who they are and how we might help"""

关键点：relevance使用Annotated[int, Ge(1), Le(5)]约束为 1~5 的整数评分，供后续排序取 Top 5；summary要求模型写一句话的欢迎语，说明对方是谁、我们能如何帮助。

Analysis.as_slack_blocks()把分析结果渲染成 Slack Block Kit 结构（两个 markdown 块：成员信息行 + 总结），并支持通过include_relevance参数决定是否在消息中显示评分（来源见 models.py）：

def as_slack_blocks(self, include_relevance: bool = False) -> list[dict[str, Any]]: profile = self.profile relevance = f'({self.relevance}/5)' if include_relevance else '' return [ { 'type': 'markdown', 'text': f'[{profile.display_name}](mailto:{profile.email}), {self.job_title} at [**{self.organization_name}**](https://{self.organization_domain}) {relevance}', }, { 'type': 'markdown', 'text': self.summary, }, ]

Agent 定义（agent.py）

Agent 是整个应用的大脑，定义见 agent.py：

agent = Agent( 'openai:gpt-5.2', instructions=dedent( """ When a new person joins our public Slack, please put together a brief snapshot so we can be most useful to them. **What to include** 1. **Who they are:** Any details about their professional role or projects (e.g. LinkedIn, GitHub, company bio). 2. **Where they work:** Name of the organisation and its domain. 3. **How we can help:** On a scale of 1–5, estimate how likely they are to benefit from **Pydantic Logfire** (our paid observability tool) based on factors such as company size, product maturity, or AI usage. *1 = probably not relevant, 5 = very strong fit.* **Our products (for context only)** • **Pydantic Validation** – Python>class AnalysisStore: @classmethod @logfire.instrument('Add analysis to store') async def add(cls, analysis: Analysis): await cls._get_store().put.aio(analysis.profile.email, analysis.model_dump()) @classmethod @logfire.instrument('List analyses from store') async def list(cls) -> list[Analysis]: return [ Analysis.model_validate(analysis) async for analysis in cls._get_store().values.aio() ] @classmethod @logfire.instrument('Clear analyses from store') async def clear(cls): await cls._get_store().clear.aio() @classmethod def _get_store(cls) -> modal.Dict: return modal.Dict.from_name('analyses', create_if_missing=True) # pyright: ignore[reportUnknownMemberType]

要点：

• 以analysis.profile.email为 key，值存储model_dump()后的字典，读取时用model_validate还原为Analysis，完整走 Pydantic 序列化闭环；
• # pyright: ignore是因为modal未完整定义类型，需要抑制静态类型检查器 pyright（Pydantic AI 所有代码包括示例都会跑 pyright）的告警。

发送 Slack 消息（slack.py）

封装 Slack 的chat.postMessageAPI，见 slack.py：

API_KEY = os.getenv('SLACK_API_KEY') assert API_KEY, 'SLACK_API_KEY is not set' @logfire.instrument('Send Slack message') async def send_slack_message(channel: str, blocks: list[dict[str, Any]]): client = httpx.AsyncClient() response = await client.post( 'https://slack.com/api/chat.postMessage', json={ 'channel': channel, 'blocks': blocks, }, headers={ 'Authorization': f'Bearer {API_KEY}', }, timeout=5, ) response.raise_for_status() result = response.json() if not result.get('ok', False): error = result.get('error', 'Unknown error') raise Exception(f'Failed to send to Slack: {error}')

使用异步 HTTP 客户端httpx.AsyncClient，通过环境变量SLACK_API_KEY（即 Modal Secretslack注入的环境变量）做 Bearer 认证，消息体全部由 Block Kitblocks驱动，并显式校验 Slack 返回的ok字段，失败即抛异常便于在 Logfire 中定位。

业务功能（functions.py）

实时处理process_slack_member（见 functions.py）：

@logfire.instrument('Process Slack member') async def process_slack_member(profile: Profile): analysis = await analyze_profile(profile) logfire.info('Analysis', analysis=analysis) if analysis is None: return await AnalysisStore().add(analysis) await send_slack_message( NEW_LEAD_CHANNEL, [ { 'type': 'header', 'text': { 'type': 'plain_text', 'text': f'New Slack member with score {analysis.relevance}/5', }, }, { 'type': 'divider', }, *analysis.as_slack_blocks(), ], )

流程：调用 Agent 分析 → 若返回None（信息不足）直接跳过 → 否则写入存储 → 组装 header（带评分）+ divider + 分析块，发送到#new-slack-leads。

每日汇总send_daily_summary（见 functions.py）：

@logfire.instrument('Send daily summary') async def send_daily_summary(): analyses = await AnalysisStore().list() logfire.info('Analyses', analyses=analyses) if len(analyses) == 0: return sorted_analyses = sorted(analyses, key=lambda x: x.relevance, reverse=True) top_analyses = sorted_analyses[:5] blocks = [ { 'type': 'header', 'text': { 'type': 'plain_text', 'text': f'Top {len(top_analyses)} new Slack members from the last 24 hours', }, }, ] for analysis in top_analyses: blocks.extend( [ { 'type': 'divider', }, *analysis.as_slack_blocks(include_relevance=True), ] ) await send_slack_message( DAILY_SUMMARY_CHANNEL, blocks, ) await AnalysisStore().clear()

流程：列出全部分析 → 按relevance降序排序取前 5 → 组装 header + 各线索块（include_relevance=True带上评分）发送到#daily-slack-leads-summary→ 清空存储，避免下次重复处理。

Webhook 入口（app.py）

用 FastAPI 定义接收 Slack Events API 的端点，见 app.py：

app = FastAPI() logfire.instrument_fastapi(app, capture_headers=True) @app.post('/') async def process_webhook(payload: dict[str, Any]) -> dict[str, Any]: if payload['type'] == 'url_verification': return {'challenge': payload['challenge']} elif ( payload['type'] == 'event_callback' and payload['event']['type'] == 'team_join' ): profile = Profile.model_validate(payload['event']['user']['profile']) process_slack_member(profile) return {'status': 'OK'} raise HTTPException(status_code=status.HTTP_422_UNPROCESSABLE_ENTITY)

要点：

• url_verification：Slack 配置订阅时会先发验证请求，需原样回传challenge；
• team_join：用Profile.model_validate直接从事件 payload 解析出 Profile，然后调用process_slack_member；
• Logfire：通过logfire.instrument_fastapi(app, capture_headers=True)对 FastAPI 全量打点。

这里的process_slack_member是个"障眼法"（见 app.py）：

def process_slack_member(profile: Profile): from .modal import process_slack_member as _process_slack_member _process_slack_member.spawn( profile.model_dump(), logfire_ctx=get_context() )

为什么不能直接调用？Slack 要求 Webhook 在 3 秒内响应，而一次完整的 Agent 分析（对话 + 网络搜索 + 发消息）显然超过 3 秒。因此这里改用modal.Function.spawn把任务投递到后台异步执行，Webhook 立即返回{'status': 'OK'}。同时通过logfire.propagate.get_context()取得当前 Logfire 上下文并随任务传递，实现分布式追踪——后台函数的执行会嵌套显示在 Webhook 请求的 trace 之下，一次请求相关的所有日志汇聚在一处。

注意函数内部才from .modal import ...：因为modal.py会导入app.py，若在模块顶层导入会产生循环导入错误。

Modal 编排（modal.py）

最后是 Modal 如何把所有组件编排成可部署应用，见 modal.py。

定义 Modal App（镜像 + 依赖 + Secrets）：

image = modal.Image.debian_slim(python_version='3.13').pip_install( 'pydantic', 'pydantic_ai_slim[openai,duckduckgo]', 'logfire[httpx,fastapi]', 'fastapi[standard]', 'httpx', ) app = modal.App( name='slack-lead-qualifier', image=image, secrets=[ modal.Secret.from_name('logfire'), modal.Secret.from_name('openai'), modal.Secret.from_name('slack'), ], )

基础镜像为 Debian + Python 3.13，安装pydantic_ai_slim[openai,duckduckgo]（带 OpenAI 与 DuckDuckGo 搜索额外依赖）、Logfire（含 httpx/fastapi 插桩）、FastAPI 与 httpx；secrets引用了前面在 Modal 控制台创建的三个 Secret。

初始化 Logfire：

def setup_logfire(): import logfire logfire.configure(service_name=app.name) logfire.instrument_pydantic_ai() logfire.instrument_httpx(capture_all=True)

logfire.instrument_pydantic_ai()自动插桩 Pydantic AI 的 Agent 运行，logfire.instrument_httpx(capture_all=True)捕获所有 HTTP 请求/响应。这段不能在文件顶层执行：modal.py在本地机器上运行，只有modal包可用；logfire等包只存在于 Modal 容器内，因此必须在函数内部调用。

Web 端点（常驻容器应对 3 秒限制）：

@app.function(min_containers=1) @modal.asgi_app() # pyright: ignore[reportUnknownMemberType] def web_app(): setup_logfire() from .app import app as _app return _app

@app.function()+@modal.asgi_app()把返回 ASGI 应用的函数发布为 Modal Web 端点。默认 Modal 按需起容器，每次请求都有冷启动时间；为了让 Webhook 满足 Slack 的 3 秒响应要求，min_containers=1让端点常驻、随时待命。这里的# pyright: ignore同样是抑制 modal 类型不完整导致的告警。

定时汇总（每天 8:00 UTC）：

@app.function(schedule=modal.Cron('0 8 * * *')) # Every day at 8am UTC async def send_daily_summary(): setup_logfire() from .functions import send_daily_summary as _send_daily_summary await _send_daily_summary()

@app.function(schedule=modal.Cron(...))定义 Cron 定时函数，每日 8 点 UTC 调用前面实现的汇总逻辑。

后台process_slack_member（对接 spawn 与分布式追踪）：

@app.function() async def process_slack_member(profile_raw: dict[str, Any], logfire_ctx: Any): setup_logfire() from logfire.propagate import attach_context from .functions import process_slack_member as _process_slack_member from .models import Profile with attach_context(logfire_ctx): profile = Profile.model_validate(profile_raw) await _process_slack_member(profile)

Web App 通过spawn调用的就是这个函数：先setup_logfire()，再用attach_context(logfire_ctx)挂载从 Webhook 请求传播过来的 Logfire 上下文，最后还原 Profile 并执行真正的业务函数，使后台执行在 Logfire 中嵌套于请求 trace 之下。

关键设计经验总结

• 用"消息驱动 + 后台执行"规避同步约束：Slack Webhook 3 秒响应限制是这类应用最常见的坑。modal.Function.spawn+ 立即返回 HTTP 200 是优雅解法，代价是引入异步一致性（分析稍后完成，观察 Logfire 即可跟踪）。
• 结构化输出 + 可空结果的组合：NativeOutput([Analysis, NoneType])让模型在信息不足时"诚实返回 None"，业务侧据此跳过，避免强编造，这是 Agent 可靠性的关键设计。
• 共享存储桥接运行时：modal.Dict让 Webhook 写入的分析与定时任务读取的分析天然打通，且 zero-ops；清空时机（汇总后 clear）保证 24 小时窗口语义。
• 循环导入的规避模式：modal.py与app.py互相依赖，通过在函数内部延迟导入打破循环。
• 可观测性内置：Logfire 同时覆盖 FastAPI、Pydantic AI、httpx 三层，再加上logfire.instrument对业务函数的打点与上下文传播，整条流水线"每一步发生了什么"都一目了然。

结语

至此，从 Slack App 配置、Modal Secrets 准备，到models.py/agent.py/store.py/slack.py/functions.py/app.py/modal.py七个模块的完整实现，再到本地modal serve热运行与modal deploy生产部署，一条完整的"新成员自动调研 → 实时线索推送 → 每日 Top 5 汇总"流水线已经闭环。这个示例最大的价值在于示范了一种可复用的模式：Pydantic AI（结构化 Agent 输出）+ Modal（Webhook / 定时 / 后台三合一编排）+ Logfire（端到端可观测性）——把这套骨架套用到你自己的销售线索、用户画像、社区运营等场景，只需要改写Agent的 instructions 和Analysis的字段定义即可。

如果希望进一步了解本示例涉及的底层能力，可以继续阅读仓库中的相关文档：Agent 定义与 Instructions、DuckDuckGo 搜索工具、Native Output 结构化输出、Logfire 集成，以及本示例的完整源码目录 examples/pydantic_ai_examples/slack_lead_qualifier/。

【免费下载链接】pydantic-aiHow Python does AI. Agents, realtime voice, image generation, embeddings. Every model, every interface, typed end to end.项目地址: https://gitcode.com/GitHub_Trending/py/pydantic-ai