企业官网建设流程全解析

在日常的 LLM 应用开发中，你是否遇到过这样的困扰：想要追踪和调试一个复杂的 AI 流水线，却发现现有的工具要么依赖云端服务，要么配置繁琐，难以在本地快速上手？尤其是在处理多步骤的 LLM 调用、数据转换或条件分支时，缺乏直观的追踪手段，导致调试效率低下，问题定位困难。OpenSmith 的出现，正是为了解决这一痛点。它是一个轻量级的本地工具，专注于帮助开发者无侵入地追踪 LLM 流水线的执行过程，所有数据均存储在本地 SQLite 数据库中，无需依赖任何云端服务。本文将带你从零开始，完整掌握 OpenSmith 的核心概念、安装配置、基础与高级用法，并通过实战案例展示如何利用其追踪能力优化你的 LLM 应用开发流程。无论你是刚接触 LLM 应用开发的初学者，还是希望提升现有项目可观测性的资深工程师，都能从中获得实用的解决方案。

1. OpenSmith 核心概念与价值

在深入使用 OpenSmith 之前，理解其设计理念和核心概念至关重要。这有助于我们更好地把握其适用场景和优势。

1.1 什么是 OpenSmith？

OpenSmith 是一个开源的本地化 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 安装 OpenSmith

OpenSmith 可以通过 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 --help

2.3 可选依赖与工具

虽然 OpenSmith 的核心功能无需额外依赖，但为了提升开发体验，建议安装以下工具：

DB Browser for SQLite（SQLite 数据库可视化工具）这是一个图形化界面工具，可以方便地浏览和查询 OpenSmith 生成的 SQLite 数据库文件。

• 下载地址：访问 DB Browser for SQLite 官网（sqlitebrowser.org）下载对应操作系统的版本。
• 安装：按照官网指引完成安装。安装后，你可以直接打开.db文件查看追踪数据。

Jupyter Notebook / JupyterLab如果你习惯在交互式环境中进行开发和调试，Jupyter 是一个很好的选择。OpenSmith 可以在 Jupyter 中无缝使用。

# 在虚拟环境中安装 Jupyter pip install jupyterlab

3. 快速开始：你的第一个追踪流水线

理论介绍完毕，现在让我们通过一个简单的示例，快速上手 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, model="gpt-3.5-turbo"): super().__init__(name) self.model = model def execute(self, input_data, context=None): # 这里是模拟的 LLM 调用逻辑 # 在实际项目中，这里会替换为真实的 OpenAI、Anthropic 等 API 调用 prompt = input_data.get("prompt", "") simulated_response = f"Simulated 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_context=trace) 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 type='table';") 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, context=None): 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, context=None): 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 pipeline

4.2 追踪配置与自定义元数据

OpenSmith 允许你精细控制追踪的详细程度，并添加业务相关的自定义元数据。

配置追踪详细程度：

import opensmith # 初始化时配置 opensmith.init( trace_db_path="./support_traces.db", log_level="INFO", # 控制日志输出级别 max_trace_duration=3600 # 设置追踪最大时长（秒） ) # 在追踪上下文中添加自定义标签 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, context=None): # 模拟真实的 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)}) raise

4.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, context=None): 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, context=None): 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, context=None): 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, context=None): intent = input_data["intent"] query = input_data["query"] if intent == "chat": response = "Hello! I'm an AI assistant. How can I help you today?" elif intent == "technical": docs = input_data.get("retrieved_docs", []) # 模拟基于检索结果的响应生成 response = f"Based on {len(docs)} relevant documents: This appears to be a technical issue. Please check the documentation." else: # 模拟普通 LLM 响应 response = f"I understand you're asking: {query}. Here's 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("Q&A 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": "What's the weather like today?"}, {"query": "Can you explain machine learning?"}, {"query": ""} # 空查询，用于测试错误处理 ] for i, test_case in enumerate(test_cases): print(f"\n=== 测试用例 {i+1} ===") print(f"输入: {test_case['query']}") try: with opensmith.trace(f"QA Test {i+1}", tags={"test_case": i+1}) as trace: result = pipeline.run(test_case, trace_context=trace) 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:pass@localhost/db") opensmith.init(storage_backend=custom_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(temperature=0.7) prompt = PromptTemplate( input_variables=["question"], template="Answer the following question: {question}" ) chain = LLMChain(llm=llm, prompt=prompt) # 使用 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_size=100, # 批量提交大小 flush_interval=30 # 自动刷新间隔（秒） ) def process_single_query(query_data): """处理单个查询""" with opensmith.trace("Batch Processing") as trace: # 模拟处理逻辑 result = f"Processed: {query_data}" trace.set_metadata("processing_result", "success") return result # 批量处理示例 def batch_process_queries(queries): with ThreadPoolExecutor(max_workers=5) as executor: results = list(executor.map(process_single_query, queries)) return results # 测试批量处理 queries = [f"query_{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', index=False) spans_df.to_csv('spans_export.csv', index=False) # 或者保存为 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, indent=2) 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_path=db_path)

7.2 运行时问题

问题 3：流水线执行性能下降

现象：随着追踪数据增多，系统运行变慢。

解决方案：

• 定期归档旧数据：

  # 自动清理30天前的数据 opensmith.init(trace_db_path="./traces.db", retention_days=30)

• 对于高频场景，考虑增加批量提交大小：

  opensmith.init(batch_size=500, flush_interval=60)

• 使用更高效的存储后端（如 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：追踪数据量过大

解决方案：

• 实现