企业官网建设流程全解析

PostHog AI 平台扩展实战：基于 MaxTool 与 Taxonomy Agent 的完整接入指南

【免费下载链接】posthog:hedgehog: PostHog is the leading platform for building self-driving products. Our developer tools – AI observability, analytics, session replay, flags, experiments, error tracking, logs, and more – capture all the context agents need to diagnose problems, uncover opportunities, and ship fixes. Steer it all from Slack, web, desktop, or the MCP.项目地址: https://gitcode.com/GitHub_Trending/po/posthog

导读

本文围绕 PostHog 开源仓库中ee/hogai（PostHog AI 平台）的官方扩展文档，系统讲解两条核心扩展路径：MaxTool（让 AI Agent 在前后端执行任意操作的产品工具，如执行 SQL、增删改查仪表盘）与Taxonomy Agent（浏览团队事件/属性分类体系的 RAG 式小智能体），并进一步覆盖查询类型扩展、访问控制与危险操作审批机制。读完本文，你将掌握从后端定义、前端挂载、注册元数据到调试迭代、权限收敛的完整 MaxTool 开发闭环，以及将 Taxonomy Agent 接入 MaxTool 的完整代码范式，全部内容均可在本仓库源码中逐一验证。

一、PostHog AI 与 MaxTool：一个"让 Agent 操作你的产品"的扩展框架

ee/hogai目录承载着 PostHog AI 平台及其核心能力。其中面向产品团队的核心扩展接口是MaxTool：借助 MaxTool API，你可以让 AI Agent 在你的产品中"做任何事"——既执行后端动作，也控制前端 UI。

MaxTool 的设计遵循前后端分离的两段式结构：

• 后端定义（一个 Python 类）：包含工具的元数据（工具是什么、如何用、何时用、接受哪些参数）——这些元数据最终会注入 LLM 的上下文；同时包含工具的实际实现逻辑。
• 前端挂载点（一个 React 组件）：使工具"可用"——只有被自动化 UI 存在时，工具才能被调用。

一个工具内部还可以再包含一次 LLM 调用：基于"根节点传入的参数 + 前端传入的上下文"，针对该工具的任务定制一段 prompt 让 LLM 执行。开发 MaxTool 需要配置相应的环境变量（API Key），仓库中相关说明在ee/hogai/README.md。

二、定义一个新的 MaxTool（后端部分）

2.1 文件约定与自动发现

按约定创建产品的max_tools.py文件（如不存在则新建）：

products/<your product>/backend/max_tools.py

遵循该约定的max_tools.py会被系统自动发现并加载。从源码看，自动发现逻辑位于ee/hogai/registry.py（_import_max_tools负责导入全部已注册的 MaxTool，ee/hogai/test/test_tool.py中test_all_tools_have_access_control_or_are_exempt正是通过它遍历所有工具做访问控制校验）。

2.2 工具类骨架：从 Args Schema 到_arun_impl

在max_tools.py中定义一个继承自MaxTool的工具类。以下是官方文档给出的完整模板：

from pydantic import BaseModel, Field from ee.hogai.llm import MaxChatOpenAI from ee.hogai.tool import MaxTool # Define your tool's arguments schema class YourToolArgs(BaseModel): parameter_name: str = Field(description="Description of the parameter") class YourToolOutput(BaseModel): result_data: int class YourTool(MaxTool): name: str = "your_tool_name" # Must match a value in AssistantTool enum description: str = "What this tool does" context_prompt_template: str = "Context about the tool state: {context_var}" args_schema: type[BaseModel] = YourToolArgs async def _arun_impl(self, parameter_name: str) -> tuple[str, YourToolOutput]: # Implement tool logic here # Access context with self.context (must have context_var from template) # If you use Django's ORM, ensure you utilize its asynchronous capabilities. # Optional: Use LLM to process inputs or generate structured outputs model = MaxChatOpenAI(model="gpt-4o", temperature=0.2).with_structured_output(YourToolOutput).with_retry() response = model.ainvoke({"question": "What is PostHog?"}) # Process and return results as (message, structured_data) return "Tool execution completed", response

