跳到主要内容

第 3 章:自建 agent —— 让它自己长出函数

前两章建好了「函数即数据 + 即时 exec」的底座。本章是项目卖点:既然函数是数据,LLM 就能生成新函数存回库。看 process_user_input 和 self_build 怎么把一句话需求,变成库里几个新函数并跑出结果。

这些都在 babyagi/functionz/packs/drafts/,作者明确标注是实验性 draft,生成的代码「minimal and may need improvement」。

3.1 核心直觉:复用优先,不够才造​

自建的中心思想不是「每次都让 LLM 写代码」,而是先翻库里有没有现成的,没有才生成——而且生成时鼓励拆成小而可复用的函数。这样库会越用越「肥」,复用率越来越高。

用户一句话需求
   │
   ▼
┌──────────────────────┐  有现成的?
│ check_existing_functions│──── 是 ──► 直接用那个函数
└──────────┬───────────┘
           │ 否
           ▼
┌──────────────────────┐
│ break_down_task       │  LLM 把任务拆成若干小函数(名/描述/参数/依赖)
└──────────┬───────────┘
           ▼
┌──────────────────────┐  逐个函数:
│ generate_functions    │   ├ find_similar_function(向量查重)
│                       │   └ 没有 → create_function → LLM 写代码 → 存回库
└──────────┬───────────┘
           ▼
┌──────────────────────┐
│ extract_function_params│  LLM 从原始需求里抽出调用参数
└──────────┬───────────┘
           ▼
   run_final_function → 输出

这条主线全在 process_user_input(code_writing_functions.py:514)。

3.2 主流程:process_user_input​

# babyagi/functionz/packs/drafts/code_writing_functions.py:514
def process_user_input(user_input):
    result = check_existing_functions(user_input)        # 1. 先查现成
    if result['function_found']:
        function_name = result['function_name']
    else:
        function_breakdown = break_down_task(user_input)  # 2. 拆任务
        context = {'user_input': user_input, 'function_breakdown': function_breakdown}
        generate_functions(function_breakdown, context)   # 3. 逐个生成并存库
        function_name = function_breakdown[0]['name']      # 假设主函数是第一个
    parameters = extract_function_parameters(user_input, function_name)  # 4. 抽参
    return run_final_function(function_name, **parameters)              # 5. 执行

注意第 3 步后的一句强假设:function_name = function_breakdown[0]['name']——它默认 LLM 拆出来的第一个函数就是入口函数。这是个脆弱约定,LLM 拆错顺序就会调错入口。

3.3 关键步骤拆解​

(1) check_existing_functions:LLM 当「检索器」​

把库里所有函数的 name + description 喂给 LLM,问「有没有哪个完美满足需求」,要求返回 JSON。解析失败就 while True 无限重试:

# babyagi/functionz/packs/drafts/code_writing_functions.py:59
response = gpt_call(prompt)
try:
    result = json.loads(response)
    if 'function_found' in result and ...:
        return result
except Exception:
    continue        # 解析失败就一直重试(没有上限)

这个 while True 无限重试模式贯穿整个 draft pack(break_down_task、decide_imports_and_apis、extract_function_parameters 都是)。好处是对 LLM 偶发的格式错误鲁棒;坏处是LLM 持续返回坏 JSON 时会卡死。self_build.py 里的 generate_queries 改良了这点——见下文。

(2) break_down_task:把任务拆成「微服务式」小函数​

prompt 明确要求 LLM 输出一个函数列表,每项含 name / description / input_parameters / output_parameters / dependencies / imports / code(占位),并反复强调「每个函数尽量小、可复用、参数化而非写死」:

# 提示词要点(code_writing_functions.py:84-92,真实 prompt 节选)
# - Each function should be as small as possible and do one thing well.
# - Use existing functions where possible (gpt_call, find_similar_function...).
# - Every sub function ... designed to be reusable by turning things into parameters.

这一步只产出骨架(code 字段是占位),真正的代码在 create_function 里生成。

(3) generate_functions → create_function:写代码并存回库​

