MCPcopy Create free account
hub / github.com/dot-agent/nextpy / ProgramExecutor

Class ProgramExecutor

nextpy/ai/engine/_program_executor.py:17–753  ·  view source on GitHub ↗

Source from the content-addressed store, hash-verified

15
16
17class ProgramExecutor:
18 def __init__(self, program):
19 """Attaches this executor to a program object."""
20
21 self.program = program
22 self.block_content = []
23 self.executing = True
24 self.should_stop = False
25 self.caught_stop_iteration = False
26 self.llm_session = None
27 self._logging = hasattr(self.program.log, "append")
28
29 # find all the handlebars-style partial inclusion tags and replace them with the partial template
30 # def replace_partial(match):
31 # parts = match.group(1).split(" ", 1)
32 # partial_name = parts[0]
33
34 # # ,args_string = match.group(1).split(" ", 1)
35 # if partial_name not in program._variables:
36 # raise ValueError("Partial '%s' not given in the keyword args:" % partial_name)
37 # out = "{{#block '"+partial_name+"'"
38 # if len(parts) > 1:
39 # out += " " + parts[1]
40 # out += "}}" + program._variables[partial_name].text + "{{/block}}"
41 # # Update the current program variables using those from the partial, but do not overwrite.
42 # # (Rebuilding the _variables map here would break returning new values to the program variables later, e.g. from gen.)
43 # update_variables = {
44 # k: v
45 # for k, v in program[partial_name]._variables.items()
46 # if k not in program._variables
47 # }
48 # program._variables.update(update_variables)
49 # return out
50 # text = re.sub(r"{{>(.*?)}}", replace_partial, program._text)
51
52 # parse the program text
53 try:
54 self.parse_tree = grammar.parse_string(program._text)
55 except (pp.ParseException, pp.ParseSyntaxException) as e:
56 initial_str = program._text[max(0, e.loc - 40) : e.loc]
57 initial_str = initial_str.split("\n")[
58 -1
59 ] # trim off any lines before the error
60 next_str = program._text[e.loc : e.loc + 40]
61 error_string = str(e)
62 if next_str.startswith("{{#") or next_str.startswith("{{~#"):
63 error_string += "\nPerhaps the block command was not correctly closed?"
64 msg = error_string + "\n\n" + initial_str
65 # msg += "\033[91m" + program._text[e.loc:e.loc+40] + "\033[0m\n"
66 msg += program._text[e.loc : e.loc + 40] + "\n"
67 msg += " " * len(initial_str) + "^\n"
68
69 raise SyntaxException(msg, e) from None
70
71 # def _check_for_simple_error(self, text):
72 # """ Check for a simple errors in the program text, and give nice error messages.
73 # """
74

Callers 1

__call__Method · 0.85

Calls

no outgoing calls

Tested by

no test coverage detected