跳到主要内容

第 2 章:执行引擎 —— 一次 execute() 的生命周期

这是整个项目最该读的代码。函数以代码字符串躺在库里,本章讲它怎么被「复活」:取码 → 递归装依赖 → exec → 绑参校验 → 调用 → 记日志 → 放触发器。核心全在 babyagi/functionz/core/execution.py。

2.1 它要解决的小问题​

你库里有一段字符串 "def hello_world():\n x = world()\n return ..."。直接 exec 它会报错——因为 world 没定义。所以执行引擎得在 exec 之前,先把这个函数的所有依赖(其他函数、第三方库)备进同一个作用域。这就是执行引擎的核心难题:环境装配。

2.2 全景:execute() 的七步​

execute(function_name, *args, **kwargs)
  │
  ├─1. db.get_function(name)            取出活跃版本 dict(含 code 字符串)
  ├─2. _add_execution_log('started')    先写一条「开始」日志,拿到 log_id
  ├─3. local_scope = {func, datetime…}  建一个共享命名空间
  ├─4. _resolve_dependencies(...)        递归:装第三方库 + exec 每个依赖函数进 scope
  ├─5. _inject_secret_keys(local_scope)  把所有 API key 塞进 scope
  ├─6. exec(code, local_scope)           复活本函数;绑参、校验、调用 → output
  └─7. _update_execution_log('success')  写成功日志 + 耗时
        └─ _execute_triggered_functions   跑触发器(带防递归)

对应源码主干在 execution.py:55 的 execute 方法,我们逐步拆。

2.3 第 4 步详解:依赖怎么递归装配(最关键)​

_resolve_dependencies 干两件事:装第三方库 和 exec 依赖函数。

(a) 第三方库:缺了就自动 pip install​

对函数声明的每个 import,先试 import_module;ImportError 就当场 pip 安装再导入:

# babyagi/functionz/core/execution.py:15 — _install_external_dependency
def _install_external_dependency(self, package_name, imp_name):
    try:
        return importlib.import_module(imp_name)
    except ImportError:
        subprocess.check_call([sys.executable, "-m", "pip", "install", package_name])
        return importlib.import_module(package_name)

装好的模块对象直接放进 local_scope[imp['name']](execution.py:31-35),这样被 exec 的函数体里 import requests / 直接用 requests 都能命中。这是「函数自带依赖、运行即装」的来源,也是为什么生成的新函数能立刻跑。

(b) 依赖函数:递归 exec 进同一个作用域,并包一层 wrapper​

对每个依赖函数,先递归解析它的依赖(深度优先),再 exec 它的代码进 local_scope,最后把它替换成一个 wrapper:

# babyagi/functionz/core/execution.py:37 — _resolve_dependencies 的依赖循环
for dep_name in function_version.get('dependencies', []):
    if dep_name not in local_scope and dep_name not in visited:
        visited.add(dep_name)                          # 防重复/环
        dep_data = self.python_func.db.get_function(dep_name)
        self._resolve_dependencies(dep_data, local_scope, ...)   # 先装它的依赖
        exec(dep_data['code'], local_scope)            # 复活依赖函数
        local_scope[dep_name] = self._create_function_wrapper(   # 包一层
            local_scope[dep_name], dep_name, parent_log_id, executed_functions)

为什么要包 wrapper?因为当 hello_world 内部调用 world() 时,我们不想让它调那个刚 exec 出来的裸函数——我们想让这次调用也走完整的 execute() 流程(从而 world 自己的依赖也能装、也能记日志)。wrapper 就干这件事:

# babyagi/functionz/core/execution.py:50 — _create_function_wrapper
def _create_function_wrapper(self, func, func_name, parent_log_id, executed_functions):
    def wrapper(*args, **kwargs):
        return self.execute(func_name, *args,           # 转回完整 execute()
                            executed_functions=executed_functions,
                            parent_log_id=parent_log_id, **kwargs)
    return wrapper

直觉: 每个函数调用都被「劫持」回 execute(),于是整棵调用树都享有同样的依赖装配 + 日志记录。visited 集合(execution.py:37)防止依赖图里的环导致无限递归。

2.4 第 6 步:复活本函数、绑参、校验、调用​

