Agent¶

Agent is the core conversational AI agent. It orchestrates LLM calls, tool execution, concurrent workers, and rolling memory automatically, so you only need to send queries and consume the resulting stream.

Lifecycle¶

The agent must be used as an async context manager. The sandbox is created on entry and closed on exit:

from opendatasci import create_agent

async with create_agent("data.csv") as agent:
    async for event in agent.astream("Describe this dataset"):
        print(event)

For advanced use cases where you need to control each dependency explicitly, construct the agent directly:

from opendatasci.agents.agents import Agent
from opendatasci.workspace.local import LocalWorkspace
from opendatasci import OpenDataSciConfig

workspace = LocalWorkspace("./data/")
config = OpenDataSciConfig(provider="anthropic")

async with Agent(workspace=workspace, config=config) as agent:
    async for event in agent.astream("Analyse sales trends"):
        ...

Streaming events¶

agent.astream() is an async generator that yields AgentStreamEvent objects as the agent works. See the Events & Types page for the full event taxonomy.

Handling an interrupt¶

Some tools pause the agent and ask the user to pick an option (for example, to confirm a destructive operation). When this happens an AgentStreamEvent with type="input_required" is yielded. Resume the agent by calling astream() again with the user's answer:

async for event in agent.astream(query):
    if event.type == "input_required":
        choice = input(f"{event.content} [{', '.join(event.metadata['choices'])}]: ")
        async for follow_up in agent.astream(choice):
            # process follow_up events as usual
            ...
    elif event.type == "token":
        print(event.content, end="", flush=True)
    elif event.type == "response":
        print()

Managing conversation history¶

Method	Description
`clear_chat_history()`	Remove all messages and rolling memory summaries. Preserves sandbox state.
`rewind_turn()`	Remove only the last turn (user message + agent response) from the conversation.
`compact_chat_history()`	Fold all turn summaries and any existing compaction into a single `ChatHistoryCompaction` record. Returns the compaction text. Use this instead of `clear_chat_history` when you want to preserve context across a long session.

# After many turns, compact instead of clearing:
summary = await agent.compact_chat_history()
print("Compacted:", summary)

Reference¶

Agent ¶

Agent(workspace: BaseWorkspace, context_store: BaseContextStore | None = None, skill_store: BaseSkillStore | None = None, sandbox_factory: BaseSandboxFactory | None = None, checkpointer: BaseCheckpointSaver[Any] | None = None, tools: list[BaseTool] | None = None, session_id: str | None = None, session_manager: BaseSessionManager | None = None, config: OpenDataSciConfig | None = None)

Bases: BaseOpenDataSciAgent

Data science and machine learning conversational AI agent.

Must be used as an async context manager; the sandbox is created on entry and closed on exit::

async with Agent(...) as agent:
    async for event in agent.astream("analyse the data"):
        ...

For most use cases prefer the :func:create_agent factory, which wires all dependencies from a file or directory path.

Parameters:

Name	Type	Description	Default
`workspace`	`BaseWorkspace`	The workspace the agent operates on.	required
`session_id`	`str \| None`	Identifier for this session. Generated automatically when omitted.	`None`
`context_store`	`BaseContextStore \| None`	Store that supplies dataset profiles and notes for the active workspace and persists the agent's plan across turns. A local file-based store is created when omitted.	`None`
`skill_store`	`BaseSkillStore \| None`	Registry that the agent queries to resolve named skills at runtime. Defaults to the built-in :class:`LocalSkillStore`.	`None`
`session_manager`	`BaseSessionManager \| None`	Tracks the session's conversation threads in the graph checkpointer; clearing the conversation creates a new thread. Defaults to a file-backed :class:`LocalSessionManager`.	`None`
`sandbox_factory`	`BaseSandboxFactory \| None`	Factory used to create the execution sandbox. The sandbox lifetime is tied to the agent's context manager scope. Defaults to :class:`SRTSandboxFactory`.	`None`
`checkpointer`	`BaseCheckpointSaver[Any] \| None`	Checkpoint backend for graph state. Defaults to an in-memory store.	`None`
`tools`	`list[BaseTool] \| None`	Full set of tools available to the agent. Plan mode and self-review mode use this list minus worker-spawning tools. Override to restrict capabilities or inject custom tools.	`None`
`config`	`OpenDataSciConfig \| None`	LLM provider and model settings. Defaults to :class:`OpenDataSciConfig` with its built-in defaults.	`None`

