开篇先聊一个很实在的问题初创公司的运营数据往往散落得到处都是。后端埋点、用户行为、支付流水、工单反馈、灰度发布记录各自躺在不同的平台里。老板问“最近一周新用户激活到底怎样”你需要登录四五个系统才能拼出一个大概。这不是运维工具不够多而是“可观测性”这件事在创业团队里长期被简化成了“日志文件 告警规则”。Atlas 这类“通过自建代理实现运营可观测性”的项目给了一个新思路与其持续人工调配监控面板不如让一组代理自己去采集、关联、分析并维护自身的观测体系。本文会围绕这个方向先讲清楚可观测性对初创公司的真正含义再拆解自建代理的整体架构与核心实现最后给出一套最小可运行的示例代码、常见坑点和工程落地建议。不管你是后端开发、DevOps还是产品技术负责人都能从里面找到可以直接参考的部分。1. 背景初创公司为什么需要“运营可观测性”1.1 从“日志查看”到“运营可观测”传统意义上观测主要是监控系统可用性CPU 是否打满、内存是否泄漏、接口是否在报 500。但对初创公司来说比“服务挂没挂”更重要的往往是“业务跑得顺不顺”。比如用户注册流程在哪个步骤流失最严重新功能上线后核心转化率是升还是降支付渠道失败率升高是因为第三方网关波动还是我们自己的参数传错了客户成功团队反馈的“有人反馈开不了发票”背后的共同模式是什么这些问题本质上是把技术指标和业务指标放在一起交叉分析。所谓“运营可观测性”就是把日志、链路、指标、业务事件统一抽象成可查询、可关联、可分析的数据模型让团队能随时回答“现在业务是什么样的”而不是只回答“系统现在还活着吗”。1.2 传统可观测性工具在初创场景下的“重”很多人一想到可观测性第一反应是部署 Prometheus Grafana Loki Tempo再不行上 SkyWalking、OpenTelemetry甚至直接买商业 APM。这个路线好不好好但很重。初创公司普遍面临三个问题人力不够。没有专职 SRE后端团队既要写业务又要维护指标采集和看板体系很难抽出时间做数据关联分析。数据源变化太快。产品功能频繁调整埋点字段说改就改手工维护的仪表盘很快就过期最终变成一张无人看的“僵尸看板”。运营人员拿不到数据。监控面板通常面向技术角色产品、运营、客户成功同学想查业务指标还得让研发帮写 SQL沟通成本非常高。这类问题不是“再加一个监控组件”能解决的而是需要一个更贴近业务、能够持续适应变化的观测层。1.3 自建代理self-building agents解决什么问题“self-building agents”直译是“自建代理”核心思想是代理系统不只是被动接收配置它能够根据当前观测目标、数据源变化和历史使用习惯自己调整自己的工具集、查询逻辑和告警策略。换句话说传统监控看板是“人定义指标系统展示指标”自建代理是“人提出业务问题代理自动拆解问题发现需要哪些数据然后自主构建查询、关联和分析流程”。当数据源新增或者业务口径变化时代理会重新评估已有观测任务是否仍然有效。这样做至少带来三个价值降低使用门槛运营人员可以直接用自然语言提问不需要精通 PromQL 或 SQL。降低维护成本看板和指标不是一次性配置而是由代理根据业务变化持续动态调整。提高异常发现速度代理能够主动扫描多个数据源之间的异常关联而不是等人工发现某张报表不对劲。2. 整体架构把运营数据变成“可提问”的对象要落地一个“运营可观测代理”可以先把它拆成三层架构。2.1 基础设施层事件采集与存储这一层负责收集各类运营事件。常见的数据源包括后端业务日志登录、下单、支付回调前端埋点页面 PV/UV、按钮点击、流程步骤用户行为数据Swish、Amplitude 以类工具基础设施指标接口耗时、错误率、流量部署与发布事件版本上线时间、回滚记录采集到的数据最好统一转换成“事件”模型丢进一个支持快速查询的存储中。对于初创项目SQLite 文件的组合在数据量不大的时候很够用规模上来后再考虑 ClickHouse、Doris 或 Elasticsearch。2.2 代理层自建与自组织的核心代理层是这套方案最关键的部分。它不只是一个“调用大模型回答问题的聊天机器人”而是一个具备工具调用、上下文记忆、任务拆解和自我评估能力的执行器。代理内部通常维护目标清单Goals当前需要观测的核心业务指标。工具注册表Tools能够执行的查询函数、数据分析函数、告警函数。记忆Memory最近处理过哪些请求做过哪些查询哪些结论被确认过。中断机制Interrupt当代理不确定、或者发现异常需要人工确认时把控制权交还给人类。2.3 应用层告警与自助问答应用层把代理能力暴露给用户。可以提供自然语言问答页面像聊天一样询问“最近三天注册转化率为什么下降”主动告警通道代理定时巡检发现异常后发送钉钉、飞书、企业微信或邮件通知。运营简报自动生成每天自动生成一段业务运营摘要。这三层合在一起就是一个简化版 Atlas 项目。下面我们用一个最小示例把这条路走通。3. 环境准备与版本说明3.1 运行环境本文示例代码使用 Python 编写建议使用以下环境版本可根据实际情况调整Python 3.10 及以上pip 包管理工具SQLite 3 自带数据库可选一个可以调用的大模型接口如 OpenAI 兼容接口、开源模型部署服务等如果本地没有大模型接口也可以先用“规则模板 模拟函数”的方式跑通流程验证代理框架本身后续再接真实模型。3.2 需要安装的依赖示例中只需要少量依赖pip install openai在示例中我会用openai库的兼容接口但为了避免大家因为网络或接口差异卡住核心流程会设计成“可插拔”真实调用 LLM 的地方会预留函数接口没有 Key 时可以用本地模拟函数替代。注意实际项目接入具体模型时API Key 不要提交到 Git 仓库建议通过环境变量注入。4. 核心实现一个最小可运行的“运营观察代理”这一节我们来实现一个简化版的自建代理它能做三件事从配置的数据源中读取运营事件。根据用户提问自动拆解任务并调用工具。遇到异常指标时输出告警或请求人工确认。4.1 项目结构observability-agent/ ├── main.py # 入口演示如何提问 ├── data_store.py # 事件存储层SQLite封装 ├── agent_core.py # 代理核心流程 ├── tools.py # 工具注册表 ├── seed_data.py # 生成示例运营数据 └── .env.example # 环境变量示例4.2 事件数据模型先定义运营事件的通用结构。在data_store.py中# 文件路径observability-agent/data_store.py import sqlite3 import json import time SCHEMA CREATE TABLE IF NOT EXISTS events ( id INTEGER PRIMARY KEY AUTOINCREMENT, event_type TEXT NOT NULL, event_name TEXT NOT NULL, properties TEXT NOT NULL, occurred_at REAL NOT NULL ); CREATE INDEX IF NOT EXISTS idx_event_type ON events(event_type); CREATE INDEX IF NOT EXISTS idx_event_time ON events(occurred_at); class EventStore: def __init__(self, db_pathobservability.db): self.db_path db_path self.conn sqlite3.connect(db_path) self.conn.row_factory sqlite3.Row self.conn.executescript(SCHEMA) self.conn.commit() def insert_event(self, event_type: str, event_name: str, properties: dict): self.conn.execute( INSERT INTO events (event_type, event_name, properties, occurred_at) VALUES (?, ?, ?, ?), (event_type, event_name, json.dumps(properties, ensure_asciiFalse), time.time()) ) self.conn.commit() def query_events(self, event_type: str None, since_hours: float 24.0): sql SELECT * FROM events WHERE occurred_at ? params [time.time() - since_hours * 3600] if event_type: sql AND event_type ? params.append(event_type) sql ORDER BY occurred_at DESC LIMIT 1000 rows self.conn.execute(sql, params).fetchall() return [dict(row) for row in rows] def close(self): self.conn.close()这里的关键在于把业务事件都统一为event_type event_name properties结构properties使用 JSON 存储方便存储任意业务字段。4.3 工具注册表代理的能力来源于工具。每个工具都是一个普通函数带有名称、描述、参数声明和执行函数。# 文件路径observability-agent/tools.py import time def tool_query_conversion_events(store, params): 查询指定时间窗口内的转化类事件 event_name params.get(event_name, order.success) since_hours params.get(since_hours, 72) rows store.query_events(event_nameevent_name, since_hourssince_hours) return { count: len(rows), samples: rows[:5] } def tool_query_error_rate(store, params): 计算指定服务或流程的错误率 since_hours params.get(since_hours, 24) error_events store.query_events(event_typeerror, since_hourssince_hours) total_events store.query_events(event_typebiz, since_hourssince_hours) error_rate len(error_events) / max(len(total_events), 1) return { error_count: len(error_events), total_count: len(total_events), error_rate: round(error_rate, 4) } def tool_list_recent_deploy(store, params): 查看最近部署事件 since_hours params.get(since_hours, 24) deploy_events store.query_events(event_typedeploy, since_hourssince_hours) return {deploy_count: len(deploy_events), deploys: deploy_events} def get_tool_registry(): return { query_conversion_events: { description: 查询转化类业务事件, function: tool_query_conversion_events, }, query_error_rate: { description: 计算错误率, function: tool_query_error_rate, }, list_recent_deploy: { description: 查看最近部署记录, function: tool_list_recent_deploy, }, }4.4 代理核心任务拆解与工具调用接下来是代理核心。为了让没有大模型 key 的人也能运行我将模型调用拆成LLMClient接口并且提供一个RuleBasedLLMClient做降级实现。# 文件路径observability-agent/agent_core.py import json from tools import get_tool_registry class RuleBasedLLMClient: 降级用的规则代理用于没有真实模型接口时演示流程。 生产环境建议把plan_step替换为真实大模型调用。 def plan_step(self, user_message: str, tool_names: list): # 简单关键词匹配选择要调用的工具 if 错误率 in user_message or 失败率 in user_message: return { tool: query_error_rate, params: {since_hours: 24}, reason: 用户询问错误相关指标 } if 部署 in user_message or 上线 in user_message: return { tool: list_recent_deploy, params: {since_hours: 24}, reason: 用户询问部署记录 } return { tool: query_conversion_events, params: {event_name: order.success, since_hours: 72}, reason: 默认查询转化事件 } class ObservabilityAgent: def __init__(self, store, llm_clientNone): self.store store self.registry get_tool_registry() self.llm_client llm_client or RuleBasedLLMClient() def ask(self, user_message: str): tool_names list(self.registry.keys()) plan self.llm_client.plan_step(user_message, tool_names) tool self.registry.get(plan[tool]) if not tool: return {error: f工具 {plan[tool]} 不存在, available_tools: tool_names} result tool[function](self.store, plan.get(params, {})) return { plan: plan, result: result, }这里先不追求复杂推理而是把代理的“可扩展点”定义清楚。后续接入真实模型时只需要替换plan_step的实现从关键词匹配升级为真正的语义理解与任务拆解。4.5 中断机制让代理在不确定时“停一停”有读者可能注意到上面示例中的代理是“一次调用一个结果”缺少对异常结果的人工确认环节。在真实的自建代理中中断interrupt机制很重要当代理发现指标异常、或者置信度不够时应该回到人类那里确认而不是继续执行下去。# 文件路径observability-agent/agent_core.py 追加代码 class InterruptPolicy: 定义代理执行过程中的中断判断 def __init__(self, error_rate_threshold0.05): self.error_rate_threshold error_rate_threshold def should_interrupt(self, tool_name: str, result: dict) - bool: # 如果是错误率查询且错误率超过阈值需要触发人工确认 if tool_name query_error_rate: rate result.get(error_rate, 0) if rate self.error_rate_threshold: return { need_confirm: True, reason: f错误率 {rate:.2%} 超过阈值 {self.error_rate_threshold:.2%}, suggested_action: 请人工检查近期部署或依赖服务状态, } return {need_confirm: False}这是一个非常简化的演示。实际工程中interrupt 可能意味着暂停整个 agent chain、等待用户输入、或者通过消息通道推送确认请求。4.6 生成示例数据并验证流程在seed_data.py里我们生成一些模拟业务事件用于验证整个流程# 文件路径observability-agent/seed_data.py import time import random from data_store import EventStore def seed(store: EventStore): # 模拟最近 48 小时的业务事件 now time.time() for i in range(200): # 模拟 200 笔成功订单 store.insert_event( event_typebiz, event_nameorder.success, properties{channel: random.choice([ios, android, web]), amount: random.randint(20, 500)} ) for i in range(15): # 模拟 15 条错误日志 store.insert_event( event_typeerror, event_namepayment.gateway_error, properties{gateway: random.choice([stripe, paypal, wechat]), code: TIMEOUT} ) for i in range(3): # 模拟最近 3 次部署 store.insert_event( event_typedeploy, event_nameversion.release, properties{version: f1.{(5i)}.0, actor: devops} )主程序main.py# 文件路径observability-agent/main.py from data_store import EventStore from seed_data import seed from agent_core import ObservabilityAgent, InterruptPolicy def main(): store EventStore() seed(store) agent ObservabilityAgent(store) interrupt_policy InterruptPolicy() questions [ 最近 24 小时支付失败率是多少, 最近有没有新的部署记录, ] for question in questions: print(f\n[用户] {question}) response agent.ask(question) plan response[plan] result response[result] print(f[代理] 调用工具: {plan[tool]}) print(f[代理] 原因: {plan[reason]}) print(f[代理] 结果: {result}) interrupt interrupt_policy.should_interrupt(plan[tool], result) if interrupt[need_confirm]: print(f[中断] 需要人工介入原因{interrupt[reason]}) print(f[中断] 建议{interrupt[suggested_action]}) store.close() if __name__ __main__: main()运行方式cd observability-agent python main.py预期输出会类似[用户] 最近 24 小时支付失败率是多少 [代理] 调用工具: query_error_rate [代理] 原因: 用户询问错误相关指标 [代理] 结果: {error_count: 15, total_count: 200, error_rate: 0.0698} [中断] 需要人工介入原因错误率 7.00% 超过阈值 5.00% [中断] 建议请人工检查近期部署或依赖服务状态到这里一个最小的自建代理链路已经跑通数据采集 → 工具注册 → 代理拆解 → 工具调用 → 中断干预。5. 常见问题与排查思路在实际把这样的自建代理用于运营可观测性时会碰到不少工程问题。下面整理几个高频场景。问题现象常见原因解决思路代理答非所问工具描述不清晰模型无法理解工具用途在工具注册表中补充更详细的入参说明和示例查询结果明显错误事件字段不同源指标口径不统一在数据接入层统一做字段映射和规范化告警轰炸错误率阈值设置过低或抖动较大引入弹性阈值结合历史分位数计算动态边界代理调用过慢多次串行调用工具等待时间过长对可并行工具做并发调用设置超时和重试上限大模型接口费用过高每个问题都走完整推理链路加入意图缓存、结果缓存或对高频固定问题走规则路由数据权限风险代理可查询的原始数据范围过大在工具层做行级权限过滤要求合法授权后才能访问敏感字段这里有一个非常容易踩坑的地方误把代理当作“万能问答机器人”。实际上代理的能力边界由工具集决定工具注册表里没有的能力再强的模型也编不出来。所以在新增数据源时第一件事不是改提示词而是注册对应的查询工具并测试边界条件。另一个常见问题是互操作。运营事件里的用户 ID 可能在不同系统里格式不一致有的带前缀有的是纯数字。如果不做标准化代理关联分析时会漏数据。建议在采集层就完成统一比如统一转为字符串并保留原始值。再就是中断机制的误触发。阈值设得太严会让运营人员频繁收到无意义确认设得太松又容易让异常被放过去。建议初期先记录代理的“建议结论”和人工反馈结果用这些历史数据校准阈值。6. 最佳实践与工程建议6.1 数据安全与权限边界自建代理比传统监控看板拥有更大的“行动能力”因为它可以根据目标自行构建查询。越是灵活越要控制边界。在工程上建议做到数据源访问使用只读账号禁止代理直接写入业务库。工具层实现行级和字段级脱敏例如手机号、身份证号默认打码。代理执行的关键操作发告警、发外部请求必须记录审计日志。所有数据访问都要经过合法授权符合公司内部数据安全规范避免触碰敏感用户数据。6.2 控制代理成本向高效代理靠拢大模型推理是有成本的尤其是在“自建代理”这一类需要多轮工具调用的场景。关于高效代理efficient agents业界的共识是“能不做语义推理就不做语义推理”。具体做法高频问题改为固定快捷命令不走模型规划。对相似查询使用结果缓存比如“日报”类问题每天只算一次。在规划阶段先让代理根据关键词粗筛可用的工具减少无效调用。设置 token 预算和最大工具调用次数避免代理陷入无限循环。6.3 代理的测试策略代理系统测试和普通后端测试不太一样除了单测之外还要做场景化验证。可以借鉴 Playwright 这类浏览端自动化测试工具的思路对代理的端到端流程做脚本化回归准备一组标准运营问题集每次修改代理核心后跑一遍问题集对比输出是否稳定。对于需要外部服务的查询使用模拟数据源避免测试依赖第三方状态。对中断机制单独测试构造异常数据确认代理能在正确节点暂停并请求人工确认。记录每次代理调用的完整轨迹用户问题、工具规划、工具结果、最终答案方便复盘定位问题。6.4 渐进式上线路径自建代理这种系统不建议一次性全面铺开。比较稳妥的上线路径是只读问答阶段先提供自然语言查询能力工具只读不主动发送告警。告警建议阶段代理生成告警内容但只推送给指定负责人不自动执行操作。闭环操作阶段经过充分验证后代理才可以在特定范围内执行自动操作例如自动创建工单、自动标记异常事件。在整个过程中要把“人工确认”作为默认策略。代理的置信度再高也应该保留人类审核环节尤其是涉及对外通知或数据库操作的时候。6.5 从可观测到持续改进最后谈一下工程心态。自建代理不是一次性搭建完就结束的系统它需要持续喂养新的数据源、新的工具和新的业务指标。运营团队使用越频繁代理积累的上下文越丰富出来的分析结果才会越准确。每过一段时间建议回看代理的问答日志问自己三个问题哪些问题是最常被问到的能不能固化成模板哪些查询结果是用户问了之后又手动纠正的是不是口径有问题哪些告警被忽略或关闭了是不是阈值或者提示方式不贴合实际把这个“回看-调整-再训练”的闭环建立起来才算是真正把代理用起来了。7. 总结本文从初创公司运营观测的痛点出发介绍了基于自建代理self-building agents的可观测性方案。重点内容包括可观测性不只是监控系统健康更是把业务事件变成可查询的对象代理层的核心是工具注册、任务拆解、上下文记忆和中断机制在工程落地时要特别关注数据安全、成本控制、端到端测试和渐进式上线。文中给出的最小 Python 示例从事件存储、工具注册到代理调用完整跑通了一条查询链路代码可以直接在本地修改运行。如果你正在为团队搭建类似的可观测平台建议先从小规模数据源和只读问答开始把代理的每一次调用轨迹都记录下来再根据真实反馈逐步扩展。下一步你可以尝试把 SQLite 替换为 ClickHouse 或 Elasticsearch 以支持更大数据量也可以把规则版 LLM 客户端替换为真实模型接口让代理具备更强的语义理解与任务拆解能力。这里的技术深度足够你继续钻研但在动手之前请一定先想清楚可观测性问题本身“业务现在怎么样”和“为什么会这样”这两件事永远比框架本身值钱。