開篇先聊一個很實在的問題初創公司的運營數據往往散落得到處都是。后端埋點、用戶行為、支付流水、工單反饋、灰度發布記錄各自躺在不同的平臺里。老板問“最近一周新用戶激活到底怎樣”你需要登錄四五個系統才能拼出一個大概。這不是運維工具不夠多而是“可觀測性”這件事在創業團隊里長期被簡化成了“日志文件 告警規則”。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 客戶端替換為真實模型接口讓代理具備更強的語義理解與任務拆解能力。這里的技術深度足夠你繼續鉆研但在動手之前請一定先想清楚可觀測性問題本身“業務現在怎么樣”和“為什么會這樣”這兩件事永遠比框架本身值錢。