Graphistry API Reference¶

from llamatelemetry.graphistry import (
    GraphistryBuilders, InferenceRecord, records_to_dataframe,
    traces_to_records, build_graph_nodes_edges, build_latency_time_series,
    GraphistryViz, TraceVisualization, MetricsVisualization, create_graph_viz,
    RAPIDSBackend, check_rapids_available, create_cudf_dataframe, run_cugraph_algorithm,
    GraphistryConnector, GraphistrySession, register_graphistry, plot_graph,
    GraphWorkload, SplitGPUManager, create_graph_from_llm_output, visualize_knowledge_graph,
)

@staticmethod
def knowledge_graph(
    entities: Optional[List[Any]] = None,
    relationships: Optional[List[Dict[str, Any]]] = None,
) -> Tuple[pd.DataFrame, pd.DataFrame]

nodes_df, edges_df = GraphistryBuilders.knowledge_graph(
    entities=["Python", "AI", "TensorFlow"],
    relationships=[
        {"source": "Python", "target": "AI", "type": "used_for"},
        {"source": "TensorFlow", "target": "AI", "type": "implements"},
    ],
)

@staticmethod
def document_similarity(
    documents: List[Any],
    similarities: List[Dict[str, Any]],
    doc_id_key: str = "id",
) -> Tuple[pd.DataFrame, pd.DataFrame]

@staticmethod
def embedding_knn(
    embeddings: List[List[float]],
    labels: Optional[List[str]] = None,
    k: int = 5,
    metric: str = "cosine",
) -> Tuple[pd.DataFrame, pd.DataFrame]

@staticmethod
def attention_graph(
    attention: List[List[float]],
    tokens: Optional[List[str]] = None,
    threshold: float = 0.0,
) -> Tuple[pd.DataFrame, pd.DataFrame]

@dataclass
class InferenceRecord:
    ts: float                           # Unix timestamp
    operation: str                      # Operation name (e.g., "chat")
    model: str                          # Model identifier
    latency_ms: float                   # End-to-end latency
    input_tokens: Optional[int]         # Prompt tokens
    output_tokens: Optional[int]        # Generated tokens
    ttfb_ms: Optional[float]            # Time to first byte
    prompt_ms: Optional[float]          # Prompt processing time
    generation_ms: Optional[float]      # Token generation time
    gpu_id: Optional[int]               # GPU device ID
    split_mode: Optional[str]           # Multi-GPU split mode
    success: Optional[bool]             # Request success
    error_type: Optional[str]           # Error type if failed

def traces_to_records(spans: List[Dict[str, Any]]) -> List[InferenceRecord]

def records_to_dataframe(records: Iterable[InferenceRecord]) -> pd.DataFrame

def build_graph_nodes_edges(
    df: pd.DataFrame,
    *,
    node_id_col: str = "operation",
    group_col: str = "model",
) -> Tuple[pd.DataFrame, pd.DataFrame]

def build_latency_time_series(df: pd.DataFrame, *, bucket: str = "1min") -> pd.DataFrame

class GraphistryViz:
    is_registered: bool  # property

def plot_inference_results(
    self,
    results: List[Any],
    color_by: str = "latency_ms",
    size_by: str = "tokens_generated",
    title: str = "Inference Results",
    **kwargs,
)

def plot_trace_graph(
    self,
    spans: List[Dict[str, Any]],
    color_by: str = "duration_ms",
    size_by: str = "tokens",
    title: str = "Trace Graph",
    **kwargs,
)

def plot_gpu_metrics(
    self,
    metrics: List[Dict[str, Any]],
    time_column: str = "timestamp",
    color_by: str = "gpu_utilization",
    size_by: str = "memory_used",
    title: str = "GPU Metrics Timeline",
    **kwargs,
)

def plot_latency_distribution(
    self,
    results: List[Any],
    bins: int = 20,
    title: str = "Latency Distribution",
    **kwargs,
)

def plot_knowledge_graph(
    self,
    entities: List[Dict[str, Any]],
    relationships: List[Dict[str, Any]],
    entity_id_col: str = "id",
    entity_label_col: str = "name",
    rel_source_col: str = "source",
    rel_target_col: str = "target",
    rel_type_col: str = "type",
    title: str = "Knowledge Graph",
    **kwargs,
)

@dataclass
class TraceVisualization:
    color_by: str = "latency_ms"
    size_by: str = "tokens"
    layout: str = "force"
    title: str = "LLM Inference Traces"
    palette: str = "viridis"

@dataclass
class MetricsVisualization:
    metric: str = "latency_ms"
    aggregation: str = "mean"
    time_window: str = "1min"

def create_graph_viz(
    edges: pd.DataFrame,
    nodes: Optional[pd.DataFrame] = None,
    source: str = "source",
    target: str = "target",
    node_id: str = "id",
    color_by: Optional[str] = None,
    size_by: Optional[str] = None,
    title: str = "Graph",
    auto_register: bool = True,
    **kwargs,
)

class RAPIDSBackend:
    cudf_available: bool      # property
    cugraph_available: bool   # property
    cuml_available: bool      # property

backend = RAPIDSBackend(gpu_id=1)
gdf = backend.create_dataframe({"src": [0, 1, 2], "dst": [1, 2, 0]})
pr = backend.pagerank(gdf)

def check_rapids_available() -> Dict[str, bool]

def run_cugraph_algorithm(
    algorithm: str,
    edges_df,
    source_col: str = "src",
    dest_col: str = "dst",
    **kwargs,
) -> Any

connector = GraphistryConnector()
g = connector.create_graph(edges_df, source="src", destination="dst", nodes_df=nodes_df)
g.plot()

def register_graphistry(
    username: Optional[str] = None,
    password: Optional[str] = None,
    api_key: Optional[str] = None,
    server: str = "hub.graphistry.com",
) -> bool

@dataclass
class GraphistrySession:
    connector: GraphistryConnector
    registered: bool
    server: str = "hub.graphistry.com"

@classmethod
def from_kaggle_secrets(cls, server="hub.graphistry.com", auto_register=True) -> GraphistrySession

@classmethod
def from_env(cls, server="hub.graphistry.com", auto_register=True) -> GraphistrySession

def create_knowledge_graph(
    self,
    entities: List[Dict[str, Any]],
    relationships: List[Dict[str, Any]],
    use_gpu: bool = True,
) -> Any

manager = SplitGPUManager(auto_detect=True)
print(manager.get_graph_env())  # {"CUDA_VISIBLE_DEVICES": "1"}
print(manager.get_llm_env())    # {"CUDA_VISIBLE_DEVICES": "0"}
args = manager.get_llama_server_args("/path/to/model.gguf")

def create_graph_from_llm_output(
    llm_response: str,
    workload: Optional[GraphWorkload] = None,
) -> Any

def visualize_knowledge_graph(
    entities: List[Dict], relationships: List[Dict], gpu_id: int = 1, **kwargs
) -> Any