用 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 QualifierSlack 销售线索筛选器讲解如何构建一个端到端的 Agentic 应用当有新成员加入公司公开 Slack 社区时自动调研其背景与所属组织评估其对公司商业产品的匹配度将分析结果发送到私有 Slack 频道并在每日固定时间发送 Top 5 线索汇总。文章会结合当前仓库中的示例源码examples/pydantic_ai_examples/slack_lead_qualifier/逐模块拆解数据模型、Agent 定义、存储、Webhook 与 Modal 部署编排读完你可以在自己的 Slack 工作区完整复现并改造这套自动化线索筛选流水线。示例概览从新成员入群到每日线索汇总这个示例解决的是一个非常典型的商业场景公司维护着一个公开的 Slack 社区每天都有新人加入其中可能混有潜在付费客户。人工逐个调研成本极高而本示例用 Pydantic AI 实现了一个三阶段自动流水线自动调研每当新成员加入社区应用收到 Slack 的team_join事件自动分析该成员的资料姓名、邮箱、职位等结合 DuckDuckGo 搜索其个人与所属组织背景评估其与公司商业产品示例中为 Pydantic Logfire的匹配度实时推送将分析结果以 Slack Block Kit 消息的形式发送到私有频道#new-slack-leads每日汇总通过定时任务每天将过去 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/模块职责关键内容models.py数据模型ProfileSlack 用户资料、Analysis分析结果含format_as_xml提示词转换与 Slack Block 渲染agent.pyAgent 定义使用openai:gpt-5.2、DuckDuckGo 搜索工具、NativeOutput结构化输出analyze_profile入口store.py分析结果存储基于modal.Dict的AnalysisStoreadd / list / clearslack.pySlack 消息发送封装chat.postMessageAPI读取SLACK_API_KEYfunctions.py业务功能process_slack_member实时处理与send_daily_summary每日汇总app.pyWebhook 入口FastAPI 端点处理 Slack Events API 的url_verification与team_joinmodal.pyModal 编排Modal App 定义、Logfire 初始化、ASGI Web 端点、定时任务、后台函数运行时分两类Web 函数FastAPI ASGI 应用常驻容器与后台/定时函数modal.Function.spawn异步执行与modal.Cron定时触发两者共享AnalysisStore实现跨运行的数据读取。环境准备Prerequisites1. Slack App需要一个有权限创建 App 的 Slack 工作区按照官方 Quickstart 创建新 App请求 Scope在Requesting scopes步骤中申请以下三个权限users.read读取用户基本信息users.read.email读取用户邮箱用于识别组织域名users.profile.read读取用户资料姓名、职位等。安装并授权安装 App 后记下 Access Token稍后存入 Modal Secret。跳过步骤 4、5订阅team_join事件需要 Webhook URL此时尚未部署可以稍后再配置。创建两个目标频道并把 App 加入其中#new-slack-leads接收每个新成员的分析结果#daily-slack-leads-summary接收每日 Top 5 汇总。这两个频道名在示例中是硬编码的定义于 functions.pyNEW_LEAD_CHANNEL #new-slack-leads DAILY_SUMMARY_CHANNEL #daily-slack-leads-summary想改频道名clone 仓库后直接修改这两个常量即可。2. Logfire Write Token注册 Logfire 账号并创建项目例如命名为slack-lead-qualifier生成 Write Token 并记下稍后存入 Modal Secret。3. OpenAI API Key在 OpenAI 平台创建 API KeyAgent 使用的模型为openai:gpt-5.2记下后同样存入 Modal Secret。4. Modal 账号与 Secrets注册 Modal 账号后按照其 Secrets 指南创建 3 个类型为 Custom 的 SecretSecret 名称Key值slackSLACK_API_KEY前面生成的 Slack Access TokenlogfireLOGFIRE_TOKEN前面生成的 Logfire Write TokenopenaiOPENAI_API_KEY前面生成的 OpenAI API Key这三个 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 可以看到本示例运行所需的额外依赖包括modal1.0.4、logfire[asyncpg,fastapi,sqlite3,httpx]3.14.1、fastapi0.117.0、httpx等。本地热运行Ephemeral App认证 Modalpython/uv-run -m modal setup以临时ephemeralModal App 方式运行CtrlC 即退出python/uv-run -m modal serve -m pydantic_ai_examples.slack_lead_qualifier.modal记下输出中Created web function web_app 后的 URL这就是你的 Webhook 端点地址。回到 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: samuelpydantic.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.pyProfile用 Pydantic 定义字段对应team_join事件里user.profile的内容前三个字段可空、email必填来源见 models.pyclass 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_tagprofile)as_prompt()借助 Pydantic AI 的format_as_xml把 Profile 序列化成 XML 字符串作为发送给模型的提示词内容让 LLM 以统一、结构化的形式读取成员资料。Analysis表示 Agent 的分析结果字段上通过 docstring 给模型提供语义约束来源见 models.pyclass 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 5summary要求模型写一句话的欢迎语说明对方是谁、我们能如何帮助。Analysis.as_slack_blocks()把分析结果渲染成 Slack Block Kit 结构两个 markdown 块成员信息行 总结并支持通过include_relevance参数决定是否在消息中显示评分来源见 models.pydef 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.pyAgent 是整个应用的大脑定义见 agent.pyagent Agent( openai:gpt-5.2, instructionsdedent( 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_missingTrue) # pyright: ignore[reportUnknownMemberType]要点以analysis.profile.email为 key值存储model_dump()后的字典读取时用model_validate还原为Analysis完整走 Pydantic 序列化闭环# pyright: ignore是因为modal未完整定义类型需要抑制静态类型检查器 pyrightPydantic AI 所有代码包括示例都会跑 pyright的告警。发送 Slack 消息slack.py封装 Slack 的chat.postMessageAPI见 slack.pyAPI_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: fBearer {API_KEY}, }, timeout5, ) response.raise_for_status() result response.json() if not result.get(ok, False): error result.get(error, Unknown error) raise Exception(fFailed 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.pylogfire.instrument(Process Slack member) async def process_slack_member(profile: Profile): analysis await analyze_profile(profile) logfire.info(Analysis, analysisanalysis) if analysis is None: return await AnalysisStore().add(analysis) await send_slack_message( NEW_LEAD_CHANNEL, [ { type: header, text: { type: plain_text, text: fNew Slack member with score {analysis.relevance}/5, }, }, { type: divider, }, *analysis.as_slack_blocks(), ], )流程调用 Agent 分析 → 若返回None信息不足直接跳过 → 否则写入存储 → 组装 header带评分 divider 分析块发送到#new-slack-leads。每日汇总send_daily_summary见 functions.pylogfire.instrument(Send daily summary) async def send_daily_summary(): analyses await AnalysisStore().list() logfire.info(Analyses, analysesanalyses) if len(analyses) 0: return sorted_analyses sorted(analyses, keylambda x: x.relevance, reverseTrue) top_analyses sorted_analyses[:5] blocks [ { type: header, text: { type: plain_text, text: fTop {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_relevanceTrue), ] ) await send_slack_message( DAILY_SUMMARY_CHANNEL, blocks, ) await AnalysisStore().clear()流程列出全部分析 → 按relevance降序排序取前 5 → 组装 header 各线索块include_relevanceTrue带上评分发送到#daily-slack-leads-summary→ 清空存储避免下次重复处理。Webhook 入口app.py用 FastAPI 定义接收 Slack Events API 的端点见 app.pyapp FastAPI() logfire.instrument_fastapi(app, capture_headersTrue) 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_codestatus.HTTP_422_UNPROCESSABLE_ENTITY)要点url_verificationSlack 配置订阅时会先发验证请求需原样回传challengeteam_join用Profile.model_validate直接从事件 payload 解析出 Profile然后调用process_slack_memberLogfire通过logfire.instrument_fastapi(app, capture_headersTrue)对 FastAPI 全量打点。这里的process_slack_member是个障眼法见 app.pydef process_slack_member(profile: Profile): from .modal import process_slack_member as _process_slack_member _process_slack_member.spawn( profile.model_dump(), logfire_ctxget_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镜像 依赖 Secretsimage modal.Image.debian_slim(python_version3.13).pip_install( pydantic, pydantic_ai_slim[openai,duckduckgo], logfire[httpx,fastapi], fastapi[standard], httpx, ) app modal.App( nameslack-lead-qualifier, imageimage, 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 与 httpxsecrets引用了前面在 Modal 控制台创建的三个 Secret。初始化 Logfiredef setup_logfire(): import logfire logfire.configure(service_nameapp.name) logfire.instrument_pydantic_ai() logfire.instrument_httpx(capture_allTrue)logfire.instrument_pydantic_ai()自动插桩 Pydantic AI 的 Agent 运行logfire.instrument_httpx(capture_allTrue)捕获所有 HTTP 请求/响应。这段不能在文件顶层执行modal.py在本地机器上运行只有modal包可用logfire等包只存在于 Modal 容器内因此必须在函数内部调用。Web 端点常驻容器应对 3 秒限制app.function(min_containers1) modal.asgi_app() # pyright: ignore[reportUnknownMemberType] def web_app(): setup_logfire() from .app import app as _app return _appapp.function()modal.asgi_app()把返回 ASGI 应用的函数发布为 Modal Web 端点。默认 Modal 按需起容器每次请求都有冷启动时间为了让 Webhook 满足 Slack 的 3 秒响应要求min_containers1让端点常驻、随时待命。这里的# pyright: ignore同样是抑制 modal 类型不完整导致的告警。定时汇总每天 8:00 UTCapp.function(schedulemodal.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(schedulemodal.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 输出 ModalWebhook / 定时 / 后台三合一编排 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创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考