对拆出的每个函数,先用向量相似度查库里有没有近似的(find_similar_function);没有才真生成:

# babyagi/functionz/packs/drafts/code_writing_functions.py:389 — generate_functions
for function in function_breakdown:
    similar_functions = find_similar_function(function['description'])
    # ...若没有描述完全一致的近似函数:
    if not function_found:
        create_function(function, function_context)

create_function(code_writing_functions.py:345)三步走:决定 imports → 让 LLM 生成完整代码 → 调 add_new_function 存库:

# create_function 主干
imports_and_apis = decide_imports_and_apis(context)            # 决定需要哪些库/API
function_data = generate_function_code(function, context)      # LLM 写完整代码
add_new_function(name=function_data['function_name'],          # ★存回同一个库
                 code=function_data['code'],
                 dependencies=function_data.get('dependencies', []), ...)

那个 add_new_function(default_functions.py:35)就是第 1 章的注册路径——于是新函数立刻成为库里可执行的一员,下次别的需求就能复用它。这就是「自建」闭环合拢的地方。

generate_function_code 的一个巧思:把「上下文函数」喂给 LLM​

生成代码前,它会把依赖函数的真实代码 + 用了相同 import 的其他函数代码一起塞进 prompt,让 LLM 照着现有风格写、并知道依赖的真实签名:

# babyagi/functionz/packs/drafts/code_writing_functions.py:244 — 收集依赖代码
for dep in dependencies:
    dep_function = get_function_wrapper(dep)
    dependency_code += f"\n# Code for dependency function '{dep}':\n{dep_function['code']}\n"
# ...再收集 import 相同的函数代码作为「同风格参考」(:253)

这让生成的新代码更容易和现有库对得上接口,而不是凭空捏一个签名。

(4) extract_function_parameters:把自然语言变成调用参数​

确定要调哪个函数后,再让 LLM 看着该函数的代码和参数定义,从原始需求里抽出实参 JSON(code_writing_functions.py:418)。这样「Add 5 and 3」就能变成 {"a": 5, "b": 3}。

3.4 self_build:批量造需求,驱动整个闭环​

self_build 比 process_user_input 高一层:给它一句用户画像(如「一个 SaaS 公司的销售」),它先让 LLM 编出 X 个该用户可能问的真实需求,再把每个需求丢给 process_user_input:

# babyagi/functionz/packs/drafts/self_build.py:80 — self_build
queries = generate_queries(user_description, X)       # LLM 造 X 个需求
for query in queries:
    output = process_user_input(query)                 # 每个都走完整自建闭环
    results.append({'query': query, 'output': output})

效果:喂一个角色,库里就自动长出一批该角色会用到的函数。这是 README 那张 self_build.png 演示的东西。

generate_queries:重试有上限(比 draft 更稳的写法)​

和前面的 while True 不同,generate_queries 用有限重试 + 收集错误:

# babyagi/functionz/packs/drafts/self_build.py:45
for attempt in range(1, max_retries + 1):    # 默认 3 次
    response = gpt_call(prompt)
    try:
        queries = json.loads(response)
        if isinstance(queries, list) and len(queries) == X and all(isinstance(q, str) for q in queries):
            return queries
    except json.JSONDecodeError as e:
        errors.append(...)                     # 攒错误信息
raise ValueError(f"Failed ... Errors: {full_error_message}")   # 超限抛出

对比之下这是更负责任的 LLM 调用范式:有上限、失败抛出可诊断的聚合错误,而不是默默死循环。

3.5 小结​

  • 自建闭环 = 复用优先:check_existing_functions / find_similar_function 先查库,不够才让 LLM 拆任务、写代码、add_new_function 存回去。
  • 新函数一旦存库,立刻是可执行、可被复用、可被下次自建发现的一等公民——这就是「自己长出自己」。
  • 工程上仍是 draft:while True 无限重试、「第一个函数即入口」的强假设、生成代码质量靠 prompt 兜底。self_build 的有限重试是其中较成熟的一处。

下一章:把这套设计的巧妙之处、真实的坑、和兄弟项目的对比收个尾。