Method pipeline

sdk/python/src/lib.rs:1599–1636

Run each item through a chain of `stages`, with no barrier between stages. Each stage is a callable `stage(ctx) -> spec_dict | None`, where `ctx = {"previous": , "item": }`. Return a spec dict (snake_case keys) to run that step, or `None` to stop the item's chain. A chain also stops when a step fails. Returns one entry per item (the last outcome dict, or `None`), in inp

(
        &self,
        py: Python<'_>,
        items: Vec<Bound<'_, PyAny>>,
        stages: Vec<Bound<'_, PyAny>>,
    )

    /// chain). Per-stage `output_schema` is not supported here — use `parallel`
    /// for schema-validated steps.
    fn pipeline(
        &self,
        py: Python<'_>,
        items: Vec<Bound<'_, PyAny>>,
        stages: Vec<Bound<'_, PyAny>>,
    ) -> PyResult<Vec<Option<PyObject>>> {
        let rust_items = items
            .iter()
            .map(|i| py_to_json_value(py, i))
            .collect::<PyResult<Vec<_>>>()?;
        let rust_stages: Vec<RustPipelineStage<serde_json::Value>> = stages
            .into_iter()
            .map(|s| {
                let stage = std::sync::Arc::new(PythonPipelineStage {
                    callback: s.unbind(),
                });
                let ps: RustPipelineStage<serde_json::Value> =
                    std::sync::Arc::new(move |prev, item| stage.invoke(prev, item));
                ps
            })
            .collect();

        let session = self.inner.clone();
        let outcomes = py.allow_threads(move || {
            get_runtime().block_on(async move {
                let executor = session.agent_executor();
                execute_pipeline(executor, rust_items, rust_stages, None).await
            })
        });

        outcomes
            .iter()
            .map(|o| match o {
                Some(outcome) => step_outcome_to_py(py, outcome).map(Some),
                None => Ok(None),
            })
            .collect()
    }

    /// Send a prompt or request and get a streaming iterator of events.
    ///