Implement RemotePregel class and PregelProtocol

[andrewnguonly]

Summary

This PR is a follow-up/reimplementation of this PR.

To Do

• Add tests
• Fix mypy errors

[vbarda]

this seems a bit strange

[andrewnguonly]

I wasn't sure what to do here because there's no concept of "cloning" a remote Pregel instance (i.e. creating a new Assistant which is a clone of the current one).

To "clone" or "copy" an Assistant, we have to...

1. Call GET /assistants/<graph_id>.
2. Update the configs (via merge).
3. Call POST /assistants to a create a new assistant, the clone/copy.

[nfcampos]

I don't think you need to call any endpoints, just save the config in the instance and then use it on every other method?

[andrewnguonly]

I don't think you need to call any endpoints, just save the config in the instance and then use it on every other method?

Using aupdate_state() as an example... What do we do with config passed in the method and self.config on the instance. Merge them?

 async def aupdate_state(
    self,
    config: RunnableConfig,
    values: Optional[Union[dict[str, Any], Any]],
    as_node: Optional[str] = None,
) -> RunnableConfig:

    # what do we do with self.config and config?
    self.config

    response: dict = await self.client.threads.update_state(  # type: ignore
        thread_id=config["configurable"]["thread_id"],
        values=values,  # type: ignore
        as_node=as_node,
        checkpoint=self._get_checkpoint(config),
    )
    return self._get_config(response["checkpoint"])

[nfcampos]

yes exactly merge them. exactly what the real pregel class does

[nfcampos]

why empty tuple?

[andrewnguonly]

I was mapping the fields from ThreadTask to PregelTask, but there's no path field in ThreadTask.

class ThreadTask(TypedDict):
    """Represents a task within a thread."""

    id: str
    name: str
    error: Optional[str]
    interrupts: list[dict]
    checkpoint: Optional[Checkpoint]
    state: Optional["ThreadState"]


class PregelTask(NamedTuple):
    id: str
    name: str
    path: tuple[Union[str, int], ...]
    error: Optional[Exception] = None
    interrupts: tuple[Interrupt, ...] = ()
    state: Union[None, RunnableConfig, "StateSnapshot"] = None

What's supposed to be in path or how do we get the path?

[nfcampos]

ah, because its an internal thing I removed from the api, I guess now that comes back to bite me

[nfcampos]

why do we pass this node_id default here?

[andrewnguonly]

I just needed to pass some default value. I can set the default to an empty string. name in DrawableNode is not Optional.

[nfcampos]

is there some way we can accept a single client and build the other one from it?, or make only one of them required at least

[nfcampos]

Most people will be using only sync or only async