Skip to content

CodeLoop

pycodeloop.core.codeloop.CodeLoop

Import

You can import the CodeLoop class directly from pycodeloop:

from pycodeloop import CodeLoop, Config
from pycodeloop.providers import GenericProvider
Example

class pycodeloop.core.codeloop.CodeLoop

config = Config(
    provider=GenericProvider.from_json("path/to/config.json")
)

flow = CodeLoop(config=config)
flow.run("list the files in this repo and summarize the project")

Parameters:

Name Type Description Default
config Optional[Config]

Configuration class.

 None

Attributes:

Name Type Description
config Config
agent Agent
session Session

Conversation history, kept across run() calls so the agent remembers earlier turns.
Source code in pycodeloop/core/codeloop.py 
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
class CodeLoop:
    """
    Import:
        You can import the **CodeLoop** class directly from pycodeloop:

            from pycodeloop import CodeLoop, Config
            from pycodeloop.providers import GenericProvider

    Example:
        `class` pycodeloop.core.codeloop.CodeLoop

            config = Config(
                provider=GenericProvider.from_json("path/to/config.json")
            )

            flow = CodeLoop(config=config)
            flow.run("list the files in this repo and summarize the project")

    Args:
        config (Optional[Config]): Configuration class.

    Attributes:
        config (Config):

        agent (Agent):

        session (Session): Conversation history, kept across `run()` calls
            so the agent remembers earlier turns.
    """

    def __init__(self, config: Config | None = None) -> None:
        self.config = config if config else Config()

        agent_kwargs = {}
        if self.config.system_prompt is not None:
            agent_kwargs["system_prompt"] = self.config.system_prompt

        self.agent = Agent(
            provider=self.config.provider,
            tools=self.config.tools,
            max_turns=self.config.max_turns,
            max_history_turns=self.config.max_history_turns,
            **agent_kwargs,
        )

        self.session = Session(system_prompt=self.agent.system_prompt, cwd=os.getcwd())

    def run(
        self,
        prompt: str,
        session_key: str | None = None,
        images: list[str] | None = None,
        cancel_event: threading.Event | None = None,
    ) -> str:
        """Run prompt to completion, keeping history across calls.

        Pass `session_key` with a `Config(storage=...)` configured to
        resume a previous conversation: the session is loaded from
        storage before the run (falling back to the in-memory session
        on a cache miss) and saved back after.
        """
        can_persist = session_key is not None and self.config.storage is not None
        if can_persist:
            stored = self.config.storage.get(session_key)
            self.session = stored or Session(
                system_prompt=self.agent.system_prompt, cwd=os.getcwd()
            )
            self.agent.on_message = lambda: self.config.storage.post(
                session_key, self.session
            )
        else:
            self.agent.on_message = None

        usage_before = self.agent.usage
        result = self.agent.run(
            prompt, session=self.session, images=images, cancel_event=cancel_event
        )
        usage_after = self.agent.usage

        _usage_tracker.record_usage(
            session_key or "global",
            usage_after.input_tokens - usage_before.input_tokens,
            usage_after.output_tokens - usage_before.output_tokens,
        )

        if can_persist:
            self.config.storage.post(session_key, self.session)

        return result

agent = Agent(provider=(self.config.provider), tools=(self.config.tools), max_turns=(self.config.max_turns), max_history_turns=(self.config.max_history_turns), **agent_kwargs) instance-attribute

config = config if config else Config() instance-attribute

session = Session(system_prompt=(self.agent.system_prompt), cwd=(os.getcwd())) instance-attribute

__init__(config=None)

Source code in pycodeloop/core/codeloop.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
def __init__(self, config: Config | None = None) -> None:
    self.config = config if config else Config()

    agent_kwargs = {}
    if self.config.system_prompt is not None:
        agent_kwargs["system_prompt"] = self.config.system_prompt

    self.agent = Agent(
        provider=self.config.provider,
        tools=self.config.tools,
        max_turns=self.config.max_turns,
        max_history_turns=self.config.max_history_turns,
        **agent_kwargs,
    )

    self.session = Session(system_prompt=self.agent.system_prompt, cwd=os.getcwd())

run(prompt, session_key=None, images=None, cancel_event=None)

Run prompt to completion, keeping history across calls.

Pass session_key with a Config(storage=...) configured to resume a previous conversation: the session is loaded from storage before the run (falling back to the in-memory session on a cache miss) and saved back after.

Source code in pycodeloop/core/codeloop.py 
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
def run(
    self,
    prompt: str,
    session_key: str | None = None,
    images: list[str] | None = None,
    cancel_event: threading.Event | None = None,
) -> str:
    """Run prompt to completion, keeping history across calls.

    Pass `session_key` with a `Config(storage=...)` configured to
    resume a previous conversation: the session is loaded from
    storage before the run (falling back to the in-memory session
    on a cache miss) and saved back after.
    """
    can_persist = session_key is not None and self.config.storage is not None
    if can_persist:
        stored = self.config.storage.get(session_key)
        self.session = stored or Session(
            system_prompt=self.agent.system_prompt, cwd=os.getcwd()
        )
        self.agent.on_message = lambda: self.config.storage.post(
            session_key, self.session
        )
    else:
        self.agent.on_message = None

    usage_before = self.agent.usage
    result = self.agent.run(
        prompt, session=self.session, images=images, cancel_event=cancel_event
    )
    usage_after = self.agent.usage

    _usage_tracker.record_usage(
        session_key or "global",
        usage_after.input_tokens - usage_before.input_tokens,
        usage_after.output_tokens - usage_before.output_tokens,
    )

    if can_persist:
        self.config.storage.post(session_key, self.session)

    return result