agentgit.agents package

Submodules

agentgit.agents.agent_service module

Service for managing RollbackAgent instances.

Handles agent creation, rollback operations, and session management.

class agentgit.agents.agent_service.AgentService(model_config=None)[source]

Bases: object

Service for managing RollbackAgent instances.

Provides high-level operations for creating agents, handling rollbacks, and managing the interaction between external and internal sessions.

cleanup_agent(external_session_id)[source]

Remove an agent from active tracking.

Parameters:

external_session_id (int) – ID of the external session.

create_new_agent(external_session_id, session_name=None, base_url=None, api_key=None, **agent_kwargs)[source]

Create a new RollbackAgent for an external session.

Parameters:

  • external_session_id (int) – ID of the external session.

  • session_name (Optional[str]) – Optional name for the internal session.

  • base_url (Optional[str]) – Optional base URL for the model provider (overrides defaults/env if provided).

  • api_key (Optional[str]) – Optional API key for the model provider (overrides defaults/env if provided).

  • **agent_kwargs – Additional arguments for the agent.

Return type:

RollbackAgent

Returns:

The created RollbackAgent instance.

get_active_agent(external_session_id)[source]

Get an active agent for a session if one exists.

Parameters:

external_session_id (int) – ID of the external session.

Return type:

Optional[RollbackAgent]

Returns:

The active RollbackAgent if found, None otherwise.

get_branch_tree(external_session_id)[source]

Get the branch tree structure for an external session.

Parameters:

external_session_id (int) – ID of the external session.

Return type:

Dict[str, Any]

Returns:

Dictionary representing the branch tree structure.

get_conversation_summary(agent)[source]

Get a summary of the conversation history.

Parameters:

agent (RollbackAgent) – The RollbackAgent instance.

Return type:

str

Returns:

A formatted summary of the conversation.

handle_agent_response(agent, response)[source]

Handle agent response and check for rollback requests.

Parameters:

  • agent (RollbackAgent) – The RollbackAgent that generated the response.

  • response (Any) – The response from the agent.

Return type:

bool

Returns:

True if a rollback was requested and should be handled, False otherwise.

list_checkpoints(internal_session_id)[source]

List all checkpoints for an internal session.

Parameters:

internal_session_id (int) – ID of the internal session.

Return type:

list

Returns:

List of checkpoints.

list_internal_sessions(external_session_id)[source]

List all internal sessions for an external session.

Parameters:

external_session_id (int) – ID of the external session.

Return type:

list

Returns:

List of internal sessions.

resume_agent(external_session_id, internal_session_id=None, base_url=None, api_key=None)[source]

Resume an existing agent session.

Parameters:

  • external_session_id (int) – ID of the external session.

  • internal_session_id (Optional[int]) – Optional specific internal session to resume. If None, uses the current internal session.

  • base_url (Optional[str]) – Optional base URL for the model provider (overrides defaults/env if provided).

  • api_key (Optional[str]) – Optional API key for the model provider (overrides defaults/env if provided).

Return type:

Optional[RollbackAgent]

Returns:

The resumed RollbackAgent instance, or None if not found.

rollback_to_checkpoint(external_session_id, checkpoint_id, base_url=None, api_key=None, rollback_tools=True)[source]

Create a new agent from a checkpoint (rollback operation).

Parameters:

  • external_session_id (int) – ID of the external session.

  • checkpoint_id (int) – ID of the checkpoint to rollback to.

  • base_url (Optional[str]) – Optional base URL for the model provider (overrides defaults/env if provided).

  • api_key (Optional[str]) – Optional API key for the model provider (overrides defaults/env if provided).

  • rollback_tools (bool) – Whether to rollback tool operations after the checkpoint.

Return type:

Optional[RollbackAgent]

Returns:

A new RollbackAgent with the checkpoint’s state, or None if failed.

agentgit.agents.rollback_agent module

LangGraph Agent with rollback capabilities.

Implements a checkpoint-based rollback system for LangGraph agents with time-travel through conversation states and full execution history preservation.

class agentgit.agents.rollback_agent.AgentState[source]

Bases: TypedDict

State definition for the LangGraph agent.

messages

Conversation history as list of messages

session_state

Custom session state dictionary

tool_invocations

History of tool invocations

rollback_requested

Flag indicating if rollback was requested

rollback_checkpoint_id

ID of checkpoint to rollback to

current_turn

