Skip to content

LlmAgent

dandy.llm.BaseLlmAgent

Bases: BaseLlmBot, BaseAgent, ABC, Generic[IntelType]

config = 'DEFAULT' class-attribute instance-attribute

config_options = LlmConfigOptions() class-attribute instance-attribute

description = None class-attribute instance-attribute

instructions_prompt = Prompt("You're a helpful assistant please follow the users instructions.") class-attribute instance-attribute

intel_class = DefaultLlmIntel class-attribute instance-attribute

plan_time_limit_seconds = settings.DEFAULT_AGENT_PLAN_TIME_LIMIT_SECONDS class-attribute instance-attribute

plan_task_count_limit = settings.DEFAULT_AGENT_PLAN_TASK_COUNT_LIMIT class-attribute instance-attribute

processors = (LlmBot,) class-attribute instance-attribute

process classmethod

Source code in dandy/llm/agent/llm_agent.py 
 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
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
@classmethod
def process(
        cls,
        prompt: PromptOrStr,
        intel_class: Union[Type[IntelType], None] = None,
        intel_object: Union[IntelType, None] = None,
        images: Union[List[str], None] = None,
        image_files: Union[List[str | Path], None] = None,
        include_fields: Union[IncEx, None] = None,
        exclude_fields: Union[IncEx, None] = None,
        postfix_system_prompt: PromptOrStrOrNone = None,
        message_history: Union[MessageHistory, None] = None,
) -> IntelType:

    recorder_event_id = cls._recorder_event_id

    recorder_add_llm_agent_create_plan_event(
        prompt,
        cls._processors_strategy,
        recorder_event_id
    )

    plan = cls._create_plan(prompt)

    recorder_add_llm_agent_finished_creating_plan_event(
        plan,
        recorder_event_id
    )

    recorder_add_llm_agent_running_plan_event(
        plan,
        recorder_event_id
    )

    while plan.is_incomplete:
        if plan.has_exceeded_time_limit:
            raise AgentOverThoughtRecoverableException(
                f'{cls.__name__} exceeded the time limit of {cls.plan_time_limit_seconds} seconds running a plan.'
            )

        task = plan.active_task

        recorder_add_llm_agent_start_task_event(
            task,
            cls._processors_strategy,
            recorder_event_id
        )

        resource = cls._processors_strategy.get_processor_from_key(task.processors_key)

        updated_task = resource.use(
            prompt=agent_do_task_prompt(task),
            intel_object=task,
            include_fields={'actual_result'}
        )

        task.actual_result = updated_task.actual_result
        plan.set_active_task_complete()

        recorder_add_llm_agent_completed_task_event(
            task,
            cls._processors_strategy,
            recorder_event_id
        )

    recorder_add_llm_agent_done_executing_plan_event(
        plan,
        recorder_event_id
    )

    recorder_add_llm_agent_processing_final_result_event(
        plan,
        recorder_event_id
    )

    if postfix_system_prompt is None:
        postfix_system_prompt = Prompt()

    postfix_system_prompt.text(f'Use the results of the below simulated plan to accomplish the user request:')
    postfix_system_prompt.line_break()
    postfix_system_prompt.prompt(plan.to_prompt())

    return cls.process_prompt_to_intel(
        prompt=prompt,
        intel_class=intel_class or cls.intel_class,
        intel_object=intel_object,
        images=images,
        image_files=image_files,
        include_fields=include_fields,
        exclude_fields=exclude_fields,
        postfix_system_prompt=postfix_system_prompt,
        message_history=message_history,
    )