From 8b2bafd48132930e76a44268e3722b78d178cc4a Mon Sep 17 00:00:00 2001 From: Quanzheng Long Date: Tue, 17 Mar 2026 16:40:48 -0700 Subject: [PATCH] rm --- saf-python-sdk/Cargo.lock | 124 +++++++ saf-python-sdk/Cargo.toml | 3 +- saf-python-sdk/README.md | 2 +- .../saf_python_sdk/advanced_graph/state.py | 2 +- .../python/saf_python_sdk/rust_core_cffi.py | 347 ------------------ 5 files changed, 128 insertions(+), 350 deletions(-) delete mode 100644 saf-python-sdk/python/saf_python_sdk/rust_core_cffi.py diff --git a/saf-python-sdk/Cargo.lock b/saf-python-sdk/Cargo.lock index 8f3dadaf4..082228bd7 100644 --- a/saf-python-sdk/Cargo.lock +++ b/saf-python-sdk/Cargo.lock @@ -2,6 +2,12 @@ # It is not intended for manual editing. version = 4 +[[package]] +name = "autocfg" +version = "1.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c08606f8c3cbf4ce6ec8e28fb0014a2c086708fe954eaa885384a6165172e7e8" + [[package]] name = "bitflags" version = "2.11.0" @@ -14,6 +20,21 @@ version = "1.0.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9330f8b2ff13f34540b44e946ef35111825727b38d33286ef986142615121801" +[[package]] +name = "heck" +version = "0.5.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "2304e00983f87ffb38b55b444b5e3b60a884b5d30c0fca7d82fe33449bbe55ea" + +[[package]] +name = "indoc" +version = "2.0.7" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "79cf5c93f93228cf8efb3ba362535fb11199ac548a09ce117c9b1adc3030d706" +dependencies = [ + "rustversion", +] + [[package]] name = "itoa" version = "1.0.17" @@ -26,6 +47,7 @@ version = "0.1.0" dependencies = [ "libc", "parking_lot", + "pyo3", "serde", "serde_json", "tokio", @@ -52,6 +74,21 @@ version = "2.8.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f8ca58f447f06ed17d5fc4043ce1b10dd205e060fb3ce5b979b8ed8e59ff3f79" +[[package]] +name = "memoffset" +version = "0.9.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "488016bfae457b036d996092f6cb448677611ce4449e970ceaf42695203f218a" +dependencies = [ + "autocfg", +] + +[[package]] +name = "once_cell" +version = "1.21.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "9f7c3e4beb33f85d45ae3e3a1792185706c8e16d043238c593331cc7cd313b50" + [[package]] name = "parking_lot" version = "0.12.5" @@ -81,6 +118,12 @@ version = "0.2.17" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd" +[[package]] +name = "portable-atomic" +version = "1.13.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49" + [[package]] name = "proc-macro2" version = "1.0.106" @@ -90,6 +133,69 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "pyo3" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7778bffd85cf38175ac1f545509665d0b9b92a198ca7941f131f85f7a4f9a872" +dependencies = [ + "cfg-if", + "indoc", + "libc", + "memoffset", + "once_cell", + "portable-atomic", + "pyo3-build-config", + "pyo3-ffi", + "pyo3-macros", + "unindent", +] + +[[package]] +name = "pyo3-build-config" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "94f6cbe86ef3bf18998d9df6e0f3fc1050a8c5efa409bf712e661a4366e010fb" +dependencies = [ + "once_cell", + "target-lexicon", +] + +[[package]] +name = "pyo3-ffi" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e9f1b4c431c0bb1c8fb0a338709859eed0d030ff6daa34368d3b152a63dfdd8d" +dependencies = [ + "libc", + "pyo3-build-config", +] + +[[package]] +name = "pyo3-macros" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fbc2201328f63c4710f68abdf653c89d8dbc2858b88c5d88b0ff38a75288a9da" +dependencies = [ + "proc-macro2", + "pyo3-macros-backend", + "quote", + "syn", +] + +[[package]] +name = "pyo3-macros-backend" +version = "0.23.5" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fca6726ad0f3da9c9de093d6f116a93c1a38e417ed73bf138472cf4064f72028" +dependencies = [ + "heck", + "proc-macro2", + "pyo3-build-config", + "quote", + "syn", +] + [[package]] name = "quote" version = "1.0.45" @@ -108,6 +214,12 @@ dependencies = [ "bitflags", ] +[[package]] +name = "rustversion" +version = "1.0.22" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d" + [[package]] name = "scopeguard" version = "1.2.0" @@ -174,6 +286,12 @@ dependencies = [ "unicode-ident", ] +[[package]] +name = "target-lexicon" +version = "0.12.16" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "61c41af27dd6d1e27b1b16b489db798443478cef1f06a660c96db617ba5de3b1" + [[package]] name = "tokio" version = "1.50.0" @@ -201,6 +319,12 @@ version = "1.0.24" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "e6e4313cd5fcd3dad5cafa179702e2b244f760991f45397d14d4ebf38247da75" +[[package]] +name = "unindent" +version = "0.2.4" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "7264e107f553ccae879d21fbea1d6724ac785e8c3bfc762137959b5802826ef3" + [[package]] name = "windows-link" version = "0.2.1" diff --git a/saf-python-sdk/Cargo.toml b/saf-python-sdk/Cargo.toml index f708e859b..a2c41930a 100644 --- a/saf-python-sdk/Cargo.toml +++ b/saf-python-sdk/Cargo.toml @@ -10,9 +10,10 @@ crate-type = ["cdylib", "rlib"] [features] default = [] -python-bindings = [] +python-bindings = ["dep:pyo3"] [dependencies] +pyo3 = { version = "0.23.5", features = ["extension-module"], optional = true } serde = { version = "1.0", features = ["derive"] } serde_json = "1.0" parking_lot = "0.12" diff --git a/saf-python-sdk/README.md b/saf-python-sdk/README.md index e7d50f8fb..45ada8435 100644 --- a/saf-python-sdk/README.md +++ b/saf-python-sdk/README.md @@ -5,7 +5,7 @@ Standalone Python SDK for the `advanced_graph` runtime backed by the Rust engine This package intentionally contains only: - `saf_python_sdk.advanced_graph` (Python API) -- Rust core engine via C bindings (`ctypes`) +- `langgraph_rust_core` (Rust execution engine via PyO3) It does not package the original `langgraph` `stategraph` stack. diff --git a/saf-python-sdk/python/saf_python_sdk/advanced_graph/state.py b/saf-python-sdk/python/saf_python_sdk/advanced_graph/state.py index a32c5c697..4a388e377 100644 --- a/saf-python-sdk/python/saf_python_sdk/advanced_graph/state.py +++ b/saf-python-sdk/python/saf_python_sdk/advanced_graph/state.py @@ -11,7 +11,7 @@ from dataclasses import dataclass from datetime import timedelta from typing import Any, Generic, TypeVar, cast -from saf_python_sdk.rust_core_cffi import PyRustEngine +from saf_python_sdk.langgraph_rust_core import PyRustEngine # type: ignore[import-untyped] from saf_python_sdk.types import Command, Send diff --git a/saf-python-sdk/python/saf_python_sdk/rust_core_cffi.py b/saf-python-sdk/python/saf_python_sdk/rust_core_cffi.py deleted file mode 100644 index 7e53332ea..000000000 --- a/saf-python-sdk/python/saf_python_sdk/rust_core_cffi.py +++ /dev/null @@ -1,347 +0,0 @@ -from __future__ import annotations - -import ctypes -import dataclasses -import json -import subprocess -from copy import deepcopy -from pathlib import Path -import threading -from typing import Any, Callable, get_args, get_origin - - -class PyRustEngine: - def __init__(self) -> None: - self._lib = _load_rust_lib() - self._engine = self._lib.rc_engine_new() - if not self._engine: - raise RuntimeError("failed to create rust engine") - - def __del__(self) -> None: - engine = getattr(self, "_engine", None) - if engine: - self._lib.rc_engine_free(engine) - self._engine = None - - def add_async_channel(self, name: str) -> None: - self._call_status(self._lib.rc_add_async_channel, name.encode()) - - def publish_obj(self, channel: str, value: Any) -> None: - payload = json.dumps(value, ensure_ascii=False).encode() - self._call_status(self._lib.rc_publish_json, channel.encode(), payload) - - def wait_any_of_obj(self, any_of_payload: Any) -> Any: - payload = json.dumps(any_of_payload, ensure_ascii=False).encode() - raw = self._consume_json_ptr(self._lib.rc_wait_any_of_json(self._engine, payload)) - if not raw.get("ok"): - raise ValueError(raw.get("error", "rust wait_any_of failed")) - return raw["event"] - - def wait_channel(self, channel: str, n: int) -> Any: - return self.wait_any_of_obj({"conditions": [{"kind": "channel", "channel": channel, "n": n}]}) - - def wait_timer(self, seconds: float) -> Any: - return self.wait_any_of_obj({"conditions": [{"kind": "timer", "seconds": seconds}]}) - - def start_stream(self, stream_mode: str | None) -> None: - encoded = stream_mode.encode() if stream_mode is not None else None - self._call_status(self._lib.rc_start_stream, encoded) - - def receive_stream_obj(self) -> Any | None: - raw = self._consume_json_ptr(self._lib.rc_receive_stream_json(self._engine)) - if not raw.get("ok"): - raise ValueError(raw.get("error", "rust receive_stream failed")) - if not raw.get("has_event", False): - return None - return raw.get("event") - - def send_custom_stream_event_obj(self, value: Any) -> None: - payload = json.dumps(value, ensure_ascii=False).encode() - self._call_status(self._lib.rc_send_custom_stream_event, payload) - - def close_stream(self) -> None: - self._call_status(self._lib.rc_close_stream) - - def run_graph_py( - self, - entry_point: str, - finish_point: str, - initial_state: Any, - callback: Callable[[str, Any, Any], dict[str, Any]], - stream_mode: str | None = None, - ) -> Any: - shared_state = initial_state - shared_state_lock = threading.Lock() - state_type = type(initial_state) - use_shared_state = dataclasses.is_dataclass(initial_state) - initial_state_json = json.dumps(_to_jsonable(shared_state), ensure_ascii=False).encode() - callback_c = _make_node_callback( - callback, - state_type, - use_shared_state, - shared_state, - shared_state_lock, - ) - stream_mode_encoded = stream_mode.encode() if stream_mode is not None else None - out = self._consume_json_ptr( - self._lib.rc_run_graph_json( - self._engine, - entry_point.encode(), - finish_point.encode(), - initial_state_json, - initial_state_json, - stream_mode_encoded, - ctypes.c_ulong(0), - callback_c, - ) - ) - if not out.get("ok"): - raise ValueError(out.get("error", "rust run_graph failed")) - if use_shared_state: - return shared_state - return _coerce_for_type(out["state"], state_type) - - def _call_status(self, func: Any, *args: Any) -> None: - raw = self._consume_json_ptr(func(self._engine, *args)) - if not raw.get("ok"): - raise ValueError(raw.get("error", "rust call failed")) - - def _consume_json_ptr(self, ptr: ctypes.c_void_p) -> dict[str, Any]: - if not ptr: - raise RuntimeError("rust returned null string pointer") - try: - text = ctypes.cast(ptr, ctypes.c_char_p).value - if text is None: - raise RuntimeError("rust returned empty string pointer") - return json.loads(text.decode()) - finally: - self._lib.rc_string_free(ptr) - - -def _make_node_callback( - callback: Callable[[str, Any, Any], dict[str, Any]], - state_type: type[Any], - use_shared_state: bool, - shared_state: Any, - shared_state_lock: threading.Lock, -) -> ctypes.CFUNCTYPE: # type: ignore[type-arg] - cb_type = ctypes.CFUNCTYPE( - ctypes.c_void_p, - ctypes.c_ulong, - ctypes.c_char_p, - ctypes.c_char_p, - ctypes.c_char_p, - ) - libc = ctypes.CDLL(None) - libc.malloc.argtypes = [ctypes.c_size_t] - libc.malloc.restype = ctypes.c_void_p - - @cb_type - def _callback( - _user_data: int, - node_ptr: bytes, - arg_ptr: bytes, - state_ptr: bytes, - ) -> ctypes.c_void_p: - try: - node = node_ptr.decode() - arg = json.loads(arg_ptr.decode()) - if use_shared_state: - with shared_state_lock: - before = deepcopy(_to_jsonable(shared_state)) - result = callback(node, arg, shared_state) - state_after_call = shared_state - else: - state_raw = json.loads(state_ptr.decode()) - state_snapshot = _coerce_for_type(state_raw, state_type) - before = deepcopy(_to_jsonable(state_snapshot)) - result = callback(node, arg, state_snapshot) - state_after_call = state_snapshot - if "suspend" in result: - envelope = {"ok": True, "suspend": result["suspend"]} - else: - update = result.get("update") - if update is not None: - update_json = _to_jsonable(update) - if update_json == before: - update = None - update_json = None - else: - if use_shared_state: - _apply_update_to_state(shared_state, update) - after = _to_jsonable(state_after_call) - if update is None and after != before: - update_json = after - elif update is None: - update_json = None - else: - update_json = _to_jsonable(update) - sends = [] - for item in result.get("sends", []): - if not isinstance(item, dict): - continue - sends.append( - { - "node": item.get("node"), - "arg": _to_jsonable(item.get("arg")), - } - ) - envelope = { - "ok": True, - "payload": { - "update": update_json, - "sends": sends, - }, - } - except Exception as exc: # noqa: BLE001 - envelope = {"ok": False, "error": f"python callback failed: {exc}"} - return _malloc_c_string(json.dumps(envelope, ensure_ascii=False).encode(), libc) - - return _callback - - -def _malloc_c_string(payload: bytes, libc: Any) -> ctypes.c_void_p: - size = len(payload) + 1 - ptr = libc.malloc(size) - if not ptr: - return ctypes.c_void_p(0) - ctypes.memmove(ptr, payload, len(payload)) - ctypes.memset(ctypes.c_void_p(ptr + len(payload)), 0, 1) - return ptr - - -def _load_rust_lib() -> ctypes.CDLL: - env = Path.cwd() - root = _find_repo_root(env) - rust_core = root / "rust-core" - lib_path = _resolve_lib_path(rust_core) - if not lib_path.exists(): - subprocess.run(["cargo", "build"], cwd=rust_core, check=True) - lib = ctypes.CDLL(str(lib_path)) - _configure_signatures(lib) - return lib - - -def _find_repo_root(start: Path) -> Path: - current = start.resolve() - for candidate in [current, *current.parents]: - if (candidate / "rust-core").exists() and (candidate / "saf-python-sdk").exists(): - return candidate - here = Path(__file__).resolve() - return here.parents[3] - - -def _resolve_lib_path(rust_core: Path) -> Path: - if (rust_core / "target" / "debug" / "liblanggraph_rust_core.dylib").exists(): - return rust_core / "target" / "debug" / "liblanggraph_rust_core.dylib" - if (rust_core / "target" / "debug" / "liblanggraph_rust_core.so").exists(): - return rust_core / "target" / "debug" / "liblanggraph_rust_core.so" - if (rust_core / "target" / "debug" / "langgraph_rust_core.dll").exists(): - return rust_core / "target" / "debug" / "langgraph_rust_core.dll" - return rust_core / "target" / "debug" / "liblanggraph_rust_core.dylib" - - -def _configure_signatures(lib: ctypes.CDLL) -> None: - cb_type = ctypes.CFUNCTYPE( - ctypes.c_void_p, - ctypes.c_ulong, - ctypes.c_char_p, - ctypes.c_char_p, - ctypes.c_char_p, - ) - lib.rc_engine_new.argtypes = [] - lib.rc_engine_new.restype = ctypes.c_void_p - lib.rc_engine_free.argtypes = [ctypes.c_void_p] - lib.rc_engine_free.restype = None - lib.rc_string_free.argtypes = [ctypes.c_void_p] - lib.rc_string_free.restype = None - lib.rc_add_async_channel.argtypes = [ctypes.c_void_p, ctypes.c_char_p] - lib.rc_add_async_channel.restype = ctypes.c_void_p - lib.rc_publish_json.argtypes = [ctypes.c_void_p, ctypes.c_char_p, ctypes.c_char_p] - lib.rc_publish_json.restype = ctypes.c_void_p - lib.rc_wait_any_of_json.argtypes = [ctypes.c_void_p, ctypes.c_char_p] - lib.rc_wait_any_of_json.restype = ctypes.c_void_p - lib.rc_start_stream.argtypes = [ctypes.c_void_p, ctypes.c_char_p] - lib.rc_start_stream.restype = ctypes.c_void_p - lib.rc_receive_stream_json.argtypes = [ctypes.c_void_p] - lib.rc_receive_stream_json.restype = ctypes.c_void_p - lib.rc_send_custom_stream_event.argtypes = [ctypes.c_void_p, ctypes.c_char_p] - lib.rc_send_custom_stream_event.restype = ctypes.c_void_p - lib.rc_close_stream.argtypes = [ctypes.c_void_p] - lib.rc_close_stream.restype = ctypes.c_void_p - lib.rc_run_graph_json.argtypes = [ - ctypes.c_void_p, - ctypes.c_char_p, - ctypes.c_char_p, - ctypes.c_char_p, - ctypes.c_char_p, - ctypes.c_char_p, - ctypes.c_ulong, - cb_type, - ] - lib.rc_run_graph_json.restype = ctypes.c_void_p - - -def _to_jsonable(value: Any) -> Any: - if value is None or isinstance(value, (str, int, float, bool)): - return value - if dataclasses.is_dataclass(value): - return {field.name: _to_jsonable(getattr(value, field.name)) for field in dataclasses.fields(value)} - model_dump = getattr(value, "model_dump", None) - if callable(model_dump): - return _to_jsonable(model_dump()) - if isinstance(value, dict): - return {str(k): _to_jsonable(v) for k, v in value.items()} - if isinstance(value, (list, tuple, set)): - return [_to_jsonable(v) for v in value] - return value - - -def _coerce_for_type(value: Any, typ: Any) -> Any: - if value is None: - return None - origin = get_origin(typ) - args = get_args(typ) - if origin is not None: - if origin in (list, tuple, set): - item_type = args[0] if args else Any - items = [_coerce_for_type(v, item_type) for v in value] - if origin is tuple: - return tuple(items) - if origin is set: - return set(items) - return items - if origin is dict: - value_type = args[1] if len(args) == 2 else Any - return {k: _coerce_for_type(v, value_type) for k, v in value.items()} - if isinstance(typ, type): - if dataclasses.is_dataclass(typ): - kwargs = {} - for field in dataclasses.fields(typ): - kwargs[field.name] = _coerce_for_type(value.get(field.name), field.type) - return typ(**kwargs) - model_validate = getattr(typ, "model_validate", None) - if callable(model_validate): - return model_validate(value) - return value - - -def _apply_update_to_state(state: Any, update: Any) -> None: - if update is None: - return - if isinstance(state, dict): - if isinstance(update, dict): - state.update(update) - return - if dataclasses.is_dataclass(update): - state.update(_to_jsonable(update)) - return - return - if dataclasses.is_dataclass(state): - if dataclasses.is_dataclass(update): - for field in dataclasses.fields(state): - setattr(state, field.name, getattr(update, field.name)) - return - if isinstance(update, dict): - for key, val in update.items(): - setattr(state, key, val)