依赖备齐后,exec 本函数,再从作用域里把它取出来:

# babyagi/functionz/core/execution.py:122
exec(function_version['code'], local_scope)
func = local_scope[function_name]            # 取出刚复活的函数对象
bound_args = self._bind_function_arguments(func, args, kwargs)   # 用 inspect 绑参
self._validate_input_parameters(function_version, bound_args)    # 校验必填参数
output = func(*bound_args.args, **bound_args.kwargs)             # 真正调用

绑参用 inspect.signature(...).bind(...) 并 apply_defaults()(execution.py:164),所以默认值、关键字参数都正确处理;校验则确保数据库记录的每个 input_parameter 都被提供(execution.py:170)。

2.5 一段「示意」代码,帮你串起来​

下面这段示意、非源码,把上面的环境装配浓缩成可读的样子:

# 示意,非源码 —— 演示「取码→装环境→exec→调用」的核心想法
def execute(name):
    fn = db.get_function(name)              # 1. 取出 dict(含 code 字符串)
    scope = {"func": framework}            # 共享命名空间

    for imp in fn["imports"]:              # 2. 装第三方库(缺了 pip install)
        scope[imp] = import_or_pip_install(imp)

    for dep in fn["dependencies"]:          # 3. 递归 exec 依赖函数
        execute_deps_into(dep, scope)
        scope[dep] = wrap_back_to_execute(dep)   # 包成回调 execute 的 wrapper

    scope.update(all_secret_keys())         # 4. 注入 API key
    exec(fn["code"], scope)                  # 5. 复活本函数
    return scope[name](**bound_args)        # 6. 调用

重点看第 3 步的 wrap_back_to_execute:这是「调用树全程被劫持回 execute」的精髓。

2.6 日志:一棵可追溯的父子树​

每次 execute 都写日志,且用两种父子关系把执行编织成树:

字段含义
parent_log_id谁在调用链上是我的上一层
triggered_by_log_id我是被哪条日志的触发器拉起来的

开始时写一条 'started'(execution.py:96),成功/失败时 update 同一条加上 output 和耗时(execution.py:137 / :147)。get_log_bundle(db_router.py:215)能顺着父/子/兄弟把一次完整执行的所有日志捞回来——dashboard 的执行轨迹就靠它。

一个真实小坑: _update_execution_log 里 update_log 的调用被缩进进了 if time_spent is not None: 块内(execution.py:204-209)。意味着只有当传了 time_spent 才真正落库——大多数路径都传了,但这是个一不留神就漏写日志的脆弱写法。

2.7 触发器:跑完连带跑别的(带防递归)​

函数成功后,查「谁声明了把我当触发器」,逐个连带执行:

# babyagi/functionz/core/execution.py:217 — _execute_triggered_functions
triggered = self.python_func.db.get_triggers_for_function(function_name)
for t in triggered:
    if t in executed_functions:          # 防递归:本链里跑过就跳过
        logger.warning("...Skipping to prevent recursion.")
        continue
    self.execute(t, *trigger_args, triggered_by_log_id=log_id, ...)

防递归靠 executed_functions 这条链——已在本次调用链里出现过的函数不再被触发,避免 A 触发 B、B 又触发 A 的死循环(execution.py:221)。

触发器拿什么参数?规则很朴素:目标函数若声明了入参,就把上游 output 当第一个参数传进去;否则空手调用(_prepare_trigger_arguments,execution.py:247)。

触发器的典型用途见第 3 章的 ai_description_generator:它声明 triggers=["function_added_or_updated"],于是每当有函数被加/改,自动用 LLM 给它生成描述——这是框架「自我维护」的小闭环。

2.8 小结​

  • execute() 的灵魂是 _resolve_dependencies:先把第三方库(缺则 pip 装)和依赖函数都 exec 进同一个 local_scope,再 exec 本函数。
  • 依赖函数被包成 wrapper,使每一层调用都重新走 execute,从而统一享有依赖装配 + 日志。
  • visited 防依赖环,executed_functions 防触发器递归。
  • 日志用 parent_log_id / triggered_by_log_id 织成可回溯的执行树。

下一章:有了这套「存 + 跑」的底座,LLM 怎么往里塞新函数,实现「自己长出自己」。