在日常的 LLM 应用开发中你是否遇到过这样的困扰想要追踪和调试一个复杂的 AI 流水线却发现现有的工具要么依赖云端服务要么配置繁琐难以在本地快速上手尤其是在处理多步骤的 LLM 调用、数据转换或条件分支时缺乏直观的追踪手段导致调试效率低下问题定位困难。OpenSmith 的出现正是为了解决这一痛点。它是一个轻量级的本地工具专注于帮助开发者无侵入地追踪 LLM 流水线的执行过程所有数据均存储在本地 SQLite 数据库中无需依赖任何云端服务。本文将带你从零开始完整掌握 OpenSmith 的核心概念、安装配置、基础与高级用法并通过实战案例展示如何利用其追踪能力优化你的 LLM 应用开发流程。无论你是刚接触 LLM 应用开发的初学者还是希望提升现有项目可观测性的资深工程师都能从中获得实用的解决方案。1. OpenSmith 核心概念与价值在深入使用 OpenSmith 之前理解其设计理念和核心概念至关重要。这有助于我们更好地把握其适用场景和优势。1.1 什么是 OpenSmithOpenSmith 是一个开源的本地化 LLM 流水线追踪工具。它的核心目标是提供一个简单、高效的方式记录和分析 LLM 应用在执行过程中的每一步操作。你可以将其理解为 LLM 应用领域的“调试器”或“日志系统”但它是专门为流水线式操作设计的。与传统日志记录不同OpenSmith 采用结构化的方式存储追踪数据。每一次 LLM 调用、每一次数据转换、每一个条件判断都会被记录为一条带有丰富上下文的 trace追踪记录。这些记录不仅包含输入输出还可能包含执行时间、消耗的 Token 数量、模型参数、自定义元数据等信息。1.2 为什么需要本地化追踪当前许多 LLM 应用开发工具或平台提供了云端追踪服务。虽然功能强大但也存在一些局限性数据隐私与安全敏感的业务数据或提示词模板可能需要发送到第三方服务器。网络依赖调试过程受网络环境影响离线开发场景无法使用。成本问题云端服务通常按使用量计费长期调试成本较高。定制化限制云端服务的功能相对固定难以根据特定业务需求进行深度定制。OpenSmith 的本地化设计彻底解决了这些问题。所有追踪数据都存储在你自己的机器上你可以完全控制数据的访问权限、存储周期和处理方式。结合 SQLite 数据库它甚至可以在资源受限的环境如个人笔记本、边缘设备中稳定运行。1.3 核心组件解析一个完整的 OpenSmith 追踪体系包含以下几个核心组件Pipeline流水线代表一个完整的 LLM 应用执行流程例如一个问答系统、一个文本摘要工具或一个多步决策代理。一个 Pipeline 由多个 Step 组成。Step步骤流水线中的单个操作单元。常见的 Step 类型包括LLM 调用、条件判断、数据提取、函数执行等。Trace追踪记录每次 Pipeline 执行都会生成一条 Trace 记录它包含了本次执行的全局上下文信息如执行ID、开始时间、总耗时等。Span跨度对应于一个 Step 的执行记录。一条 Trace 包含多个 Span每个 Span 记录了对应 Step 的详细输入、输出、错误信息、耗时等。SQLite Database所有追踪数据的存储后端。OpenSmith 使用 SQLite 作为默认存储无需额外安装数据库服务。这种组件化设计使得 OpenSmith 能够清晰地表征复杂的 LLM 应用执行过程为后续的分析和调试提供了坚实的基础。2. 环境准备与安装为了确保后续实战环节的顺利进行我们需要先完成 OpenSmith 的环境搭建。本节将详细介绍从基础环境检查到完整安装的全过程。2.1 系统与环境要求OpenSmith 对运行环境的要求较为宽松但建议满足以下条件以获得最佳体验操作系统Windows 10/11, macOS 10.14, 或主流的 Linux 发行版如 Ubuntu 18.04、CentOS 7。本文示例将以 Ubuntu 22.04 和 Python 3.9 环境为主但操作逻辑在不同平台间是相通的。Python 版本Python 3.8 及以上版本。强烈建议使用虚拟环境如 venv 或 conda来管理项目依赖避免包冲突。存储空间预留至少 100MB 的可用磁盘空间用于安装依赖和存储追踪数据。实际占用取决于追踪数据量的多少。权限确保当前用户对安装目录和工作目录有读写权限。在开始安装前请打开终端Windows 用户可使用 PowerShell 或 CMD执行以下命令验证 Python 环境python --version # 或 python3 --version如果输出类似Python 3.9.18的信息说明 Python 环境已就绪。2.2 安装 OpenSmithOpenSmith 可以通过 Python 的包管理工具 pip 直接安装。目前它作为一个 Python 包发布在 PyPI 上。步骤 1创建并激活虚拟环境推荐为了避免与系统或其他项目的 Python 包发生冲突首先创建一个独立的虚拟环境。# 创建名为 opensmith-env 的虚拟环境 python -m venv opensmith-env # 激活虚拟环境 # Linux/macOS: source opensmith-env/bin/activate # Windows: opensmith-env\Scripts\activate激活后你的命令行提示符前通常会显示虚拟环境名称(opensmith-env)。步骤 2使用 pip 安装 OpenSmith在激活的虚拟环境中执行安装命令pip install opensmith这个命令会自动从 PyPI 下载 OpenSmith 及其所有依赖如 SQLite 驱动、必要的网络库等。步骤 3验证安装安装完成后可以通过以下方式验证是否成功python -c import opensmith; print(opensmith.__version__)如果输出了版本号例如0.1.0则说明安装成功。你也可以尝试运行内置的帮助命令来查看基础信息python -m opensmith --help2.3 可选依赖与工具虽然 OpenSmith 的核心功能无需额外依赖但为了提升开发体验建议安装以下工具DB Browser for SQLiteSQLite 数据库可视化工具这是一个图形化界面工具可以方便地浏览和查询 OpenSmith 生成的 SQLite 数据库文件。下载地址访问 DB Browser for SQLite 官网sqlitebrowser.org下载对应操作系统的版本。安装按照官网指引完成安装。安装后你可以直接打开.db文件查看追踪数据。Jupyter Notebook / JupyterLab如果你习惯在交互式环境中进行开发和调试Jupyter 是一个很好的选择。OpenSmith 可以在 Jupyter 中无缝使用。# 在虚拟环境中安装 Jupyter pip install jupyterlab3. 快速开始你的第一个追踪流水线理论介绍完毕现在让我们通过一个简单的示例快速上手 OpenSmith 的基本用法。这个示例将创建一个最简的 LLM 调用流水线并展示如何查看追踪结果。3.1 项目结构初始化首先创建一个新的项目目录并进入mkdir my-first-opensmith-pipeline cd my-first-opensmith-pipeline确保你处于之前创建的虚拟环境中。项目目录下我们只需要一个 Python 脚本文件。3.2 编写基础流水线代码创建一个名为simple_pipeline.py的文件内容如下# simple_pipeline.py import opensmith from opensmith.pipeline import Pipeline, Step # 1. 定义一个简单的 LLM 调用步骤模拟 class SimpleLLMStep(Step): def __init__(self, name, modelgpt-3.5-turbo): super().__init__(name) self.model model def execute(self, input_data, contextNone): # 这里是模拟的 LLM 调用逻辑 # 在实际项目中这里会替换为真实的 OpenAI、Anthropic 等 API 调用 prompt input_data.get(prompt, ) simulated_response fSimulated response from {self.model} for: {prompt} # 记录本次执行的元数据 self.record_metadata({ model_used: self.model, prompt_length: len(prompt), simulated_tokens: len(simulated_response) // 4 # 粗略模拟 token 计数 }) return {response: simulated_response} # 2. 创建并运行流水线 def main(): # 初始化 OpenSmith指定追踪数据存储路径默认为 ./traces.db opensmith.init(trace_db_path./my_traces.db) # 创建流水线 pipeline Pipeline(My First Pipeline) # 向流水线中添加步骤 llm_step SimpleLLMStep(question_answerer) pipeline.add_step(llm_step) # 准备输入数据 input_data {prompt: What is the capital of France?} # 执行流水线并自动追踪 with opensmith.trace(pipeline.name) as trace: result pipeline.run(input_data, trace_contexttrace) print(Pipeline Result:, result) if __name__ __main__: main()3.3 运行与结果分析在终端中运行这个脚本python simple_pipeline.py你会看到控制台输出类似以下内容Pipeline Result: {response: Simulated response from gpt-3.5-turbo for: What is the capital of France?}同时在当前目录下会生成一个 SQLite 数据库文件my_traces.db。这个文件包含了本次流水线执行的完整追踪记录。3.4 查看追踪数据你可以使用 DB Browser for SQLite 打开my_traces.db文件或者使用 Python 脚本查询数据。以下是使用 Python 查询的示例创建一个名为query_traces.py的文件# query_traces.py import sqlite3 # 连接到追踪数据库 conn sqlite3.connect(./my_traces.db) cursor conn.cursor() # 查看有哪些表 cursor.execute(SELECT name FROM sqlite_master WHERE typetable;) tables cursor.fetchall() print(数据库中的表:, tables) # 查询最近的追踪记录 cursor.execute(SELECT * FROM traces ORDER BY start_time DESC LIMIT 1;) latest_trace cursor.fetchone() print(\n最新的 Trace 记录:) print(latest_trace) # 查询对应的 Span 记录 cursor.execute(SELECT * FROM spans WHERE trace_id?;, (latest_trace[0],)) spans cursor.fetchall() print(\n对应的 Span 记录:) for span in spans: print(span) conn.close()运行查询脚本python query_traces.py你将看到类似以下的输出展示了 OpenSmith 记录的结构化数据数据库中的表: [(traces,), (spans,), (metadata,)] 最新的 Trace 记录: (1, My First Pipeline, 2024-01-15 10:30:00, 2024-01-15 10:30:00, 150, completed, None) 对应的 Span 记录: (1, 1, question_answerer, Step, 2024-01-15 10:30:00, 2024-01-15 10:30:00, 120, completed, {input: {prompt: What is the capital of France?}}, {response: Simulated response from gpt-3.5-turbo for: What is the capital of France?}, None)通过这个简单的例子你已经成功使用 OpenSmith 追踪了一个基本的 LLM 流水线。接下来我们将深入探讨更复杂的用法和实战场景。4. 核心功能深度解析掌握了基础用法后我们需要深入了解 OpenSmith 的各项核心功能以便在复杂场景中灵活运用。本节将详细解析关键特性及其配置方式。4.1 流水线设计与步骤类型OpenSmith 的强大之处在于能够清晰地表征复杂的 LLM 应用逻辑。一个典型的流水线可能包含多种类型的步骤基本步骤类型示例from opensmith.pipeline import Pipeline, Step import opensmith # 条件判断步骤 class ConditionalStep(Step): def execute(self, input_data, contextNone): query input_data.get(query, ) if weather in query.lower(): return {next_step: weather_agent} else: return {next_step: general_agent} # 数据转换步骤 class DataEnrichmentStep(Step): def execute(self, input_data, contextNone): user_query input_data.get(query, ) # 添加时间戳、用户上下文等丰富信息 enriched_data { original_query: user_query, timestamp: 2024-01-15 10:30:00, user_context: {user_id: 123, preferences: {}} } return enriched_data # 多步骤流水线组装 def create_complex_pipeline(): pipeline Pipeline(Customer Support Pipeline) # 添加步骤输入验证 → 意图分类 → 分支处理 → 响应生成 pipeline.add_step(DataEnrichmentStep(enrich_input)) pipeline.add_step(ConditionalStep(route_intent)) # ... 添加更多步骤 return pipeline4.2 追踪配置与自定义元数据OpenSmith 允许你精细控制追踪的详细程度并添加业务相关的自定义元数据。配置追踪详细程度import opensmith # 初始化时配置 opensmith.init( trace_db_path./support_traces.db, log_levelINFO, # 控制日志输出级别 max_trace_duration3600 # 设置追踪最大时长秒 ) # 在追踪上下文中添加自定义标签 with opensmith.trace(Support Query, tags{priority: high, user_tier: vip}) as trace: # 添加自定义元数据到追踪记录 trace.set_metadata(business_context, { department: billing, query_complexity: medium }) # 流水线执行...在步骤级别记录详细信息class DetailedLLMStep(Step): def execute(self, input_data, contextNone): # 模拟真实的 API 调用参数 api_parameters { model: gpt-4, temperature: 0.7, max_tokens: 1000 } # 记录详细的请求参数 self.record_metadata({ api_parameters: api_parameters, retry_count: 0, cache_hit: False }) # 模拟 API 调用 try: # 这里应该是真实的 API 调用代码 response self.call_llm_api(input_data[prompt], api_parameters) # 记录响应元数据 self.record_metadata({ response_time_ms: 450, tokens_used: response.get(usage, {}).get(total_tokens, 0), finish_reason: response.get(choices, [{}])[0].get(finish_reason) }) return {response: response} except Exception as e: # 记录错误信息 self.record_metadata({error: str(e)}) raise4.3 SQLite 数据库模式详解理解 OpenSmith 使用的数据库模式有助于你进行自定义查询和分析。主要表结构如下traces 表追踪记录主表CREATE TABLE traces ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, -- 流水线名称 start_time TEXT NOT NULL, -- 开始时间ISO 格式 end_time TEXT, -- 结束时间 duration_ms INTEGER, -- 总耗时毫秒 status TEXT, -- 执行状态completed, failed, etc. error_message TEXT -- 错误信息如果有 );spans 表步骤记录表CREATE TABLE spans ( id INTEGER PRIMARY KEY AUTOINCREMENT, trace_id INTEGER NOT NULL, -- 关联的 trace ID name TEXT NOT NULL, -- 步骤名称 type TEXT NOT NULL, -- 步骤类型LLM, Condition, etc. start_time TEXT NOT NULL, -- 步骤开始时间 end_time TEXT, -- 步骤结束时间 duration_ms INTEGER, -- 步骤耗时 status TEXT, -- 步骤状态 input_data TEXT, -- 输入数据JSON 格式 output_data TEXT, -- 输出数据JSON 格式 error_data TEXT, -- 错误数据JSON 格式 FOREIGN KEY (trace_id) REFERENCES traces (id) );metadata 表元数据表CREATE TABLE metadata ( id INTEGER PRIMARY KEY AUTOINCREMENT, span_id INTEGER NOT NULL, -- 关联的 span ID key TEXT NOT NULL, -- 元数据键 value TEXT, -- 元数据值JSON 格式 FOREIGN KEY (span_id) REFERENCES spans (id) );掌握这些表结构后你可以编写复杂的 SQL 查询来分析流水线性能、错误模式和使用模式。5. 实战案例构建可追踪的问答系统现在我们将通过一个完整的实战案例展示如何用 OpenSmith 构建和优化一个真实的 LLM 应用。这个案例将模拟一个多步骤的问答系统包含意图识别、信息检索和响应生成等环节。5.1 项目需求与架构设计假设我们需要构建一个智能问答系统具备以下能力理解用户问题的意图普通问答、技术支持、闲聊等根据意图选择不同的处理策略对于技术问题能够检索相关知识库生成准确、友好的回答系统架构设计如下用户输入 → 输入预处理 → 意图分类 → [技术问题] → 知识库检索 → LLM 生成回答 | [普通问答] → 直接 LLM 回答 | [闲聊] → 预设模板回答5.2 实现完整的可追踪流水线创建项目文件qa_system.py# qa_system.py import opensmith from opensmith.pipeline import Pipeline, Step import json import time from datetime import datetime # 初始化 OpenSmith opensmith.init(trace_db_path./qa_traces.db) class InputValidatorStep(Step): 输入验证步骤 def execute(self, input_data, contextNone): user_input input_data.get(query, ).strip() if not user_input: raise ValueError(Query cannot be empty) if len(user_input) 1000: raise ValueError(Query too long) self.record_metadata({ input_length: len(user_input), validation_result: passed }) return {validated_query: user_input} class IntentClassifierStep(Step): 意图分类步骤模拟 def execute(self, input_data, contextNone): query input_data[validated_query].lower() # 简单的基于关键词的意图分类 tech_keywords [error, bug, install, config, how to] chat_keywords [hello, hi, how are you, weather] intent general if any(keyword in query for keyword in tech_keywords): intent technical elif any(keyword in query for keyword in chat_keywords): intent chat self.record_metadata({ detected_intent: intent, classification_confidence: high # 模拟置信度 }) return {query: input_data[validated_query], intent: intent} class KnowledgeBaseRetrievalStep(Step): 知识库检索步骤模拟 def execute(self, input_data, contextNone): if input_data[intent] ! technical: return input_data # 非技术问题跳过检索 query input_data[query] # 模拟知识库检索 time.sleep(0.1) # 模拟网络延迟 relevant_docs [ {title: Common Installation Issues, content: Check system requirements...}, {title: Configuration Guide, content: Edit config file at /etc/app/...} ] self.record_metadata({ retrieved_docs_count: len(relevant_docs), retrieval_time_ms: 100 }) return {**input_data, retrieved_docs: relevant_docs} class ResponseGeneratorStep(Step): 响应生成步骤 def execute(self, input_data, contextNone): intent input_data[intent] query input_data[query] if intent chat: response Hello! Im an AI assistant. How can I help you today? elif intent technical: docs input_data.get(retrieved_docs, []) # 模拟基于检索结果的响应生成 response fBased on {len(docs)} relevant documents: This appears to be a technical issue. Please check the documentation. else: # 模拟普通 LLM 响应 response fI understand youre asking: {query}. Heres what I think: This is an interesting question that requires careful consideration. self.record_metadata({ response_type: intent, response_length: len(response) }) return {final_response: response} def create_qa_pipeline(): 创建问答流水线 pipeline Pipeline(QA System Pipeline) pipeline.add_step(InputValidatorStep(input_validation)) pipeline.add_step(IntentClassifierStep(intent_classification)) pipeline.add_step(KnowledgeBaseRetrievalStep(knowledge_retrieval)) pipeline.add_step(ResponseGeneratorStep(response_generation)) return pipeline def main(): 主函数测试问答系统 pipeline create_qa_pipeline() # 测试用例 test_cases [ {query: How do I fix the installation error?}, {query: Whats the weather like today?}, {query: Can you explain machine learning?}, {query: } # 空查询用于测试错误处理 ] for i, test_case in enumerate(test_cases): print(f\n 测试用例 {i1} ) print(f输入: {test_case[query]}) try: with opensmith.trace(fQA Test {i1}, tags{test_case: i1}) as trace: result pipeline.run(test_case, trace_contexttrace) print(f响应: {result.get(final_response, No response)}) except Exception as e: print(f错误: {e}) if __name__ __main__: main()5.3 运行与分析追踪数据运行问答系统python qa_system.py系统会处理多个测试用例并在qa_traces.db中生成详细的追踪记录。接下来我们编写一个分析脚本来提取有价值的洞察。创建analyze_traces.py# analyze_traces.py import sqlite3 import json from datetime import datetime def analyze_qa_performance(): 分析问答系统性能 conn sqlite3.connect(./qa_traces.db) cursor conn.cursor() print( 问答系统性能分析 \n) # 1. 总体统计 cursor.execute(SELECT COUNT(*), AVG(duration_ms) FROM traces;) total_traces, avg_duration cursor.fetchone() print(f总执行次数: {total_traces}) print(f平均执行时间: {avg_duration:.2f} ms\n) # 2. 步骤性能分析 cursor.execute( SELECT name, COUNT(*) as execution_count, AVG(duration_ms) as avg_duration, MAX(duration_ms) as max_duration, SUM(CASE WHEN status ! completed THEN 1 ELSE 0 END) as error_count FROM spans GROUP BY name ORDER BY avg_duration DESC; ) print(步骤性能分析:) print(- * 60) for step_name, count, avg_dur, max_dur, errors in cursor.fetchall(): print(f{step_name:25} | 执行: {count:3} | 平均: {avg_dur:6.1f} ms | 最大: {max_dur:5} ms | 错误: {errors}) # 3. 意图分布分析 cursor.execute( SELECT m.value, COUNT(*) FROM metadata m JOIN spans s ON m.span_id s.id WHERE m.key detected_intent AND s.name intent_classification GROUP BY m.value; ) print(f\n意图分布:) print(- * 30) for intent, count in cursor.fetchall(): print(f{intent:15} : {count} 次) # 4. 错误分析 cursor.execute( SELECT t.name, t.error_message, s.name as step_name, s.error_data FROM traces t LEFT JOIN spans s ON t.id s.trace_id AND s.status ! completed WHERE t.status ! completed LIMIT 10; ) errors cursor.fetchall() if errors: print(f\n最近错误记录:) print(- * 50) for trace_name, trace_error, step_name, step_error in errors: print(f流水线: {trace_name}) print(f步骤: {step_name}) print(f错误: {trace_error or step_error}) print() conn.close() if __name__ __main__: analyze_qa_performance()运行分析脚本python analyze_traces.py你将获得类似以下的性能报告 问答系统性能分析 总执行次数: 4 平均执行时间: 325.50 ms 步骤性能分析: --------------------------------------------------------- knowledge_retrieval | 执行: 1 | 平均: 100.0 ms | 最大: 100 ms | 错误: 0 response_generation | 执行: 3 | 平均: 15.3 ms | 最大: 20 ms | 错误: 0 intent_classification | 执行: 3 | 平均: 5.0 ms | 最大: 10 ms | 错误: 0 input_validation | 执行: 3 | 平均: 2.3 ms | 最大: 5 ms | 错误: 1 意图分布: ------------------------------ technical : 1 次 general : 2 次 chat : 1 次 最近错误记录: -------------------------------------------------- 流水线: QA Test 4 步骤: input_validation 错误: Query cannot be empty通过这样的分析你可以清晰地了解系统的性能瓶颈、错误模式和用户行为模式为优化提供数据支持。6. 高级特性与集成方案OpenSmith 除了基础追踪功能外还提供了一些高级特性可以帮助你在更复杂的场景中发挥作用。本节将介绍这些特性及其实际应用。6.1 自定义追踪存储后端虽然 SQLite 是默认选择但 OpenSmith 支持自定义存储后端。你可以实现自己的存储适配器将数据保存到其他数据库或系统中。实现自定义存储后端示例from opensmith.storage import TraceStorage import psycopg2 # 假设使用 PostgreSQL class PostgreSQLStorage(TraceStorage): def __init__(self, connection_string): self.conn psycopg2.connect(connection_string) def save_trace(self, trace_data): # 实现将 trace_data 保存到 PostgreSQL 的逻辑 with self.conn.cursor() as cursor: cursor.execute( INSERT INTO traces (name, start_time, end_time, duration_ms, status) VALUES (%s, %s, %s, %s, %s) RETURNING id , (trace_data[name], trace_data[start_time], trace_data[end_time], trace_data[duration_ms], trace_data[status])) trace_id cursor.fetchone()[0] self.conn.commit() return trace_id def save_span(self, span_data): # 实现保存 span 的逻辑 pass # 实现其他必要方法... # 使用自定义存储 custom_storage PostgreSQLStorage(postgresql://user:passlocalhost/db) opensmith.init(storage_backendcustom_storage)6.2 与现有 LLM 框架集成OpenSmith 可以轻松集成到流行的 LLM 开发框架中如 LangChain、LlamaIndex 等。LangChain 集成示例from langchain.llms import OpenAI from langchain.chains import LLMChain from langchain.prompts import PromptTemplate import opensmith from opensmith.integrations.langchain import LangChainTracer # 创建 LangChain 组件 llm OpenAI(temperature0.7) prompt PromptTemplate( input_variables[question], templateAnswer the following question: {question} ) chain LLMChain(llmllm, promptprompt) # 使用 OpenSmith 追踪器 tracer LangChainTracer() # 执行并追踪 with opensmith.trace(LangChain QA) as trace: result chain.run( What is the capital of France?, callbacks[tracer.with_trace(trace)] ) print(result)6.3 批量处理与性能优化当处理大量请求时需要考虑性能优化策略。批量追踪配置import opensmith from concurrent.futures import ThreadPoolExecutor # 配置批量处理 opensmith.init( trace_db_path./batch_traces.db, batch_size100, # 批量提交大小 flush_interval30 # 自动刷新间隔秒 ) def process_single_query(query_data): 处理单个查询 with opensmith.trace(Batch Processing) as trace: # 模拟处理逻辑 result fProcessed: {query_data} trace.set_metadata(processing_result, success) return result # 批量处理示例 def batch_process_queries(queries): with ThreadPoolExecutor(max_workers5) as executor: results list(executor.map(process_single_query, queries)) return results # 测试批量处理 queries [fquery_{i} for i in range(50)] results batch_process_queries(queries) print(f处理了 {len(results)} 个查询)6.4 追踪数据导出与分析OpenSmith 的追踪数据可以导出为各种格式便于与其他分析工具集成。数据导出示例import sqlite3 import json import pandas as pd from datetime import datetime, timedelta def export_traces_to_analysis(): 导出追踪数据用于分析 conn sqlite3.connect(./qa_traces.db) # 导出到 Pandas DataFrame traces_df pd.read_sql_query(SELECT * FROM traces, conn) spans_df pd.read_sql_query(SELECT * FROM spans, conn) metadata_df pd.read_sql_query(SELECT * FROM metadata, conn) conn.close() # 保存为 CSV traces_df.to_csv(traces_export.csv, indexFalse) spans_df.to_csv(spans_export.csv, indexFalse) # 或者保存为 JSON 格式 export_data { export_time: datetime.now().isoformat(), traces: traces_df.to_dict(records), spans: spans_df.to_dict(records) } with open(traces_export.json, w) as f: json.dump(export_data, f, indent2) print(数据导出完成) # 简单的数据分析 print(f\n数据分析摘要:) print(f总追踪记录: {len(traces_df)}) print(f总步骤记录: {len(spans_df)}) print(f成功率: {(traces_df[status] completed).mean() * 100:.1f}%) export_traces_to_analysis()7. 常见问题与解决方案在实际使用 OpenSmith 的过程中你可能会遇到一些典型问题。本节整理了常见问题及其解决方案帮助你快速排错。7.1 安装与配置问题问题 1导入 opensmith 时出现 ModuleNotFoundError错误信息ModuleNotFoundError: No module named opensmith解决方案确认虚拟环境已激活且 OpenSmith 已正确安装pip list | grep opensmith如果未找到重新安装pip install opensmith检查 Python 路径是否正确确保使用的是虚拟环境中的 Python。问题 2数据库文件权限错误错误信息sqlite3.OperationalError: unable to open database file解决方案检查当前用户对目标目录是否有写权限尝试指定绝对路径opensmith.init(trace_db_path/home/user/project/traces.db)或者使用用户主目录import os db_path os.path.expanduser(~/traces.db) opensmith.init(trace_db_pathdb_path)7.2 运行时问题问题 3流水线执行性能下降现象随着追踪数据增多系统运行变慢。解决方案定期归档旧数据# 自动清理30天前的数据 opensmith.init(trace_db_path./traces.db, retention_days30)对于高频场景考虑增加批量提交大小opensmith.init(batch_size500, flush_interval60)使用更高效的存储后端如 PostgreSQL问题 4追踪数据不完整现象某些步骤的输入输出数据没有正确记录。解决方案确保所有步骤都正确调用record_metadata方法检查数据序列化OpenSmith 使用 JSON 序列化确保所有记录的数据都是 JSON 可序列化的验证步骤执行流程确保没有异常导致追踪上下文提前结束7.3 数据查询与分析问题问题 5复杂查询性能差解决方案为常用查询字段添加索引CREATE INDEX idx_traces_start_time ON traces(start_time); CREATE INDEX idx_spans_trace_id ON spans(trace_id); CREATE INDEX idx_metadata_span_id ON metadata(span_id);使用数据库连接池避免频繁连接开销考虑定期将数据导出到分析数据库如 ClickHouse进行处理问题 6追踪数据量过大解决方案实现 OpenSmith：本地化LLM流水线追踪工具的原理与实践指南
在日常的 LLM 应用开发中你是否遇到过这样的困扰想要追踪和调试一个复杂的 AI 流水线却发现现有的工具要么依赖云端服务要么配置繁琐难以在本地快速上手尤其是在处理多步骤的 LLM 调用、数据转换或条件分支时缺乏直观的追踪手段导致调试效率低下问题定位困难。OpenSmith 的出现正是为了解决这一痛点。它是一个轻量级的本地工具专注于帮助开发者无侵入地追踪 LLM 流水线的执行过程所有数据均存储在本地 SQLite 数据库中无需依赖任何云端服务。本文将带你从零开始完整掌握 OpenSmith 的核心概念、安装配置、基础与高级用法并通过实战案例展示如何利用其追踪能力优化你的 LLM 应用开发流程。无论你是刚接触 LLM 应用开发的初学者还是希望提升现有项目可观测性的资深工程师都能从中获得实用的解决方案。1. OpenSmith 核心概念与价值在深入使用 OpenSmith 之前理解其设计理念和核心概念至关重要。这有助于我们更好地把握其适用场景和优势。1.1 什么是 OpenSmithOpenSmith 是一个开源的本地化 LLM 流水线追踪工具。它的核心目标是提供一个简单、高效的方式记录和分析 LLM 应用在执行过程中的每一步操作。你可以将其理解为 LLM 应用领域的“调试器”或“日志系统”但它是专门为流水线式操作设计的。与传统日志记录不同OpenSmith 采用结构化的方式存储追踪数据。每一次 LLM 调用、每一次数据转换、每一个条件判断都会被记录为一条带有丰富上下文的 trace追踪记录。这些记录不仅包含输入输出还可能包含执行时间、消耗的 Token 数量、模型参数、自定义元数据等信息。1.2 为什么需要本地化追踪当前许多 LLM 应用开发工具或平台提供了云端追踪服务。虽然功能强大但也存在一些局限性数据隐私与安全敏感的业务数据或提示词模板可能需要发送到第三方服务器。网络依赖调试过程受网络环境影响离线开发场景无法使用。成本问题云端服务通常按使用量计费长期调试成本较高。定制化限制云端服务的功能相对固定难以根据特定业务需求进行深度定制。OpenSmith 的本地化设计彻底解决了这些问题。所有追踪数据都存储在你自己的机器上你可以完全控制数据的访问权限、存储周期和处理方式。结合 SQLite 数据库它甚至可以在资源受限的环境如个人笔记本、边缘设备中稳定运行。1.3 核心组件解析一个完整的 OpenSmith 追踪体系包含以下几个核心组件Pipeline流水线代表一个完整的 LLM 应用执行流程例如一个问答系统、一个文本摘要工具或一个多步决策代理。一个 Pipeline 由多个 Step 组成。Step步骤流水线中的单个操作单元。常见的 Step 类型包括LLM 调用、条件判断、数据提取、函数执行等。Trace追踪记录每次 Pipeline 执行都会生成一条 Trace 记录它包含了本次执行的全局上下文信息如执行ID、开始时间、总耗时等。Span跨度对应于一个 Step 的执行记录。一条 Trace 包含多个 Span每个 Span 记录了对应 Step 的详细输入、输出、错误信息、耗时等。SQLite Database所有追踪数据的存储后端。OpenSmith 使用 SQLite 作为默认存储无需额外安装数据库服务。这种组件化设计使得 OpenSmith 能够清晰地表征复杂的 LLM 应用执行过程为后续的分析和调试提供了坚实的基础。2. 环境准备与安装为了确保后续实战环节的顺利进行我们需要先完成 OpenSmith 的环境搭建。本节将详细介绍从基础环境检查到完整安装的全过程。2.1 系统与环境要求OpenSmith 对运行环境的要求较为宽松但建议满足以下条件以获得最佳体验操作系统Windows 10/11, macOS 10.14, 或主流的 Linux 发行版如 Ubuntu 18.04、CentOS 7。本文示例将以 Ubuntu 22.04 和 Python 3.9 环境为主但操作逻辑在不同平台间是相通的。Python 版本Python 3.8 及以上版本。强烈建议使用虚拟环境如 venv 或 conda来管理项目依赖避免包冲突。存储空间预留至少 100MB 的可用磁盘空间用于安装依赖和存储追踪数据。实际占用取决于追踪数据量的多少。权限确保当前用户对安装目录和工作目录有读写权限。在开始安装前请打开终端Windows 用户可使用 PowerShell 或 CMD执行以下命令验证 Python 环境python --version # 或 python3 --version如果输出类似Python 3.9.18的信息说明 Python 环境已就绪。2.2 安装 OpenSmithOpenSmith 可以通过 Python 的包管理工具 pip 直接安装。目前它作为一个 Python 包发布在 PyPI 上。步骤 1创建并激活虚拟环境推荐为了避免与系统或其他项目的 Python 包发生冲突首先创建一个独立的虚拟环境。# 创建名为 opensmith-env 的虚拟环境 python -m venv opensmith-env # 激活虚拟环境 # Linux/macOS: source opensmith-env/bin/activate # Windows: opensmith-env\Scripts\activate激活后你的命令行提示符前通常会显示虚拟环境名称(opensmith-env)。步骤 2使用 pip 安装 OpenSmith在激活的虚拟环境中执行安装命令pip install opensmith这个命令会自动从 PyPI 下载 OpenSmith 及其所有依赖如 SQLite 驱动、必要的网络库等。步骤 3验证安装安装完成后可以通过以下方式验证是否成功python -c import opensmith; print(opensmith.__version__)如果输出了版本号例如0.1.0则说明安装成功。你也可以尝试运行内置的帮助命令来查看基础信息python -m opensmith --help2.3 可选依赖与工具虽然 OpenSmith 的核心功能无需额外依赖但为了提升开发体验建议安装以下工具DB Browser for SQLiteSQLite 数据库可视化工具这是一个图形化界面工具可以方便地浏览和查询 OpenSmith 生成的 SQLite 数据库文件。下载地址访问 DB Browser for SQLite 官网sqlitebrowser.org下载对应操作系统的版本。安装按照官网指引完成安装。安装后你可以直接打开.db文件查看追踪数据。Jupyter Notebook / JupyterLab如果你习惯在交互式环境中进行开发和调试Jupyter 是一个很好的选择。OpenSmith 可以在 Jupyter 中无缝使用。# 在虚拟环境中安装 Jupyter pip install jupyterlab3. 快速开始你的第一个追踪流水线理论介绍完毕现在让我们通过一个简单的示例快速上手 OpenSmith 的基本用法。这个示例将创建一个最简的 LLM 调用流水线并展示如何查看追踪结果。3.1 项目结构初始化首先创建一个新的项目目录并进入mkdir my-first-opensmith-pipeline cd my-first-opensmith-pipeline确保你处于之前创建的虚拟环境中。项目目录下我们只需要一个 Python 脚本文件。3.2 编写基础流水线代码创建一个名为simple_pipeline.py的文件内容如下# simple_pipeline.py import opensmith from opensmith.pipeline import Pipeline, Step # 1. 定义一个简单的 LLM 调用步骤模拟 class SimpleLLMStep(Step): def __init__(self, name, modelgpt-3.5-turbo): super().__init__(name) self.model model def execute(self, input_data, contextNone): # 这里是模拟的 LLM 调用逻辑 # 在实际项目中这里会替换为真实的 OpenAI、Anthropic 等 API 调用 prompt input_data.get(prompt, ) simulated_response fSimulated response from {self.model} for: {prompt} # 记录本次执行的元数据 self.record_metadata({ model_used: self.model, prompt_length: len(prompt), simulated_tokens: len(simulated_response) // 4 # 粗略模拟 token 计数 }) return {response: simulated_response} # 2. 创建并运行流水线 def main(): # 初始化 OpenSmith指定追踪数据存储路径默认为 ./traces.db opensmith.init(trace_db_path./my_traces.db) # 创建流水线 pipeline Pipeline(My First Pipeline) # 向流水线中添加步骤 llm_step SimpleLLMStep(question_answerer) pipeline.add_step(llm_step) # 准备输入数据 input_data {prompt: What is the capital of France?} # 执行流水线并自动追踪 with opensmith.trace(pipeline.name) as trace: result pipeline.run(input_data, trace_contexttrace) print(Pipeline Result:, result) if __name__ __main__: main()3.3 运行与结果分析在终端中运行这个脚本python simple_pipeline.py你会看到控制台输出类似以下内容Pipeline Result: {response: Simulated response from gpt-3.5-turbo for: What is the capital of France?}同时在当前目录下会生成一个 SQLite 数据库文件my_traces.db。这个文件包含了本次流水线执行的完整追踪记录。3.4 查看追踪数据你可以使用 DB Browser for SQLite 打开my_traces.db文件或者使用 Python 脚本查询数据。以下是使用 Python 查询的示例创建一个名为query_traces.py的文件# query_traces.py import sqlite3 # 连接到追踪数据库 conn sqlite3.connect(./my_traces.db) cursor conn.cursor() # 查看有哪些表 cursor.execute(SELECT name FROM sqlite_master WHERE typetable;) tables cursor.fetchall() print(数据库中的表:, tables) # 查询最近的追踪记录 cursor.execute(SELECT * FROM traces ORDER BY start_time DESC LIMIT 1;) latest_trace cursor.fetchone() print(\n最新的 Trace 记录:) print(latest_trace) # 查询对应的 Span 记录 cursor.execute(SELECT * FROM spans WHERE trace_id?;, (latest_trace[0],)) spans cursor.fetchall() print(\n对应的 Span 记录:) for span in spans: print(span) conn.close()运行查询脚本python query_traces.py你将看到类似以下的输出展示了 OpenSmith 记录的结构化数据数据库中的表: [(traces,), (spans,), (metadata,)] 最新的 Trace 记录: (1, My First Pipeline, 2024-01-15 10:30:00, 2024-01-15 10:30:00, 150, completed, None) 对应的 Span 记录: (1, 1, question_answerer, Step, 2024-01-15 10:30:00, 2024-01-15 10:30:00, 120, completed, {input: {prompt: What is the capital of France?}}, {response: Simulated response from gpt-3.5-turbo for: What is the capital of France?}, None)通过这个简单的例子你已经成功使用 OpenSmith 追踪了一个基本的 LLM 流水线。接下来我们将深入探讨更复杂的用法和实战场景。4. 核心功能深度解析掌握了基础用法后我们需要深入了解 OpenSmith 的各项核心功能以便在复杂场景中灵活运用。本节将详细解析关键特性及其配置方式。4.1 流水线设计与步骤类型OpenSmith 的强大之处在于能够清晰地表征复杂的 LLM 应用逻辑。一个典型的流水线可能包含多种类型的步骤基本步骤类型示例from opensmith.pipeline import Pipeline, Step import opensmith # 条件判断步骤 class ConditionalStep(Step): def execute(self, input_data, contextNone): query input_data.get(query, ) if weather in query.lower(): return {next_step: weather_agent} else: return {next_step: general_agent} # 数据转换步骤 class DataEnrichmentStep(Step): def execute(self, input_data, contextNone): user_query input_data.get(query, ) # 添加时间戳、用户上下文等丰富信息 enriched_data { original_query: user_query, timestamp: 2024-01-15 10:30:00, user_context: {user_id: 123, preferences: {}} } return enriched_data # 多步骤流水线组装 def create_complex_pipeline(): pipeline Pipeline(Customer Support Pipeline) # 添加步骤输入验证 → 意图分类 → 分支处理 → 响应生成 pipeline.add_step(DataEnrichmentStep(enrich_input)) pipeline.add_step(ConditionalStep(route_intent)) # ... 添加更多步骤 return pipeline4.2 追踪配置与自定义元数据OpenSmith 允许你精细控制追踪的详细程度并添加业务相关的自定义元数据。配置追踪详细程度import opensmith # 初始化时配置 opensmith.init( trace_db_path./support_traces.db, log_levelINFO, # 控制日志输出级别 max_trace_duration3600 # 设置追踪最大时长秒 ) # 在追踪上下文中添加自定义标签 with opensmith.trace(Support Query, tags{priority: high, user_tier: vip}) as trace: # 添加自定义元数据到追踪记录 trace.set_metadata(business_context, { department: billing, query_complexity: medium }) # 流水线执行...在步骤级别记录详细信息class DetailedLLMStep(Step): def execute(self, input_data, contextNone): # 模拟真实的 API 调用参数 api_parameters { model: gpt-4, temperature: 0.7, max_tokens: 1000 } # 记录详细的请求参数 self.record_metadata({ api_parameters: api_parameters, retry_count: 0, cache_hit: False }) # 模拟 API 调用 try: # 这里应该是真实的 API 调用代码 response self.call_llm_api(input_data[prompt], api_parameters) # 记录响应元数据 self.record_metadata({ response_time_ms: 450, tokens_used: response.get(usage, {}).get(total_tokens, 0), finish_reason: response.get(choices, [{}])[0].get(finish_reason) }) return {response: response} except Exception as e: # 记录错误信息 self.record_metadata({error: str(e)}) raise4.3 SQLite 数据库模式详解理解 OpenSmith 使用的数据库模式有助于你进行自定义查询和分析。主要表结构如下traces 表追踪记录主表CREATE TABLE traces ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL, -- 流水线名称 start_time TEXT NOT NULL, -- 开始时间ISO 格式 end_time TEXT, -- 结束时间 duration_ms INTEGER, -- 总耗时毫秒 status TEXT, -- 执行状态completed, failed, etc. error_message TEXT -- 错误信息如果有 );spans 表步骤记录表CREATE TABLE spans ( id INTEGER PRIMARY KEY AUTOINCREMENT, trace_id INTEGER NOT NULL, -- 关联的 trace ID name TEXT NOT NULL, -- 步骤名称 type TEXT NOT NULL, -- 步骤类型LLM, Condition, etc. start_time TEXT NOT NULL, -- 步骤开始时间 end_time TEXT, -- 步骤结束时间 duration_ms INTEGER, -- 步骤耗时 status TEXT, -- 步骤状态 input_data TEXT, -- 输入数据JSON 格式 output_data TEXT, -- 输出数据JSON 格式 error_data TEXT, -- 错误数据JSON 格式 FOREIGN KEY (trace_id) REFERENCES traces (id) );metadata 表元数据表CREATE TABLE metadata ( id INTEGER PRIMARY KEY AUTOINCREMENT, span_id INTEGER NOT NULL, -- 关联的 span ID key TEXT NOT NULL, -- 元数据键 value TEXT, -- 元数据值JSON 格式 FOREIGN KEY (span_id) REFERENCES spans (id) );掌握这些表结构后你可以编写复杂的 SQL 查询来分析流水线性能、错误模式和使用模式。5. 实战案例构建可追踪的问答系统现在我们将通过一个完整的实战案例展示如何用 OpenSmith 构建和优化一个真实的 LLM 应用。这个案例将模拟一个多步骤的问答系统包含意图识别、信息检索和响应生成等环节。5.1 项目需求与架构设计假设我们需要构建一个智能问答系统具备以下能力理解用户问题的意图普通问答、技术支持、闲聊等根据意图选择不同的处理策略对于技术问题能够检索相关知识库生成准确、友好的回答系统架构设计如下用户输入 → 输入预处理 → 意图分类 → [技术问题] → 知识库检索 → LLM 生成回答 | [普通问答] → 直接 LLM 回答 | [闲聊] → 预设模板回答5.2 实现完整的可追踪流水线创建项目文件qa_system.py# qa_system.py import opensmith from opensmith.pipeline import Pipeline, Step import json import time from datetime import datetime # 初始化 OpenSmith opensmith.init(trace_db_path./qa_traces.db) class InputValidatorStep(Step): 输入验证步骤 def execute(self, input_data, contextNone): user_input input_data.get(query, ).strip() if not user_input: raise ValueError(Query cannot be empty) if len(user_input) 1000: raise ValueError(Query too long) self.record_metadata({ input_length: len(user_input), validation_result: passed }) return {validated_query: user_input} class IntentClassifierStep(Step): 意图分类步骤模拟 def execute(self, input_data, contextNone): query input_data[validated_query].lower() # 简单的基于关键词的意图分类 tech_keywords [error, bug, install, config, how to] chat_keywords [hello, hi, how are you, weather] intent general if any(keyword in query for keyword in tech_keywords): intent technical elif any(keyword in query for keyword in chat_keywords): intent chat self.record_metadata({ detected_intent: intent, classification_confidence: high # 模拟置信度 }) return {query: input_data[validated_query], intent: intent} class KnowledgeBaseRetrievalStep(Step): 知识库检索步骤模拟 def execute(self, input_data, contextNone): if input_data[intent] ! technical: return input_data # 非技术问题跳过检索 query input_data[query] # 模拟知识库检索 time.sleep(0.1) # 模拟网络延迟 relevant_docs [ {title: Common Installation Issues, content: Check system requirements...}, {title: Configuration Guide, content: Edit config file at /etc/app/...} ] self.record_metadata({ retrieved_docs_count: len(relevant_docs), retrieval_time_ms: 100 }) return {**input_data, retrieved_docs: relevant_docs} class ResponseGeneratorStep(Step): 响应生成步骤 def execute(self, input_data, contextNone): intent input_data[intent] query input_data[query] if intent chat: response Hello! Im an AI assistant. How can I help you today? elif intent technical: docs input_data.get(retrieved_docs, []) # 模拟基于检索结果的响应生成 response fBased on {len(docs)} relevant documents: This appears to be a technical issue. Please check the documentation. else: # 模拟普通 LLM 响应 response fI understand youre asking: {query}. Heres what I think: This is an interesting question that requires careful consideration. self.record_metadata({ response_type: intent, response_length: len(response) }) return {final_response: response} def create_qa_pipeline(): 创建问答流水线 pipeline Pipeline(QA System Pipeline) pipeline.add_step(InputValidatorStep(input_validation)) pipeline.add_step(IntentClassifierStep(intent_classification)) pipeline.add_step(KnowledgeBaseRetrievalStep(knowledge_retrieval)) pipeline.add_step(ResponseGeneratorStep(response_generation)) return pipeline def main(): 主函数测试问答系统 pipeline create_qa_pipeline() # 测试用例 test_cases [ {query: How do I fix the installation error?}, {query: Whats the weather like today?}, {query: Can you explain machine learning?}, {query: } # 空查询用于测试错误处理 ] for i, test_case in enumerate(test_cases): print(f\n 测试用例 {i1} ) print(f输入: {test_case[query]}) try: with opensmith.trace(fQA Test {i1}, tags{test_case: i1}) as trace: result pipeline.run(test_case, trace_contexttrace) print(f响应: {result.get(final_response, No response)}) except Exception as e: print(f错误: {e}) if __name__ __main__: main()5.3 运行与分析追踪数据运行问答系统python qa_system.py系统会处理多个测试用例并在qa_traces.db中生成详细的追踪记录。接下来我们编写一个分析脚本来提取有价值的洞察。创建analyze_traces.py# analyze_traces.py import sqlite3 import json from datetime import datetime def analyze_qa_performance(): 分析问答系统性能 conn sqlite3.connect(./qa_traces.db) cursor conn.cursor() print( 问答系统性能分析 \n) # 1. 总体统计 cursor.execute(SELECT COUNT(*), AVG(duration_ms) FROM traces;) total_traces, avg_duration cursor.fetchone() print(f总执行次数: {total_traces}) print(f平均执行时间: {avg_duration:.2f} ms\n) # 2. 步骤性能分析 cursor.execute( SELECT name, COUNT(*) as execution_count, AVG(duration_ms) as avg_duration, MAX(duration_ms) as max_duration, SUM(CASE WHEN status ! completed THEN 1 ELSE 0 END) as error_count FROM spans GROUP BY name ORDER BY avg_duration DESC; ) print(步骤性能分析:) print(- * 60) for step_name, count, avg_dur, max_dur, errors in cursor.fetchall(): print(f{step_name:25} | 执行: {count:3} | 平均: {avg_dur:6.1f} ms | 最大: {max_dur:5} ms | 错误: {errors}) # 3. 意图分布分析 cursor.execute( SELECT m.value, COUNT(*) FROM metadata m JOIN spans s ON m.span_id s.id WHERE m.key detected_intent AND s.name intent_classification GROUP BY m.value; ) print(f\n意图分布:) print(- * 30) for intent, count in cursor.fetchall(): print(f{intent:15} : {count} 次) # 4. 错误分析 cursor.execute( SELECT t.name, t.error_message, s.name as step_name, s.error_data FROM traces t LEFT JOIN spans s ON t.id s.trace_id AND s.status ! completed WHERE t.status ! completed LIMIT 10; ) errors cursor.fetchall() if errors: print(f\n最近错误记录:) print(- * 50) for trace_name, trace_error, step_name, step_error in errors: print(f流水线: {trace_name}) print(f步骤: {step_name}) print(f错误: {trace_error or step_error}) print() conn.close() if __name__ __main__: analyze_qa_performance()运行分析脚本python analyze_traces.py你将获得类似以下的性能报告 问答系统性能分析 总执行次数: 4 平均执行时间: 325.50 ms 步骤性能分析: --------------------------------------------------------- knowledge_retrieval | 执行: 1 | 平均: 100.0 ms | 最大: 100 ms | 错误: 0 response_generation | 执行: 3 | 平均: 15.3 ms | 最大: 20 ms | 错误: 0 intent_classification | 执行: 3 | 平均: 5.0 ms | 最大: 10 ms | 错误: 0 input_validation | 执行: 3 | 平均: 2.3 ms | 最大: 5 ms | 错误: 1 意图分布: ------------------------------ technical : 1 次 general : 2 次 chat : 1 次 最近错误记录: -------------------------------------------------- 流水线: QA Test 4 步骤: input_validation 错误: Query cannot be empty通过这样的分析你可以清晰地了解系统的性能瓶颈、错误模式和用户行为模式为优化提供数据支持。6. 高级特性与集成方案OpenSmith 除了基础追踪功能外还提供了一些高级特性可以帮助你在更复杂的场景中发挥作用。本节将介绍这些特性及其实际应用。6.1 自定义追踪存储后端虽然 SQLite 是默认选择但 OpenSmith 支持自定义存储后端。你可以实现自己的存储适配器将数据保存到其他数据库或系统中。实现自定义存储后端示例from opensmith.storage import TraceStorage import psycopg2 # 假设使用 PostgreSQL class PostgreSQLStorage(TraceStorage): def __init__(self, connection_string): self.conn psycopg2.connect(connection_string) def save_trace(self, trace_data): # 实现将 trace_data 保存到 PostgreSQL 的逻辑 with self.conn.cursor() as cursor: cursor.execute( INSERT INTO traces (name, start_time, end_time, duration_ms, status) VALUES (%s, %s, %s, %s, %s) RETURNING id , (trace_data[name], trace_data[start_time], trace_data[end_time], trace_data[duration_ms], trace_data[status])) trace_id cursor.fetchone()[0] self.conn.commit() return trace_id def save_span(self, span_data): # 实现保存 span 的逻辑 pass # 实现其他必要方法... # 使用自定义存储 custom_storage PostgreSQLStorage(postgresql://user:passlocalhost/db) opensmith.init(storage_backendcustom_storage)6.2 与现有 LLM 框架集成OpenSmith 可以轻松集成到流行的 LLM 开发框架中如 LangChain、LlamaIndex 等。LangChain 集成示例from langchain.llms import OpenAI from langchain.chains import LLMChain from langchain.prompts import PromptTemplate import opensmith from opensmith.integrations.langchain import LangChainTracer # 创建 LangChain 组件 llm OpenAI(temperature0.7) prompt PromptTemplate( input_variables[question], templateAnswer the following question: {question} ) chain LLMChain(llmllm, promptprompt) # 使用 OpenSmith 追踪器 tracer LangChainTracer() # 执行并追踪 with opensmith.trace(LangChain QA) as trace: result chain.run( What is the capital of France?, callbacks[tracer.with_trace(trace)] ) print(result)6.3 批量处理与性能优化当处理大量请求时需要考虑性能优化策略。批量追踪配置import opensmith from concurrent.futures import ThreadPoolExecutor # 配置批量处理 opensmith.init( trace_db_path./batch_traces.db, batch_size100, # 批量提交大小 flush_interval30 # 自动刷新间隔秒 ) def process_single_query(query_data): 处理单个查询 with opensmith.trace(Batch Processing) as trace: # 模拟处理逻辑 result fProcessed: {query_data} trace.set_metadata(processing_result, success) return result # 批量处理示例 def batch_process_queries(queries): with ThreadPoolExecutor(max_workers5) as executor: results list(executor.map(process_single_query, queries)) return results # 测试批量处理 queries [fquery_{i} for i in range(50)] results batch_process_queries(queries) print(f处理了 {len(results)} 个查询)6.4 追踪数据导出与分析OpenSmith 的追踪数据可以导出为各种格式便于与其他分析工具集成。数据导出示例import sqlite3 import json import pandas as pd from datetime import datetime, timedelta def export_traces_to_analysis(): 导出追踪数据用于分析 conn sqlite3.connect(./qa_traces.db) # 导出到 Pandas DataFrame traces_df pd.read_sql_query(SELECT * FROM traces, conn) spans_df pd.read_sql_query(SELECT * FROM spans, conn) metadata_df pd.read_sql_query(SELECT * FROM metadata, conn) conn.close() # 保存为 CSV traces_df.to_csv(traces_export.csv, indexFalse) spans_df.to_csv(spans_export.csv, indexFalse) # 或者保存为 JSON 格式 export_data { export_time: datetime.now().isoformat(), traces: traces_df.to_dict(records), spans: spans_df.to_dict(records) } with open(traces_export.json, w) as f: json.dump(export_data, f, indent2) print(数据导出完成) # 简单的数据分析 print(f\n数据分析摘要:) print(f总追踪记录: {len(traces_df)}) print(f总步骤记录: {len(spans_df)}) print(f成功率: {(traces_df[status] completed).mean() * 100:.1f}%) export_traces_to_analysis()7. 常见问题与解决方案在实际使用 OpenSmith 的过程中你可能会遇到一些典型问题。本节整理了常见问题及其解决方案帮助你快速排错。7.1 安装与配置问题问题 1导入 opensmith 时出现 ModuleNotFoundError错误信息ModuleNotFoundError: No module named opensmith解决方案确认虚拟环境已激活且 OpenSmith 已正确安装pip list | grep opensmith如果未找到重新安装pip install opensmith检查 Python 路径是否正确确保使用的是虚拟环境中的 Python。问题 2数据库文件权限错误错误信息sqlite3.OperationalError: unable to open database file解决方案检查当前用户对目标目录是否有写权限尝试指定绝对路径opensmith.init(trace_db_path/home/user/project/traces.db)或者使用用户主目录import os db_path os.path.expanduser(~/traces.db) opensmith.init(trace_db_pathdb_path)7.2 运行时问题问题 3流水线执行性能下降现象随着追踪数据增多系统运行变慢。解决方案定期归档旧数据# 自动清理30天前的数据 opensmith.init(trace_db_path./traces.db, retention_days30)对于高频场景考虑增加批量提交大小opensmith.init(batch_size500, flush_interval60)使用更高效的存储后端如 PostgreSQL问题 4追踪数据不完整现象某些步骤的输入输出数据没有正确记录。解决方案确保所有步骤都正确调用record_metadata方法检查数据序列化OpenSmith 使用 JSON 序列化确保所有记录的数据都是 JSON 可序列化的验证步骤执行流程确保没有异常导致追踪上下文提前结束7.3 数据查询与分析问题问题 5复杂查询性能差解决方案为常用查询字段添加索引CREATE INDEX idx_traces_start_time ON traces(start_time); CREATE INDEX idx_spans_trace_id ON spans(trace_id); CREATE INDEX idx_metadata_span_id ON metadata(span_id);使用数据库连接池避免频繁连接开销考虑定期将数据导出到分析数据库如 ClickHouse进行处理问题 6追踪数据量过大解决方案实现