crewAI 可观测性体系Langfuse/Phoenix 集成与执行链路追踪本文基于 crewAI v1.11.0介绍如何为多智能体系统建立完整的可观测性基础设施。一、为什么多智能体系统需要可观测性一个 crewAI Crew 在生产环境中运行时你关心哪些问题这次执行花了多少钱LLM API 成本哪个 Agent 最慢哪个 Task 最容易失败某个特定的输出为什么质量低——是 Prompt 问题还是工具问题这个月的成本趋势是上升还是下降这些问题用verboseTrue的控制台日志根本无法回答。你需要的是结构化的、可查询的、可可视化的可观测性系统。crewAI 基于 OpenTelemetry 标准提供了全链路追踪支持可以无缝集成 Langfuse、Phoenix、LangSmith 等主流观测平台。二、OpenTelemetry 埋点crewAI 的追踪基础crewAI 内部已经预置了完整的 OpenTelemetry 埋点涵盖追踪覆盖范围 ├── Crew 级别 │ ├── crew_kickoff开始执行 │ ├── crew_finished执行完成 │ └── crew_error执行失败 │ ├── Agent 级别 │ ├── agent_execution_started │ ├── agent_tool_call工具调用 │ ├── agent_tool_result工具返回 │ └── agent_execution_finished │ ├── Task 级别 │ ├── task_started │ ├── task_completed │ └── task_failed │ └── LLM 调用级别 ├── llm_request请求参数、Prompt ├── llm_response返回内容 └── token_usageToken 消耗统计开发者无需手动添加追踪代码只需配置目标平台即可。三、Langfuse 集成Langfuse 是目前最流行的 LLM 可观测性平台支持开源自托管和云版本。3.1 安装与配置pipinstalllangfuse# .env 或环境变量配置LANGFUSE_SECRET_KEYsk-lf-...LANGFUSE_PUBLIC_KEYpk-lf-...LANGFUSE_HOSThttps://cloud.langfuse.com# 或自托管地址3.2 在 crewAI 中启用 Langfuse方式一自动集成推荐fromlangfuse.decoratorsimportlangfuse_context,observefromcrewaiimportCrew,Agent,Task# crewAI 通过 OpenTelemetry 自动向 Langfuse 上报追踪数据# 只需要确保环境变量配置正确无需任何代码改动crewCrew(agents[researcher,analyst],tasks[task1,task2],verboseTrue)resultcrew.kickoff(inputs{topic:AI趋势})# 执行完成后Langfuse 仪表板中会自动出现本次执行的追踪记录方式二显式 Session 追踪fromlangfuseimportLangfusefromcrewaiimportCrew langfuseLangfuse()defrun_crew_with_tracing(topic:str,user_id:strNone):带完整追踪的 Crew 执行# 创建 Langfuse 会话tracelangfuse.trace(nametech_research_crew,user_iduser_id,input{topic:topic},metadata{crew_version:1.0.0,environment:os.getenv(ENVIRONMENT,production)})try:crewCrew(agents[researcher,analyst],tasks[task1,task2])resultcrew.kickoff(inputs{topic:topic})# 记录成功结果trace.update(output{result:result.raw[:500]},status_messagesuccess)returnresultexceptExceptionase:# 记录失败信息trace.update(status_messageferror:{str(e)})raisefinally:langfuse.flush()# 确保数据上报3.3 Langfuse 仪表板功能配置完成后Langfuse 提供可观测性维度 ├── Traces追踪 │ ├── 完整的执行调用链Agent → Tool → LLM │ ├── 每步的输入/输出 │ └── 时间线瀑布图 │ ├── Metrics指标 │ ├── Token 消耗输入/输出/总计 │ ├── API 成本按模型分类 │ ├── 端到端延迟 │ └── 成功/失败率 │ ├── Scores评分 │ ├── 人工评分1-5星 │ └── LLM-as-Judge 自动评分 │ └── Datasets数据集 ├── 保存高质量/低质量执行样本 └── 用于提示词优化的测试集四、PhoenixArize集成Phoenix 是 Arize AI 开源的 LLM 观测工具特别擅长 RAG 系统的评估和调试pipinstallarize-phoenix openinference-instrumentation-crewaiimportphoenixaspxfromopeninference.instrumentation.crewaiimportCrewAIInstrumentor# 启动本地 Phoenix 服务会在 localhost:6006 开启可视化界面sessionpx.launch_app()# 启用 crewAI 自动追踪CrewAIInstrumentor().instrument()# 正常运行 Crew所有数据自动发送到 PhoenixcrewCrew(agents[researcher],tasks[task])resultcrew.kickoff(inputs{topic:AI})print(f查看追踪数据{session.url})# 输出查看追踪数据http://localhost:6006Phoenix 的特色功能RAG 评估自动评估知识检索的相关性和完整性幻觉检测标记 LLM 输出中可能的幻觉内容对比分析并排比较不同 Prompt 版本的效果五、自定义指标监控除了第三方平台有时候需要将指标接入自有监控系统如 Prometheus Grafanafromprometheus_clientimportCounter,Histogram,Gauge,start_http_serverimporttime# 定义 Prometheus 指标crew_executions_totalCounter(crewai_executions_total,Total number of crew executions,[crew_name,status])crew_execution_durationHistogram(crewai_execution_duration_seconds,Duration of crew execution,[crew_name],buckets[10,30,60,120,300,600])token_usage_totalCounter(crewai_token_usage_total,Total tokens used,[crew_name,model,type]# type: input/output)# 监控装饰器defmonitored_crew(crew_name:str):defdecorator(func):defwrapper(*args,**kwargs):start_timetime.time()try:resultfunc(*args,**kwargs)crew_executions_total.labels(crew_namecrew_name,statussuccess).inc()returnresultexceptExceptionase:crew_executions_total.labels(crew_namecrew_name,statuserror).inc()raisefinally:durationtime.time()-start_time crew_execution_duration.labels(crew_namecrew_name).observe(duration)returnwrapperreturndecorator# 使用监控装饰器monitored_crew(crew_nametech_research)defrun_research_crew(topic:str):crewCrew(agents[researcher],tasks[task])resultcrew.kickoff(inputs{topic:topic})# 记录 Token 使用ifresult.token_usage:token_usage_total.labels(crew_nametech_research,modelgpt-4o,typeinput).inc(result.token_usage.prompt_tokens)token_usage_total.labels(crew_nametech_research,modelgpt-4o,typeoutput).inc(result.token_usage.completion_tokens)returnresult# 启动 Prometheus 指标服务器端口 8000start_http_server(8000)六、成本追踪与预算控制importosfromdatetimeimportdatetime,datefromcollectionsimportdefaultdictclassCostTracker:LLM API 成本追踪器# 价格表$/百万tokensPRICING{gpt-4o:{input:2.5,output:10.0},gpt-4o-mini:{input:0.15,output:0.6},claude-opus-4-6:{input:15.0,output:75.0},claude-sonnet-4-6:{input:3.0,output:15.0},}def__init__(self,daily_budget_usd:float10.0):self.daily_budgetdaily_budget_usd self.daily_costsdefaultdict(float)deftrack(self,model:str,prompt_tokens:int,completion_tokens:int)-float:计算并追踪本次调用成本pricingself.PRICING.get(model,{input:0,output:0})cost(prompt_tokens/1_000_000*pricing[input]completion_tokens/1_000_000*pricing[output])todaydate.today().isoformat()self.daily_costs[today]cost# 预算预警ifself.daily_costs[today]self.daily_budget*0.8:print(f⚠️ 成本预警今日已用 ${self.daily_costs[today]:.3f}f接近预算上限 ${self.daily_budget})returncostdefget_daily_cost(self)-float:returnself.daily_costs.get(date.today().isoformat(),0.0)defreport(self)-str:today_costself.get_daily_cost()returnf今日成本${today_cost:.4f}/ 预算${self.daily_budget:.2f}# 在 Crew 执行后追踪成本cost_trackerCostTracker(daily_budget_usd20.0)resultcrew.kickoff()ifresult.token_usage:costcost_tracker.track(modelgpt-4o,prompt_tokensresult.token_usage.prompt_tokens,completion_tokensresult.token_usage.completion_tokens)print(f本次成本${cost:.4f})print(cost_tracker.report())七、日志结构化生产级日志规范importloggingimportjsonimportsysfromdatetimeimportdatetimeclassJSONFormatter(logging.Formatter):JSON 格式化日志便于 ELK/Datadog 等日志平台解析defformat(self,record):log_entry{timestamp:datetime.utcnow().isoformat()Z,level:record.levelname,logger:record.name,message:record.getMessage(),}ifhasattr(record,crew_name):log_entry[crew_name]record.crew_nameifhasattr(record,agent_role):log_entry[agent_role]record.agent_roleifhasattr(record,task_id):log_entry[task_id]record.task_idifrecord.exc_info:log_entry[exception]self.formatException(record.exc_info)returnjson.dumps(log_entry,ensure_asciiFalse)# 配置结构化日志defsetup_logging(level:strINFO):handlerlogging.StreamHandler(sys.stdout)handler.setFormatter(JSONFormatter())root_loggerlogging.getLogger()root_logger.setLevel(getattr(logging,level))root_logger.addHandler(handler)# crewAI 内部日志logging.getLogger(crewai).setLevel(logging.WARNING)# 生产环境减少噪声setup_logging(os.getenv(LOG_LEVEL,INFO))八、小结可观测性是多智能体系统进入生产环境的入场券。没有可观测性你在生产环境中基本上是盲飞Langfuse最推荐的 LLM 可观测性平台成本追踪 完整追踪链路 评分系统Phoenix特别适合 RAG 系统的评估和调试自定义 Prometheus 指标需要集成自有监控系统时的选择结构化日志生产环境必须方便日志平台检索和告警三个核心监控指标成本每次执行花了多少钱、质量输出是否符合预期、可靠性成功率和错误分布。这三个指标决定了你的 crewAI 系统在生产中是否可持续运行。系列导航上一篇crewAI CLI 与项目结构从原型到生产的工程化规范下一篇crewAI 部署形态本地、Docker、K8s 与 Serverless 化实践基于 crewAI v1.11.0 官方文档撰写于 2026 年 3 月