对照ee/hogai/tool.py的MaxTool基类，可以确认以下几个关键契约：

• 返回值必须是二元组(content, artifact)：_arun_impl返回tuple[str, Any]，其中 artifact 会成为ui_payload传给前端。基类强制response_format = "content_and_artifact"（见 tool.py 类属性定义）。
• 异步优先：基类中_run_impl已被标记为 DEPRECATED，_arun_impl才是标准入口；_run/_arun会自动先做资源级权限检查，再执行上下文注入与危险操作审批（见下文）。
• 子类命名约束：__init_subclass__强制子类类名必须以Tool结尾，否则抛出ValueError("The name of a MaxTool subclass must end with 'Tool', for clarity")。
• 工具名强校验：name必须是AssistantTool枚举的合法值，否则会在类定义阶段报错（提示去schema-assistant-messages.ts修正或执行pnpm schema:build）。
• 注册即自动发生：__init_subclass__会把工具注册进CONTEXTUAL_TOOL_NAME_TO_TOOL注册表（CONTEXTUAL_TOOL_NAME_TO_TOOL[accepted_name] = cls）。

context_prompt_template的作用是把工具的"状态上下文"以占位符形式注入根节点的上下文消息，从而强引导根节点决定"何时、是否"使用该工具。其底层实现（format_context_prompt_injection）只替换{合法标识符}形式的占位符，{{/}}作为字面花括号转义保留，缺失的 key 会被替换为None——这些行为在 test_tool.py 的TestMaxTool中都有对应测试（包括模板中含 Hog/JS 代码块fun onEvent(event) { ... }时不会被误解析的用例）。

2.3 在工具内部使用 LLM：MaxChatOpenAI与MaxChatAnthropic

ee/hogai/llm.py为 LangChain 的 OpenAI/Anthropic 模型提供了 PostHog 定制的子类：

• 自动注入项目/组织/用户上下文：每次调用都会把项目名、时区、当前时间、组织名、用户名/邮箱、部署区域等信息作为系统提示词的末尾注入（PROJECT_ORG_USER_CONTEXT_PROMPT），包括 App 内部 URL 必须使用根相对路径等约束。可通过inject_context=False关闭。
• 自动重试：默认max_retries = 3，stream_usage = True。
• 计费标记：billable=True时，本次生成会被标记为$ai_billable计入 AI 计费 credit；工具级与 workflow 级（如 impersonation）可叠加控制，未计费时会累加posthog_ai_billing_skipped_total指标。
• 代理绕行：MaxChatAnthropic支持bypass_proxy=True以绕过 egress 代理（Smokescreen），专供私有 LLM gateway 使用。

2.4 注册工具名到AssistantTool联合类型

把你的工具名加入frontend/src/queries/schema/schema-assistant-messages.ts中的AssistantTool联合类型（schema-assistant-messages.ts），然后运行：

pnpm schema:build

AssistantTool是一个字符串字面量联合类型，目前包含search_session_recordings、execute_sql、upsert_dashboard、read_taxonomy、filter_session_recordings、create_insight、call_mcp_server等几十个工具名。它是前后端强一致性的"契约层"——后端工具名的合法值由它约束，前端挂载的name也由它约束。

2.5 定义前端工具元数据：TOOL_DEFINITIONS

在frontend/src/scenes/max/max-constants.tsx的TOOL_DEFINITIONS中补充工具元数据：

