(
py: Python<'py>,
name: String,
request: PyLLMRequest,
func: Py<PyAny>,
collector: Py<PyAny>,
finalizer: Py<PyAny>,
handle: Option<PyScopeHandle>,
attributes: Option<P
| 806 | ) -> "object", text_signature = "(name: str, request: LlmRequest, func: object, collector: object, finalizer: object, *, handle: ScopeHandle | None = None, attributes: LlmAttributes | None = None, data: object | None = None, metadata: object | None = None, model_name: str | None = None, codec: object | None = None, response_codec: object | None = None) -> object")] |
| 807 | #[allow(clippy::too_many_arguments)] |
| 808 | fn llm_stream_call_execute<'py>( |
| 809 | py: Python<'py>, |
| 810 | name: String, |
| 811 | request: PyLLMRequest, |
| 812 | func: Py<PyAny>, |
| 813 | collector: Py<PyAny>, |
| 814 | finalizer: Py<PyAny>, |
| 815 | handle: Option<PyScopeHandle>, |
| 816 | attributes: Option<PyLLMAttributes>, |
| 817 | data: Option<&Bound<'py, PyAny>>, |
| 818 | metadata: Option<&Bound<'py, PyAny>>, |
| 819 | model_name: Option<String>, |
| 820 | codec: Option<&Bound<'py, PyAny>>, |
| 821 | response_codec: Option<&Bound<'py, PyAny>>, |
| 822 | ) -> PyResult<Bound<'py, PyAny>> { |
| 823 | let attrs = attributes |
| 824 | .map(|a| a.inner) |
| 825 | .unwrap_or(LlmAttributes::empty()); |
| 826 | let data_json = opt_py_to_json(data)?; |
| 827 | let metadata_json = opt_py_to_json(metadata)?; |
| 828 | let exec_fn = py_callable::wrap_py_llm_stream_exec_fn(func); |
| 829 | let default_fn: LlmStreamExecutionNextFn = Arc::new(move |req| exec_fn(req)); |
| 830 | let collector_fn = py_callable::wrap_py_collector_fn(collector); |
| 831 | let finalizer_fn = py_callable::wrap_py_finalizer_fn(finalizer); |
| 832 | let parent_handle = handle.map(|h| h.inner).unwrap_or_else(task_scope_top); |
| 833 | let codec_arc: Option<Arc<dyn LlmCodec>> = codec.map(|c| { |
| 834 | Arc::new(py_callable::PyLlmCodecWrapper { |
| 835 | py_codec: c.clone().unbind(), |
| 836 | }) as Arc<dyn LlmCodec> |
| 837 | }); |
| 838 | let response_codec_arc = py_llm_response_codec(response_codec); |
| 839 | |
| 840 | let scope_stack = current_scope_stack_handle(); |
| 841 | pyo3_async_runtimes::tokio::future_into_py(py, async move { |
| 842 | TASK_SCOPE_STACK |
| 843 | .scope(scope_stack, async move { |
| 844 | let params = core_llm_api::LlmStreamCallExecuteParams::builder() |
| 845 | .name(name) |
| 846 | .request(request.inner) |
| 847 | .func(default_fn) |
| 848 | .collector(collector_fn) |
| 849 | .finalizer(finalizer_fn) |
| 850 | .parent(parent_handle) |
| 851 | .attributes(attrs) |
| 852 | .data_opt(data_json) |
| 853 | .metadata_opt(metadata_json) |
| 854 | .model_name_opt(model_name) |
| 855 | .codec_opt(codec_arc) |
| 856 | .response_codec_opt(response_codec_arc) |
| 857 | .build(); |
| 858 | let rust_stream = core_llm_api::llm_stream_call_execute(params) |
| 859 | .await |
| 860 | .map_err(to_py_err)?; |
| 861 | |
| 862 | // Spawn a tokio task that drains the Rust stream into an mpsc channel |
| 863 | let (tx, rx) = tokio::sync::mpsc::channel::<FlowResult<serde_json::Value>>(32); |
| 864 | tokio::spawn(forward_stream_to_channel(rust_stream, tx)); |
| 865 |
nothing calls this directly
no test coverage detected