astream `async` ¶

astream(user_input: str) -> AsyncIterator[AgentStreamEvent]

Stream a response to user_input, yielding :class:AgentStreamEvent objects.

If the previous call ended with an :class:~opendatasci.streaming.InputRequiredEvent, pass the user's answer here to resume; otherwise user_input starts a new turn.

rewind_turn `async` ¶

rewind_turn() -> None

Remove the last turn from the conversation history.

clear_chat_history `async` ¶

clear_chat_history() -> None

Clear all conversation context (preserves session state such as the sandbox).

Drops the conversation history, turn summaries, compaction, active skills, mode flags, and any pending interrupt by starting a fresh checkpointer thread; cancels any in-flight turn summarization; and deletes the session's persisted plan so it is no longer recalled.

compact_chat_history `async` ¶

compact_chat_history() -> str

Fold the rolling turn summaries into a single compaction summary.

Includes any existing compaction, all turn summaries, and the current completed turn (if any) in the compaction context. Clears turn summaries and replaces any existing compaction with the new one. An ongoing (incomplete) turn is left untouched.

Returns the compaction text, or a placeholder when there is nothing to compact.

Session manager¶

A session manager tracks the mapping from a session to its conversation threads in the graph checkpointer. Clearing the conversation (clear_chat_history) creates a new thread, so no prior state is visible to the LLM on the next turn.

The default LocalSessionManager persists this mapping to .opendatasci/session.json inside the workspace. For cloud or multi-process deployments — where two agent instances must share a conversation — replace it with a custom BaseSessionManager backed by shared storage:

import uuid
from opendatasci.session.session_manager import BaseSessionManager

class RedisSessionManager(BaseSessionManager):
    def get_or_create_thread(self) -> uuid.UUID: ...
    def create_thread(self) -> uuid.UUID: ...
    def get_current_thread(self) -> uuid.UUID: ...

from opendatasci.agents.agents import Agent
from opendatasci import LocalWorkspace, OpenDataSciConfig

async with Agent(
    workspace=LocalWorkspace("data/"),
    session_manager=RedisSessionManager(),
    config=OpenDataSciConfig(),
) as agent:
    ...

BaseSessionManager ¶

Bases: ABC

Tracks the conversation threads of a single session.

get_or_create_thread `abstractmethod` ¶

get_or_create_thread() -> uuid.UUID

Return the session's current thread, creating the first one if needed.

create_thread `abstractmethod` ¶

create_thread() -> uuid.UUID

Create a new thread for the session and make it current.

get_current_thread `abstractmethod` ¶

get_current_thread() -> uuid.UUID

Return the session's current thread.

Raises:

Type	Description
`LookupError`	if the session has no threads yet.

LocalSessionManager ¶

LocalSessionManager(workspace_path: Path, session_id: str)

Bases: BaseSessionManager

File-backed session manager for single-process local runs.

Persists every session's state to .opendatasci/session.json in the workspace, as a mapping of session id to::

{
    "created_at": "<iso8601>",
    "last_updated_at": "<iso8601>",
    "threads": [{"thread_id": "<uuid>", "created_at": "<iso8601>"}, ...]
}

created_at is stamped when the session entry is first written and last_updated_at on every write. The file is read on every lookup and rewritten on every thread creation, so no thread state is held in memory. Not safe for concurrent writers; the TUI runs it on a single event loop in a single process.

Parameters:

Name	Type	Description	Default
`workspace_path`	`Path`	Root directory of the active workspace.	required
`session_id`	`str`	Identifier of the session whose threads are managed.	required

ConcurrentWorkerAgent¶

ConcurrentWorkerAgent is the sub-agent spawned internally when the orchestrator delegates subtasks to concurrent workers. You do not normally construct this directly.

ConcurrentWorkerAgent ¶

ConcurrentWorkerAgent(tools: list[BaseTool], config: OpenDataSciConfig | None = None, llm: BaseChatModel | None = None)

One-shot worker agent that executes a single delegated subtask to completion.

ainvoke `async` ¶

ainvoke(task: str, system_prompt: str, on_event: OnEventCallback | None = None, messages_out: list[Any] | None = None, initial_active_skills: list[Skill] | None = None, initial_active_skill_domains: list[SkillDomain] | None = None) -> str

Execute task to completion and return the final text response.