export const TOOL_DEFINITIONS: ... = { // ... existing tools ... your_tool_name: { name: 'Do something', description: 'Do something to blah blah', product: Scene.YourProduct, // or null for the rare global tool flag: FEATURE_FLAGS.YOUR_FLAG, // optional indication that this is flagged }, }

该元数据既用于场景 UI 展示"这个能力可用"，也用于 Max 面板向用户解释工具能力。以真实的search_session_recordings为例（max-constants.tsx）：

search_session_recordings: { name: 'Search recordings', description: 'Search recordings quickly', product: Scene.Replay, icon: iconForType('session_replay'), displayFormatter: (toolCall) => { if (toolCall.status === 'completed') { return 'Searched recordings' } return 'Searching recordings...' }, },

2.6 仓库内置示例工具

ee/hogai/tools目录下是官方示例集合，文档点名了两个：

• execute_sql（tool.py）：SQL 生成与执行。其create_tool_class工厂方法会用 SQL 表达式文档、支持函数/聚合文档动态拼装系统提示词（EXECUTE_SQL_SYSTEM_PROMPT）；_arun_impl支持filters（HogQLFilters）、viz_title、viz_description、display（图表类型）、chart_settings等参数，外部数据连接（connection_id）存在时会把查询标记connectionId并跳过本地 ClickHouse 校验，交由 runner 按连接 schema 校验。
• upsert_dashboard（tool.py）：创建与编辑仪表盘。参数用action判别联合区分create/update；它声明了资源级权限[("dashboard", "editor")]，并把"更新会删除已有 insight"的操作用is_dangerous_operation标记为危险操作，走用户审批流（见第五节）。

三、在前端挂载工具：MaxTool组件

3.1 组件用法

使用MaxTool组件包裹能从 AI 协助中受益的 UI 元素（组件实现见 MaxTool.tsx）：

import { MaxTool } from 'scenes/max/MaxTool' function YourComponent() { return ( <MaxTool name="your_tool_name" // Must match backend tool name - enforced by the AssistantTool enum displayName="Human-friendly name" context={{ // Context data passed to backend - can be empty if there truly is no context context_var: relevantData, }} callback={(toolOutput) => { // Handle structured output from tool updateUIWithToolResults(toolOutput) }} initialMaxPrompt="Optional initial prompt for Max" onMaxOpen={() => { // Optional actions when Max panel opens }} > {/* Your UI component that will have Max assistant */} <YourUIComponent /> </MaxTool> ) }

挂载完成后，工具会自动以TOOL_DEFINITIONS元数据为基础，在场景 UI 和 Max 面板中显示为可用能力，帮助用户理解该能力。

3.2 真实挂载案例：会话录制筛选

文档推荐参考frontend/src/scenes/session-recordings/filters/RecordingsUniversalFiltersEmbed.tsx（README 中描述其挂载search_session_recordings；当前代码实际挂载的是同族的filter_session_recordings，两者都在AssistantTool联合类型中）。核心代码（RecordingsUniversalFiltersEmbed.tsx）：

<MaxTool identifier="filter_session_recordings" context={{ current_filters: filters, current_session_id: currentSessionRecordingId, }} callback={applyFilters} initialMaxPrompt="Show me recordings where " suggestions={[ 'Show recordings of people who visited signup in the last 24 hours', 'Show recordings showing user frustration', 'Show recordings of people who faced bugs', ]} onMaxOpen={() => setIsFiltersExpanded(false)} className="grow" > <LemonButton ...>...</LemonButton> <CurrentFilterIndicator /> </MaxTool>

可以看到context中传入的current_filters与current_session_id正好对应后端SearchSessionRecordingsTool的context_prompt_template占位符（"Current recordings filters are: {current_filters}.\nCurrent session ID being viewed: {current_session_id}."）。前后端上下文通过contextprop 与context_prompt_template的占位符一一对应，形成完整链路（见 max_tools.py）。

注意：MaxTool.tsx的 docstring 已标注该组件被标记为 deprecated，未来将由 context-aware AI 取代，开发新功能前建议与 team-posthog-ai 沟通——这一点属于从源码注释可确认的现状，规划新工具时值得留意。

四、迭代与调试

工具初版落地后，文档强调"test the heck out of it"：像普通用户一样把所有用法都试一遍，并持续调优四个面：

• 工具名（name）
• 工具描述（description）
• 上下文消息的 prompt（context_prompt_template）
• 前端传入的 context

开发期间获得完整可观测性的方式是使用本地 PostHog AI 可观测性面板：

http://localhost:8010/ai-observability/traces

其中每一条trace 代表提交给 Max 的一条人类消息，展示为回答该消息所执行的完整步骤序列。这在调试多步骤工具（如 Taxonomy Agent 的多次工具调用）时尤为关键。

五、访问控制：两级权限模型

MaxTool 支持资源级与对象级两级访问控制，两者在权限不足时都抛出MaxToolAccessDeniedError（定义于ee/hogai/tool_errors.py）。主访问检查逻辑位于products/access_control/backend/facade/user_access_control.py（UserAccessControl类，MaxTool.user_access_control属性即为其实例，见 tool.py）。

5.1 资源级访问控制

根据用户对某类资源的权限限制工具执行（例如：用户没有 editor 权限时禁止创建 feature flag）。在_arun_impl()被调用之前自动执行（_run/_arun会先调用_check_resource_access）。

1. 在工具中覆写get_required_resource_access()：

def get_required_resource_access(self): return [("feature_flag", "editor")] # Single resource # Or multiple: return [("dashboard", "editor"), ("insight", "viewer")]

2. 如果你的工具需要接入访问控制，把它从ee/hogai/test/test_tool.py的TOOLS_WITHOUT_ACCESS_CONTROL豁免集合中移除。

支持的资源类型见posthog/scopes.py的APIScopeObject（例如feature_flag、dashboard、insight、experiment、survey）；访问级别为none、viewer、editor、manager。

5.2 对象级访问控制

限制对特定对象实例的访问（如某个具体仪表盘或 insight）。在获取对象后调用check_object_access()：

async def _arun_impl(self, dashboard_id: str) -> tuple[str, Any]: dashboard = await Dashboard.objects.aget(id=dashboard_id) await self.check_object_access(dashboard, "editor", resource="dashboard", action="edit") # ... rest of implementation

check_object_access底层走UserAccessControl.check_access_level_for_object，资源名缺省时从obj._meta.model_name推导，用于错误信息。

5.3 豁免机制

若工具不需要访问控制（只读、不涉及受保护资源），需显式加入TOOLS_WITHOUT_ACCESS_CONTROL并注明原因。测试test_all_tools_have_access_control_or_are_exempt会强制这一纪律：所有已注册工具要么声明get_required_resource_access()返回非空列表，要么出现在豁免集合中，否则测试失败。当前豁免列表（test_tool.py）中的典型条目与理由包括：

• search、read_taxonomy、todo_write、switch_mode、manage_memories—— 不查看/修改受保护资源；
• read_data、list_data、create_notebook、finalize_plan—— 在_arun_impl内做动态/条件访问检查，或无受保护资源修改；
• diagnose_proxy—— 在_arun_impl内部显式检查OrganizationMembership.Level >= ADMIN，资源级 RBAC 无法识别成员级别。

5.4 危险操作审批（Dangerous Operation）

除访问控制外，MaxTool还内建了危险操作审批流（ee/hogai/tool.py的is_dangerous_operation/format_dangerous_operation_preview/_handle_dangerous_operation）：工具可覆写is_dangerous_operation声明某些操作需要用户批准，审批请求通过 LangGraph 的interrupt()暂停执行并返回ApprovalRequest（含proposal_id、tool_name、preview、payload）给前端；用户批准/拒绝后以ApprovalResumePayload恢复。源码中特别处理了一个安全细节：审批人修改过的参数必须写回调用方原 kwargs 引用，确保"执行的是用户批准过的操作，而不是最初请求的操作"——TestDangerousOperationBindsApprovedArguments测试验证了这一行为（用户把count=200改成count=5后，工具实际执行的是 5）。

upsert_dashboard是危险操作审批的典型实践：更新仪表盘若会删除已有 insight，则is_dangerous_operation返回True，format_dangerous_operation_preview会生成包含仪表盘名、新增/删除 insight 清单（带数量与名称）的富文本预览，供用户在审批卡片上确认。

5.5 错误体系

ee/hogai/tool_errors.py定义了分层的工具错误，test_tool.py的TestMaxToolErrorHierarchy验证了其契约：

错误摘要to_summary(max_length)会以类名: 消息格式截断输出，防止超长错误污染上下文。

六、LLM 工具的最佳实践

文档给出四条经验法则：

• 从前端提供关于当前状态的全面上下文（context 越完整，LLM 决策越准）；
• 用多样化的输入和边界情况测试；
• 保持 prompt 清晰结构化，给出显式规则；
• 允许用户既从零开始完成任务，也能对已有结果进行细化。

七、扩展新的查询类型（Query Executor 体系）

PostHog AI 可以从前端上下文读取多种查询类型（trends、funnels、retention、HogQL 查询等）。要新增查询类型支持，需要同时扩展QueryExecutor与Root node。注意：这不会扩展查询类型的生成能力，那需要与 PostHog AI 团队沟通。

7.1 更新查询执行器与格式化器（ee/hogai/context/insight/）

1. 在context/insight/format/下新增一个实现"查询结果 → AI 可读格式"的格式化类，并确保从context/insight/format/__init__.py导入导出。现有格式化器包括trends.py、funnel.py、lifecycle.py、paths.py、retention.py、stickiness.py、boxplot.py、sql.py。
2. 在context/insight/query_executor.py的_compress_results()方法中新增格式化分支：

elif isinstance(query, YourNewAssistantQuery | YourNewQuery): return YourNewResultsFormatter(query, response["results"]).format()

3. 在context/insight/prompts.py为你的查询类型添加示例 prompt（向 LLM 解释结果格式）。现有示例 prompt 包括TRENDS_EXAMPLE_PROMPT、FUNNEL_STEPS_EXAMPLE_PROMPT、FUNNEL_TIME_TO_CONVERT_EXAMPLE_PROMPT、FUNNEL_TRENDS_EXAMPLE_PROMPT、LIFECYCLE_EXAMPLE_PROMPT、PATHS_EXAMPLE_PROMPT、RETENTION_EXAMPLE_PROMPT、SQL_EXAMPLE_PROMPT、STICKINESS_EXAMPLE_PROMPT、BOX_PLOT_EXAMPLE_PROMPT，以及兜底的FALLBACK_EXAMPLE_PROMPT。
4. 更新context/insight/query_executor.py的get_example_prompt()函数以处理新类型：

if isinstance(viz_message.answer, YourNewAssistantQuery): return YOUR_NEW_EXAMPLE_PROMPT

get_example_prompt的现有实现（query_executor.py）会按查询类型分发到对应示例，funnel 还会根据funnelVizType细分（STEPS / TIME_TO_CONVERT / TRENDS），boxplot 由 trends 派生。

7.2 创建格式化器类

按现有格式化器的模式创建format/your_formatter.py：

class YourNewResultsFormatter: def __init__(self, query: YourNewQuery, results: dict, team: Optional[Team] = None, utc_now_datetime: Optional[datetime] = None): self._query = query self._results = results self._team = team self._utc_now_datetime = utc_now_datetime def format(self) -> str: # Format your query results for AI consumption # Return a string representation optimized for LLM understanding pass

7.3 添加测试

• 在test/test_query_executor.py为新查询类型添加测试用例；
• 在test/format/test_format.py为新格式化器添加测试用例；
• 测试须同时覆盖成功执行与错误处理路径。

7.4 关键设计考虑（源码印证）

• 查询执行：AssistantQueryExecutor类负责完整查询生命周期，包括异步轮询与错误处理（基于posthog.clickhouse.client.execute_async.get_query_status与ExecutionMode阻塞/非阻塞执行模式，见 query_executor.py 的导入与arun_and_format_query）；
• 结果格式化：每种查询类型需要专门的格式化器，把原始结果转为 AI 可读格式（存在NULL_MARKER、TRUNCATED_MARKER等约定标记）；
• 错误处理：自定义格式化失败时回退到原始 JSON（used_fallback标志会驱动使用FALLBACK_EXAMPLE_PROMPT）；
• 上下文感知：Root node 提供 UI 上下文（dashboards、insights、events、actions），帮助 AI 理解当前状态；
• 记忆集成：系统可访问 core memory 与 onboarding 状态提供上下文响应。

八、Taxonomy Agent：构建 RAG 式分类体系智能体

Taxonomy Agent 用于构建小型的、聚焦的、agentic RAG 式智能体：它们浏览团队的分类体系（事件 events、实体属性 entity properties、事件属性 event properties），并产出结构化答案。

8.1 快速开始：四步搭建

第 1 步：定义结构化输出（智能体必须返回的 schema）：

from pydantic import BaseModel class MaxToolTaxonomyOutput(BaseModel): # The schema that the agent should return as a response # See an example: from posthog.schema import MaxRecordingUniversalFilters

第 2 步：创建 toolkit，添加一个类型化的final_answer工具（可选：把属性输出格式改为 YAML）以及任意自定义工具，本例为hello_world：

from pydantic import BaseModel, Field from ee.hogai.chat_agent.taxonomy.toolkit import TaxonomyAgentToolkit from ee.hogai.chat_agent.taxonomy.tools import base_final_answer from posthog.models import Team class final_answer(base_final_answer[MaxToolTaxonomyOutput]): # Usually the final answer tool will be different for each max_tool based on the expected output. __doc__ = base_final_answer.__doc__ # Inherit from the base final answer or create your own. class hello_world(BaseModel): """Tool for saying hello to the user, should be used in the very beginning of the conversation. Use it before you use any other tool.""" name: str = Field(description="The name of the person to say hello to.") def hello_world_tool(name: str) -> str: return f"Hello, {name}!" class YourToolkit(TaxonomyAgentToolkit): def __init__(self, team: Team): super().__init__(team) # You must override this method if you are adding a custom tool that is only applicable to your usecase def handle_tools(self, tool_name: str, tool_input: TaxonomyTool) -> tuple[str, str]: """Override the handle_tools method to add custom tools.""" if tool_name == "hello_world": result = hello_world_tool(tool_input.arguments.name) return tool_name, result return super().handle_tools(tool_name, tool_input) def _get_custom_tools(self) -> list: return [final_answer, hello_world] # Optional: prefer YAML over XML for property lists, but not a must to override # If not overriden XML will be used def _format_properties(self, props: list[tuple[str, str | None, str | None]]) -> str: return self._format_properties_yaml(props)

底层TaxonomyAgentToolkit（toolkit.py）内置了 taxonomy 查询能力：EventTaxonomyQuery（事件分类）与ActorsPropertyTaxonomyQuery（实体属性分类）分别由EventTaxonomyQueryRunner与ActorsPropertyTaxonomyQueryRunner执行，并支持虚拟属性组（virtual_properties.py）、属性值采样、XML/YAML 两种属性格式化等能力。

第 3 步：定义循环节点与工具节点，并在图中绑定：

from langchain_core.prompts import ChatPromptTemplate from posthog.models import Team, User from ee.hogai.chat_agent.taxonomy.nodes import TaxonomyAgentNode, TaxonomyAgentToolsNode from ee.hogai.chat_agent.taxonomy.agent import TaxonomyAgent from ee.hogai.chat_agent.taxonomy.types import TaxonomyAgentState class LoopNode(TaxonomyAgentNode[TaxonomyAgentState, TaxonomyAgentState[MaxToolTaxonomyOutput]]): def __init__(self, team: Team, user: User, toolkit_class: type[YourToolkit]): super().__init__(team, user, toolkit_class=toolkit_class) def _get_system_prompt(self) -> ChatPromptTemplate: """ To allow for maximum flexibility you override the system prompt to tailor the taxonomy search agent to your needs. The taxonomy agent comes with some prepackaged default prompts. Check them here ee/hogai/chat_agent/taxonomy/prompts.py """ system = [ "Here you add your custom prompt, you can define things like taxonomy operators, filter logic, or any other instruction you need for your usecase.", *super()._get_default_system_prompts(), # You can reuse the default prompts we provide if they match your criteria ] return ChatPromptTemplate([("system", m) for m in system], template_format="mustache") class ToolsNode(TaxonomyAgentToolsNode[TaxonomyAgentState, TaxonomyAgentState[MaxToolTaxonomyOutput]]): """ This is the tool node where the tool call flow and the tool execution is handled. You can override the methods to your needs, although in most cases you shall not need to do so. """ def __init__(self, team: Team, user: User, toolkit_class: type[YourToolkit]): super().__init__(team, user, toolkit_class=toolkit_class) class YourTaxonomyGraph(TaxonomyAgent[TaxonomyAgentState, TaxonomyAgentState[MaxToolTaxonomyOutput]]): def __init__(self, team: Team, user: User, tool_call_id: str): super().__init__( team, user, tool_call_id, loop_node_class=LoopNode, tools_node_class=ToolsNode, toolkit_class=YourToolkit, )

第 4 步：调用它（通常从一个MaxTool中调用）：

graph = YourTaxonomyGraph(team=self._team, user=self._user) graph_context = { "change": "Show me recordings of users in Germany that used a mobile device while performing a payment", "output": None, "tool_progress_messages": [], **self.context, } result = await graph.compile_full_graph().ainvoke(graph_context) # Currently we support Pydantic objects or str as an output type if isinstance(result["output"], MaxToolTaxonomyOutput): content = "✅ Updated taxonomy selection" payload = result["output"] else: content = "❌ Need more info to proceed" payload = MaxToolTaxonomyOutput.model_validate(result["output"])

8.2 真实案例：会话录制筛选（products/replay/backend/max_tools.py）

products/replay/backend/max_tools.py是一个把 Taxonomy Agent 接入MaxTool的完整生产范例（max_tools.py）：

• SessionReplayFilterOptionsToolkit覆写_get_custom_tools返回类型化final_answer[MaxRecordingUniversalFilters]，并覆写_format_properties用 YAML 输出属性；
• SessionReplayFilterNode在默认系统提示词前叠加PRODUCT_DESCRIPTION_PROMPT、SESSION_REPLAY_EXAMPLES_PROMPT、FILTER_FIELDS_TAXONOMY_PROMPT、DATE_FIELDS_PROMPT等产品专属 prompt；
• SessionReplayFilterOptionsGraph将上述节点与 toolkit 绑定为完整 graph；
• SearchSessionRecordingsTool(MaxTool)的_arun_impl调用_invoke_graph：把用户请求（change）与当前筛选器 JSON 组装成 user prompt，调用graph.compile_full_graph().ainvoke(graph_context)；若输出不是MaxRecordingUniversalFilters实例，则回退使用最近一次工具调用的输入并结合当前筛选器做model_validate。

该工具同时声明资源级权限[("session_recording", "viewer")]，并定义了context_prompt_template把当前筛选器与会话 ID 注入根节点，形成"前端 context → 根节点决策 → 子 graph 执行 → 结构化输出回填 UI"的完整闭环。

九、总结

从ee/hogai/README.md及其对应的源码实现可以看到，PostHog AI 的扩展面清晰收敛为三条主线：

1. MaxTool：以"后端 Python 类 + 前端 React 挂载"的双端契约为核心，配合AssistantTool枚举、TOOL_DEFINITIONS元数据、资源级/对象级访问控制与危险操作审批，把任何产品能力封装为 AI Agent 可调用、可解释、可审计的工具；
2. 查询类型扩展：通过AssistantQueryExecutor+ 格式化器 + 示例 prompt 的三件套，让 AI 读懂任意新的分析查询结果（含失败回退 JSON 的健壮性设计）；
3. Taxonomy Agent：以TaxonomyAgent/TaxonomyAgentNode/TaxonomyAgentToolsNode/TaxonomyAgentToolkit为骨架的 agentic RAG 小智能体，可快速接入 MaxTool，实现对团队分类体系的结构化问答。

无论是为自有产品添加"让 Max 帮你干活"的能力，还是为会话录制、仪表盘、SQL 等既有工具调优提示词与权限，本文的代码范式与源码路径都可直接作为开发起点，并配合本地http://localhost:8010/ai-observability/traces做全链路 trace 调试。

【免费下载链接】posthog:hedgehog: PostHog is the leading platform for building self-driving products. Our developer tools – AI observability, analytics, session replay, flags, experiments, error tracking, logs, and more – capture all the context agents need to diagnose problems, uncover opportunities, and ship fixes. Steer it all from Slack, web, desktop, or the MCP.项目地址: https://gitcode.com/GitHub_Trending/po/posthog