Current conversation turn number

user_id

ID of the user owning this session

user_preferences

User preferences for agent behavior

current_turn: int
messages: Annotated[Sequence[BaseMessage]]
rollback_checkpoint_id: Optional[int]
rollback_requested: bool
session_state: Dict[str, Any]
tool_invocations: List[Dict[str, Any]]
user_id: Optional[int]
user_preferences: Dict[str, Any]
class agentgit.agents.rollback_agent.RollbackAgent(external_session_id, model, tools=None, auto_checkpoint=True, internal_session_repo=None, checkpoint_repo=None, checkpointer=None, skip_session_creation=False, reverse_tools=None, user=None, **kwargs)[source]

Bases: object

LangGraph agent with checkpoint and rollback capabilities.

Implements automatic checkpoint creation, database persistence, and non-destructive rollback with branch preservation.

external_session_id

ID of the external session this agent belongs to

internal_session

Current internal session being used

auto_checkpoint

Whether to automatically create checkpoints after tool calls

graph

The compiled LangGraph workflow

tool_rollback_registry

Registry for tool rollback operations

async arun(message, config=None)[source]

Async version of run method.

Parameters:

  • message (str) – The user message to process

  • config (Optional[RunnableConfig]) – Optional LangGraph configuration

Return type:

Any

Returns:

The agent’s response

cleanup_auto_checkpoints_tool(keep_latest=5)[source]

Clean up old automatic checkpoints.

Parameters:

keep_latest (int) – Number of latest checkpoints to keep

Return type:

str

Returns:

Cleanup summary

create_checkpoint_tool(name=None)[source]

Create a manual checkpoint of the current conversation state.

Parameters:

name (Optional[str]) – Optional name for the checkpoint

Return type:

str

Returns:

Confirmation message

delete_checkpoint_tool(checkpoint_id)[source]

Delete a specific checkpoint.

Parameters:

checkpoint_id (int) – ID of the checkpoint to delete

Return type:

str

Returns:

Confirmation message

classmethod from_checkpoint(checkpoint_id, external_session_id, model, checkpoint_repo, internal_session_repo, **kwargs)[source]

Create a new agent from a checkpoint (rollback).

Creates a new internal session branched from the checkpoint, preserving the forward timeline.

Parameters:

  • checkpoint_id (int) – ID of checkpoint to restore from

  • external_session_id (int) – ID of external session

  • model – LangChain model to use

  • checkpoint_repo (CheckpointRepository) – Checkpoint repository

  • internal_session_repo (InternalSessionRepository) – Internal session repository

  • **kwargs – Additional agent configuration

Return type:

RollbackAgent

Returns:

New RollbackAgent with checkpoint’s state

get_checkpoint_info_tool(checkpoint_id)[source]

Get detailed information about a checkpoint.

Parameters:

checkpoint_id (int) – ID of the checkpoint

Return type:

str

Returns:

Detailed checkpoint information

get_conversation_history()[source]

Get conversation history for this session.

Return type:

List[Dict[str, Any]]

Returns:

List of conversation messages

get_session_state()[source]

Get current session state.

Return type:

Dict[str, Any]

Returns:

Session state dictionary

get_tool_track()[source]

Get current recorded tool invocation track.

Return type:

List[Any]

Returns:

List of tool invocation records

list_checkpoints_tool()[source]

List all available checkpoints for the current session.

Return type:

str

Returns:

Formatted list of checkpoints

redo_tools()[source]

Re-execute forward handlers for recorded tools.

Return type:

List[Any]

Returns:

List of new invocation records

rollback_to_checkpoint_tool(checkpoint_id_or_name)[source]

Request rollback to a specific checkpoint.

Parameters:

checkpoint_id_or_name – ID or name of the checkpoint

Return type:

str

Returns:

Status message

rollback_tools()[source]

Run reverse handlers for recorded tools in reverse order.

Return type:

List[Any]

Returns:

List of reverse invocation results

rollback_tools_from_track_index(start_index)[source]

Run reverse handlers from specific track index.

Parameters:

start_index (int) – Index to start rollback from

Return type:

List[Any]

Returns:

List of reverse invocation results

run(message, config=None)[source]

Run the agent and save state after completion.

Parameters:

  • message (str) – The user message to process

  • config (Optional[RunnableConfig]) – Optional LangGraph configuration

Return type:

Any

Returns:

The agent’s response

Module contents