如果你是一名开发者最近在关注 AI 领域的开源动态可能会发现一个现象很多新项目都热衷于构建“全能型”智能体试图用一个模型解决所有问题。但当你真正想把它集成到自己的项目里或者解决一个具体的、垂直的业务需求时常常会感到无从下手——文档复杂、依赖众多、效果难以评估。今天要讨论的EDG-Nobody项目恰恰提供了一个截然不同的思路。它不是一个试图包揽一切的“大而全”框架而是一个高度聚焦、开箱即用的单任务执行引擎。它的核心价值在于将“意图识别”与“任务执行”彻底解耦让开发者能像调用一个函数库一样精准、可靠地驱动 AI 模型去完成特定操作。“信不信我电你”这个标题看似戏谑却精准地隐喻了它的能力它不负责和你聊天辩论意图理解但只要你明确下达“通电”这个指令它就能稳定、高效地执行并且把执行结果清晰地反馈给你。这对于需要将 AI 能力嵌入到现有工作流、自动化脚本或业务系统中的开发者来说意味着更低的集成成本和更高的可控性。本文将带你彻底拆解 EDG-Nobody。我们不会停留在概念层面而是会深入其架构并通过一个从零开始的完整实战示例展示如何用它来驱动 AI 模型执行一个具体的文件处理任务。你会看到在“智能体”概念喧嚣的背后一种更务实、更工程化的 AI 应用范式正在悄然兴起。1. 这篇文章真正要解决的问题当AI需要“干活”时我们缺什么在当前的 AI 应用开发中我们常常面临一个断层上游的 LLM大语言模型负责理解和生成语言下游的业务系统有明确的 API 和操作逻辑。如何让 LLM 的“思考”转化为系统可执行的“动作”是中间缺失的关键一环。传统的做法是写一个庞大的提示词Prompt试图让 LLM 自己理解任务、自己决定调用哪个工具、自己生成参数。这种方式存在几个明显痛点可靠性差LLM 的输出具有不确定性可能格式错误、调用错误工具或生成非法参数。调试困难当任务执行失败时很难定位是提示词问题、模型理解问题还是工具本身的问题。安全性低LLM 可能被诱导执行危险操作因为工具调用的逻辑和边界控制是模糊的。集成复杂需要为每个任务编写复杂的胶水代码处理各种异常和格式转换。EDG-Nobody 的解决思路是“契约式执行”。它要求开发者预先、明确地定义好一个“任务契约”Task这个任务叫什么、需要什么输入参数、具体执行什么逻辑。然后它提供一个标准的执行引擎外部系统可以是另一个LLM也可以是规则引擎只需要按契约“点名”并传入参数Nobody 就会严格按定义执行并返回结构化的结果。这相当于为 AI 驱动的自动化建立了一套“API 标准”。开发者不再需要教 AI“如何操作电脑”而是告诉 AI“我这里有‘重命名文件’、‘发送邮件’、‘查询数据库’这几个标准接口你告诉我用哪个、参数是什么剩下的我来保证执行。”接下来我们就从核心概念开始看看这套机制是如何运转的。2. 基础概念与核心原理Task, Skill 与 Executor要理解 EDG-Nobody必须掌握三个核心概念Task任务、Skill技能和Executor执行器。它们构成了一个清晰的分层架构。概念角色类比关键特点Task任务契约/接口像编程中的“函数声明”。它定义了任务的目的、所需的输入参数名称、类型、描述和输出的格式。纯定义不包含具体实现。是外部系统与 Nobody 交互的合同。Skill技能实现/逻辑像编程中的“函数实现”。它包含了完成某个 Task 所需的具体代码逻辑如调用操作系统命令、访问网络API、处理数据等。一个 Skill 实现一个或多个 Task。是“干活”的主体。Executor执行器运行时/引擎像编程中的“函数调用器”。它负责加载 Skills接收对外部发起的 Task 执行请求找到对应的 Skill传入参数执行代码并返回结果。管理 Skill 的生命周期提供统一的执行、监控和错误处理环境。它们之间的关系与工作流如下定义阶段开发者编写一个 Task 定义描述要做什么然后编写一个 Skill 来实现它具体怎么做。注册阶段将编写好的 Skill 注册到 Executor 中。执行阶段外部系统向 Executor 发起请求“请执行名为rename_file的 Task并传入参数old_path‘a.txt‘和new_path‘b.txt‘。”路由与执行Executor 根据 Task 名找到已注册的对应 Skill将参数传递给它并运行其代码。返回结果Skill 执行完毕后将结果返回给 ExecutorExecutor 再以结构化的方式如 JSON返回给调用方。这种架构的核心优势在于关注点分离LLM或规则引擎只负责“决策”根据用户意图选择合适的 Task 并生成合规的参数。它不需要关心具体实现细节。Nobody 只负责“执行”保证 Task 被正确、安全、稳定地执行。它不关心决策过程。这使得整个系统更易于维护、测试和扩展。你可以独立地优化提示词决策层也可以独立地升级 Skill 的实现执行层两者通过清晰的 Task 契约进行通信。3. 环境准备与前置条件在开始实战之前我们需要准备好 Python 开发环境。EDG-Nobody 是一个 Python 库因此你需要Python 版本建议使用 Python 3.8 及以上版本。你可以通过以下命令检查python --version # 或 python3 --version包管理工具使用pip进行安装。建议使用虚拟环境如venv或conda来隔离项目依赖。# 创建虚拟环境以 venv 为例 python -m venv nobody_env # 激活虚拟环境 # Windows: nobody_env\Scripts\activate # Linux/Mac: source nobody_env/bin/activate安装 EDG-Nobody通过 pip 从 PyPI 安装。pip install edg-nobody安装成功后你可以验证一下pip show edg-nobody这会显示包的版本和基本信息。我们的目标是创建一个简单的 Skill它能够读取一个文本文件使用 LLM 总结其内容然后将总结写入另一个文件。这个例子涵盖了文件操作和 AI 模型调用是一个典型的自动化场景。4. 核心流程拆解从定义到执行的五步让我们把创建并使用一个自定义 Skill 的过程分解为五个清晰的步骤步骤 1定义 Task我们要做什么明确任务接口任务名、输入参数文件名、输出结果。步骤 2实现 Skill我们怎么做编写 Python 类实现具体的业务逻辑读文件、调用 LLM、写文件。步骤 3注册 Skill让系统知道将我们写好的 Skill 类注册到 Executor 中。步骤 4执行 Task发起请求编写客户端代码向 Executor 发起执行我们定义的 Task 的请求。步骤 5验证结果检查输出检查生成的文件是否符合预期处理可能出现的错误。下面我们将按照这五个步骤完成一个完整的示例。5. 完整示例与代码实现构建一个文件总结 Skill假设我们有一个目录里面有很多会议记录文本文件。我们想用一个 Skill 来自动总结每个文件的核心内容并将总结保存到另一个目录。5.1 项目结构首先创建项目文件夹并组织文件。file_summarizer_project/ ├── skills/ # 存放自定义Skill的目录 │ ├── __init__.py │ └── file_summary_skill.py ├── tasks/ # 存放Task定义的目录可选逻辑分离 │ └── file_tasks.py ├── main.py # 主程序注册并执行Skill ├── input/ # 存放待总结的文本文件 │ └── meeting_notes_20240415.txt └── output/ # 存放总结后的输出文件5.2 步骤一定义 Task我们在tasks/file_tasks.py中定义 Task。Task 使用 Pydantic 模型来定义强类型的输入输出。# file: tasks/file_tasks.py from typing import Any from pydantic import BaseModel, Field from edg.nobody import Task # 定义输入参数模型 class FileSummaryInput(BaseModel): 文件总结任务的输入参数 input_file_path: str Field(..., description待总结的原始文本文件的完整路径) output_file_path: str Field(..., description总结内容要保存到的文件完整路径) language: str Field(defaultzh, description总结输出的语言例如 zh 或 en) # 定义输出结果模型 class FileSummaryOutput(BaseModel): 文件总结任务的输出结果 success: bool Field(..., description任务是否执行成功) summary: str Field(default, description生成的总结文本) error_message: str Field(default, description如果失败错误信息是什么) # 创建 Task 实例 # 这里我们给Task起名summarize_text_file并关联其输入输出模型 summarize_file_task Task( namesummarize_text_file, description读取一个文本文件使用AI总结其内容并将总结写入另一个文件。, input_modelFileSummaryInput, output_modelFileSummaryOutput )关键点解释FileSummaryInput和FileSummaryOutput使用了 Pydantic这能自动进行数据验证和序列化。Task对象将名称、描述和输入输出模型绑定在一起形成了一个完整的“契约”。description字段非常重要它可以帮助上游的 LLM 理解这个 Task 的用途。5.3 步骤二实现 Skill接下来在skills/file_summary_skill.py中实现 Skill。我们需要继承BaseSkill类并使用skill装饰器来关联它和刚才定义的 Task。# file: skills/file_summary_skill.py import os from typing import Any from edg.nobody import BaseSkill, skill from ..tasks.file_tasks import summarize_file_task, FileSummaryInput, FileSummaryOutput # 假设我们使用 OpenAI 的 API需要安装 openai 库: pip install openai # 注意这里仅为示例实际使用时请妥善管理你的 API Key。 from openai import OpenAI class FileSummarySkill(BaseSkill): 一个用于总结文本文件内容的Skill。 # 使用装饰器声明这个Skill实现了哪个Task skill(tasksummarize_file_task) async def summarize_text_file(self, input_data: FileSummaryInput) - FileSummaryOutput: 技能的核心执行逻辑。 try: # 1. 读取输入文件 if not os.path.exists(input_data.input_file_path): return FileSummaryOutput( successFalse, error_messagef输入文件不存在: {input_data.input_file_path} ) with open(input_data.input_file_path, r, encodingutf-8) as f: file_content f.read() if not file_content.strip(): return FileSummaryOutput( successFalse, error_message输入文件为空无法总结。 ) # 2. 调用 LLM 进行总结 (示例使用 OpenAI GPT-3.5) # 重要在实际项目中API Key 应从环境变量或安全配置中读取不要硬编码 client OpenAI(api_keyos.environ.get(OPENAI_API_KEY)) prompt f 请用{input_data.language}总结以下文本的主要内容要求简洁明了不超过200字 {file_content} response client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.5, max_tokens300 ) summary response.choices[0].message.content.strip() # 3. 将总结写入输出文件 output_dir os.path.dirname(input_data.output_file_path) if output_dir and not os.path.exists(output_dir): os.makedirs(output_dir) with open(input_data.output_file_path, w, encodingutf-8) as f: f.write(summary) # 4. 返回成功结果 return FileSummaryOutput( successTrue, summarysummary, error_message ) except Exception as e: # 捕获所有未预料的异常 return FileSummaryOutput( successFalse, summary, error_messagef技能执行过程中发生错误: {str(e)} )关键点解释继承与装饰器Skill 类继承BaseSkill核心方法用skill(task...)装饰这是 Nobody 框架识别 Skill 的标准方式。异步方法Skill 的执行方法通常是async的以支持 I/O 密集型操作如网络请求、文件读写。强类型参数方法参数input_data的类型就是我们在 Task 中定义的FileSummaryInput框架会确保传入的数据符合模型规范。结构化返回返回值必须是FileSummaryOutput类型这保证了输出格式的确定性。错误处理在 Skill 内部进行细致的错误处理如文件不存在、内容为空、API调用失败并返回结构化的错误信息这是生产级应用的关键。5.4 步骤三与四注册并执行 Skill现在我们在main.py中把所有的部分组装起来。# file: main.py import asyncio import os from edg.nobody import Executor from skills.file_summary_skill import FileSummarySkill from tasks.file_tasks import FileSummaryInput async def main(): # 1. 创建 Executor 实例 executor Executor() # 2. 创建我们的 Skill 实例 file_summary_skill FileSummarySkill() # 3. 将 Skill 注册到 Executor # Executor 会自动发现 Skill 中用 skill 装饰的方法并将其关联的 Task 注册进来。 executor.register_skill(file_summary_skill) # 4. 准备输入参数 input_params FileSummaryInput( input_file_path./input/meeting_notes_20240415.txt, output_file_path./output/summary_20240415.txt, languagezh ) # 5. 执行 Task # 调用 execute_task 方法传入 Task 名称和参数。 result await executor.execute_task( task_namesummarize_text_file, # 必须与 Task 定义中的 name 完全一致 input_datainput_params ) # 6. 处理结果 print(任务执行完成) print(f成功: {result.success}) if result.success: print(f总结内容: {result.summary}) print(f总结已保存至: {input_params.output_file_path}) else: print(f错误信息: {result.error_message}) if __name__ __main__: # 设置你的 OpenAI API Key (示例请使用安全的方式) os.environ[OPENAI_API_KEY] your-openai-api-key-here # 运行异步主函数 asyncio.run(main())5.5 准备输入文件在input/meeting_notes_20240415.txt中放入一些文本内容例如项目周会纪要 (2024-04-15) 参会人员张三、李四、王五、赵六。 议题 1. 后端API性能优化方案讨论。李四提出引入缓存层预计可将平均响应时间从 120ms 降低至 40ms。王五建议对数据库慢查询进行索引优化。 2. 前端组件库升级计划。赵六展示了新组件库的Demo一致同意在下个迭代周期开始迁移。 3. 下阶段里程碑确定。目标是在五月底完成核心功能闭环并发布内部测试版。 后续行动 - 李四负责编写缓存方案详细设计文档截止04-18。 - 王五负责分析并优化前10条慢查询SQL截止04-19。 - 全体成员审阅前端迁移计划截止04-17。6. 运行结果与效果验证6.1 运行程序在项目根目录下确保虚拟环境已激活且依赖已安装edg-nobody,openai,pydantic然后运行python main.py6.2 预期输出如果一切顺利你将在控制台看到类似以下输出任务执行完成 成功: True 总结内容: 本次项目周会主要讨论了后端API性能优化、前端组件库升级以及下阶段里程碑规划。后端方面计划引入缓存层和优化数据库索引以提升性能前端决定在下一迭代周期迁移至新组件库。会议确定了五月底完成核心功能并发布内部测试版的目标并分配了相应的后续行动任务包括编写缓存设计文档、优化慢查询SQL等。 总结已保存至: ./output/summary_20240415.txt同时检查output/目录会发现新生成了summary_20240415.txt文件内容就是上面的总结文本。6.3 如何验证成功控制台输出success字段为True且输出了总结内容。文件系统在指定的输出路径生成了新文件且内容非空、符合总结预期。错误处理验证你可以尝试修改main.py中的输入文件路径为一个不存在的文件程序会捕获错误并返回successFalse以及相应的错误信息而不会崩溃。这证明了 Skill 内部错误处理的有效性。7. 常见问题与排查思路在开发和运行过程中你可能会遇到以下问题问题现象可能原因排查方式解决方案导入错误ModuleNotFoundError: No module named ‘edg‘1.edg-nobody未安装。2. 在错误的 Python 环境未激活虚拟环境中运行。1. 运行pip list | grep edg-nobody。2. 检查终端提示符是否在虚拟环境中。1. 在正确的虚拟环境中执行pip install edg-nobody。2. 确保激活了项目专用的虚拟环境。执行时报错Task ‘xxx‘ not found1. Task 名称拼写错误。2. Skill 未正确注册到 Executor。3. Skill 中的skill装饰器未正确关联 Task。1. 检查execute_task调用时的task_name字符串。2. 检查executor.register_skill()是否被调用。3. 检查 Skill 方法上的skill(task…)是否引用了正确的 Task 对象。1. 确保task_name与 Task 定义中的name属性完全一致大小写敏感。2. 确保register_skill在execute_task之前被调用。3. 确保from … import …路径正确Task 对象是同一个。Skill 方法未被执行但也没报错Skill 的执行方法是异步的 (async def)但被同步调用了反之亦然或者 Executor 运行在错误的事件循环中。检查main()函数是否用asyncio.run()运行。检查 Skill 方法是否定义为async。1. 主入口使用asyncio.run(main())。2. 确保 Skill 中实现 Task 的方法是async的。3. 确保execute_task被await。OpenAI API 调用失败1. API Key 未设置或错误。2. 网络问题。3. 额度不足或模型不可用。1. 检查环境变量OPENAI_API_KEY。2. 尝试用curl或简单脚本测试 API。3. 查看 OpenAI 控制台账单和状态。1. 使用os.environ[“OPENAI_API_KEY”] “sk-…”或在.env文件中设置。2. 检查代理或防火墙设置。3. 更换 API Key 或检查额度。输入参数验证失败传入execute_task的input_data字典或对象不符合 Task 输入模型 (FileSummaryInput) 的定义。查看 Executor 返回的错误信息通常会指出哪个字段缺失或类型错误。确保传入的参数是FileSummaryInput的实例或者是一个完全匹配的字典。使用 Pydantic 模型的dict()或json()方法确保格式正确。文件操作权限错误程序对指定的输入文件没有读取权限或对输出目录没有写入权限。检查文件路径是否存在以及 Python 进程的用户权限。使用绝对路径并确保程序有足够的权限。在代码中添加更详细的路径检查和创建目录的逻辑。8. 最佳实践与工程建议将 EDG-Nobody 用于实际项目时遵循以下建议可以避免很多坑Task 设计要“高内聚、低耦合”一个 Task 应该只做一件明确的事情。例如“总结文件”是一个 Task“总结文件并发送邮件”最好是两个独立的 Task。这样每个 Skill 更简单也更容易复用和组合。Skill 实现要健壮充分的错误处理如示例所示对文件是否存在、内容是否为空、API 调用是否成功等都要进行检查和容错。资源管理确保打开的文件、网络连接等资源被正确关闭可以使用with语句或try…finally。超时控制对于网络请求等可能长时间阻塞的操作设置合理的超时时间。安全管理是重中之重永远不要硬编码密钥将 API Key、数据库密码等敏感信息存储在环境变量或专业的密钥管理服务中。限制 Skill 权限在 Docker 容器或具有最小权限的系统用户下运行 Executor。Skill 能执行系统命令这意味着如果 Skill 代码被恶意注入或利用风险很高。对输入进行消毒如果 Task 参数涉及文件路径、系统命令等必须进行严格的验证防止路径遍历../../../etc/passwd或命令注入攻击。配置化管理将 Task 名称、文件路径、模型参数等提取到配置文件如config.yaml或.env中而不是散落在代码里。日志与监控在 Skill 和 Executor 中添加详细的日志记录包括任务开始、结束、耗时、输入参数脱敏后和结果状态。这对于调试和运维至关重要。可以考虑将执行记录成功/失败、耗时上报到监控系统。测试策略单元测试 Skill像测试普通函数一样测试每个 Skill 的逻辑可以 Mock 掉外部依赖如文件系统、API 客户端。集成测试 Task测试从 Executor 调用 Task 的完整流程使用测试专用的配置和资源。与 LLM 协作的模式工具描述生成可以利用 Task 的name、description和输入模型的字段描述自动生成供 LLM 使用的“工具描述”如 OpenAI 的 Function Calling 格式。这是 Nobody 架构与 LLM 联动的天然优势。参数校验前置在将 LLM 生成的参数交给 Nobody 执行前可以先用 Pydantic 模型进行预校验给出更友好的错误提示而不是等到 Skill 执行时才失败。EDG-Nobody 代表的是一种“确定性执行层”的思想。它并不取代 LLM 的创造性而是为 LLM 的决策提供一个安全、可靠、可预测的执行环境。当你需要构建严肃的、生产级的 AI 应用时这种将“思考”与“行动”分离的架构往往是更稳健的选择。它让 AI 更像一个拥有标准操作手册的助手而不是一个行为不可预测的黑盒。 EDG-Nobody：构建AI驱动的确定性任务执行引擎
如果你是一名开发者最近在关注 AI 领域的开源动态可能会发现一个现象很多新项目都热衷于构建“全能型”智能体试图用一个模型解决所有问题。但当你真正想把它集成到自己的项目里或者解决一个具体的、垂直的业务需求时常常会感到无从下手——文档复杂、依赖众多、效果难以评估。今天要讨论的EDG-Nobody项目恰恰提供了一个截然不同的思路。它不是一个试图包揽一切的“大而全”框架而是一个高度聚焦、开箱即用的单任务执行引擎。它的核心价值在于将“意图识别”与“任务执行”彻底解耦让开发者能像调用一个函数库一样精准、可靠地驱动 AI 模型去完成特定操作。“信不信我电你”这个标题看似戏谑却精准地隐喻了它的能力它不负责和你聊天辩论意图理解但只要你明确下达“通电”这个指令它就能稳定、高效地执行并且把执行结果清晰地反馈给你。这对于需要将 AI 能力嵌入到现有工作流、自动化脚本或业务系统中的开发者来说意味着更低的集成成本和更高的可控性。本文将带你彻底拆解 EDG-Nobody。我们不会停留在概念层面而是会深入其架构并通过一个从零开始的完整实战示例展示如何用它来驱动 AI 模型执行一个具体的文件处理任务。你会看到在“智能体”概念喧嚣的背后一种更务实、更工程化的 AI 应用范式正在悄然兴起。1. 这篇文章真正要解决的问题当AI需要“干活”时我们缺什么在当前的 AI 应用开发中我们常常面临一个断层上游的 LLM大语言模型负责理解和生成语言下游的业务系统有明确的 API 和操作逻辑。如何让 LLM 的“思考”转化为系统可执行的“动作”是中间缺失的关键一环。传统的做法是写一个庞大的提示词Prompt试图让 LLM 自己理解任务、自己决定调用哪个工具、自己生成参数。这种方式存在几个明显痛点可靠性差LLM 的输出具有不确定性可能格式错误、调用错误工具或生成非法参数。调试困难当任务执行失败时很难定位是提示词问题、模型理解问题还是工具本身的问题。安全性低LLM 可能被诱导执行危险操作因为工具调用的逻辑和边界控制是模糊的。集成复杂需要为每个任务编写复杂的胶水代码处理各种异常和格式转换。EDG-Nobody 的解决思路是“契约式执行”。它要求开发者预先、明确地定义好一个“任务契约”Task这个任务叫什么、需要什么输入参数、具体执行什么逻辑。然后它提供一个标准的执行引擎外部系统可以是另一个LLM也可以是规则引擎只需要按契约“点名”并传入参数Nobody 就会严格按定义执行并返回结构化的结果。这相当于为 AI 驱动的自动化建立了一套“API 标准”。开发者不再需要教 AI“如何操作电脑”而是告诉 AI“我这里有‘重命名文件’、‘发送邮件’、‘查询数据库’这几个标准接口你告诉我用哪个、参数是什么剩下的我来保证执行。”接下来我们就从核心概念开始看看这套机制是如何运转的。2. 基础概念与核心原理Task, Skill 与 Executor要理解 EDG-Nobody必须掌握三个核心概念Task任务、Skill技能和Executor执行器。它们构成了一个清晰的分层架构。概念角色类比关键特点Task任务契约/接口像编程中的“函数声明”。它定义了任务的目的、所需的输入参数名称、类型、描述和输出的格式。纯定义不包含具体实现。是外部系统与 Nobody 交互的合同。Skill技能实现/逻辑像编程中的“函数实现”。它包含了完成某个 Task 所需的具体代码逻辑如调用操作系统命令、访问网络API、处理数据等。一个 Skill 实现一个或多个 Task。是“干活”的主体。Executor执行器运行时/引擎像编程中的“函数调用器”。它负责加载 Skills接收对外部发起的 Task 执行请求找到对应的 Skill传入参数执行代码并返回结果。管理 Skill 的生命周期提供统一的执行、监控和错误处理环境。它们之间的关系与工作流如下定义阶段开发者编写一个 Task 定义描述要做什么然后编写一个 Skill 来实现它具体怎么做。注册阶段将编写好的 Skill 注册到 Executor 中。执行阶段外部系统向 Executor 发起请求“请执行名为rename_file的 Task并传入参数old_path‘a.txt‘和new_path‘b.txt‘。”路由与执行Executor 根据 Task 名找到已注册的对应 Skill将参数传递给它并运行其代码。返回结果Skill 执行完毕后将结果返回给 ExecutorExecutor 再以结构化的方式如 JSON返回给调用方。这种架构的核心优势在于关注点分离LLM或规则引擎只负责“决策”根据用户意图选择合适的 Task 并生成合规的参数。它不需要关心具体实现细节。Nobody 只负责“执行”保证 Task 被正确、安全、稳定地执行。它不关心决策过程。这使得整个系统更易于维护、测试和扩展。你可以独立地优化提示词决策层也可以独立地升级 Skill 的实现执行层两者通过清晰的 Task 契约进行通信。3. 环境准备与前置条件在开始实战之前我们需要准备好 Python 开发环境。EDG-Nobody 是一个 Python 库因此你需要Python 版本建议使用 Python 3.8 及以上版本。你可以通过以下命令检查python --version # 或 python3 --version包管理工具使用pip进行安装。建议使用虚拟环境如venv或conda来隔离项目依赖。# 创建虚拟环境以 venv 为例 python -m venv nobody_env # 激活虚拟环境 # Windows: nobody_env\Scripts\activate # Linux/Mac: source nobody_env/bin/activate安装 EDG-Nobody通过 pip 从 PyPI 安装。pip install edg-nobody安装成功后你可以验证一下pip show edg-nobody这会显示包的版本和基本信息。我们的目标是创建一个简单的 Skill它能够读取一个文本文件使用 LLM 总结其内容然后将总结写入另一个文件。这个例子涵盖了文件操作和 AI 模型调用是一个典型的自动化场景。4. 核心流程拆解从定义到执行的五步让我们把创建并使用一个自定义 Skill 的过程分解为五个清晰的步骤步骤 1定义 Task我们要做什么明确任务接口任务名、输入参数文件名、输出结果。步骤 2实现 Skill我们怎么做编写 Python 类实现具体的业务逻辑读文件、调用 LLM、写文件。步骤 3注册 Skill让系统知道将我们写好的 Skill 类注册到 Executor 中。步骤 4执行 Task发起请求编写客户端代码向 Executor 发起执行我们定义的 Task 的请求。步骤 5验证结果检查输出检查生成的文件是否符合预期处理可能出现的错误。下面我们将按照这五个步骤完成一个完整的示例。5. 完整示例与代码实现构建一个文件总结 Skill假设我们有一个目录里面有很多会议记录文本文件。我们想用一个 Skill 来自动总结每个文件的核心内容并将总结保存到另一个目录。5.1 项目结构首先创建项目文件夹并组织文件。file_summarizer_project/ ├── skills/ # 存放自定义Skill的目录 │ ├── __init__.py │ └── file_summary_skill.py ├── tasks/ # 存放Task定义的目录可选逻辑分离 │ └── file_tasks.py ├── main.py # 主程序注册并执行Skill ├── input/ # 存放待总结的文本文件 │ └── meeting_notes_20240415.txt └── output/ # 存放总结后的输出文件5.2 步骤一定义 Task我们在tasks/file_tasks.py中定义 Task。Task 使用 Pydantic 模型来定义强类型的输入输出。# file: tasks/file_tasks.py from typing import Any from pydantic import BaseModel, Field from edg.nobody import Task # 定义输入参数模型 class FileSummaryInput(BaseModel): 文件总结任务的输入参数 input_file_path: str Field(..., description待总结的原始文本文件的完整路径) output_file_path: str Field(..., description总结内容要保存到的文件完整路径) language: str Field(defaultzh, description总结输出的语言例如 zh 或 en) # 定义输出结果模型 class FileSummaryOutput(BaseModel): 文件总结任务的输出结果 success: bool Field(..., description任务是否执行成功) summary: str Field(default, description生成的总结文本) error_message: str Field(default, description如果失败错误信息是什么) # 创建 Task 实例 # 这里我们给Task起名summarize_text_file并关联其输入输出模型 summarize_file_task Task( namesummarize_text_file, description读取一个文本文件使用AI总结其内容并将总结写入另一个文件。, input_modelFileSummaryInput, output_modelFileSummaryOutput )关键点解释FileSummaryInput和FileSummaryOutput使用了 Pydantic这能自动进行数据验证和序列化。Task对象将名称、描述和输入输出模型绑定在一起形成了一个完整的“契约”。description字段非常重要它可以帮助上游的 LLM 理解这个 Task 的用途。5.3 步骤二实现 Skill接下来在skills/file_summary_skill.py中实现 Skill。我们需要继承BaseSkill类并使用skill装饰器来关联它和刚才定义的 Task。# file: skills/file_summary_skill.py import os from typing import Any from edg.nobody import BaseSkill, skill from ..tasks.file_tasks import summarize_file_task, FileSummaryInput, FileSummaryOutput # 假设我们使用 OpenAI 的 API需要安装 openai 库: pip install openai # 注意这里仅为示例实际使用时请妥善管理你的 API Key。 from openai import OpenAI class FileSummarySkill(BaseSkill): 一个用于总结文本文件内容的Skill。 # 使用装饰器声明这个Skill实现了哪个Task skill(tasksummarize_file_task) async def summarize_text_file(self, input_data: FileSummaryInput) - FileSummaryOutput: 技能的核心执行逻辑。 try: # 1. 读取输入文件 if not os.path.exists(input_data.input_file_path): return FileSummaryOutput( successFalse, error_messagef输入文件不存在: {input_data.input_file_path} ) with open(input_data.input_file_path, r, encodingutf-8) as f: file_content f.read() if not file_content.strip(): return FileSummaryOutput( successFalse, error_message输入文件为空无法总结。 ) # 2. 调用 LLM 进行总结 (示例使用 OpenAI GPT-3.5) # 重要在实际项目中API Key 应从环境变量或安全配置中读取不要硬编码 client OpenAI(api_keyos.environ.get(OPENAI_API_KEY)) prompt f 请用{input_data.language}总结以下文本的主要内容要求简洁明了不超过200字 {file_content} response client.chat.completions.create( modelgpt-3.5-turbo, messages[{role: user, content: prompt}], temperature0.5, max_tokens300 ) summary response.choices[0].message.content.strip() # 3. 将总结写入输出文件 output_dir os.path.dirname(input_data.output_file_path) if output_dir and not os.path.exists(output_dir): os.makedirs(output_dir) with open(input_data.output_file_path, w, encodingutf-8) as f: f.write(summary) # 4. 返回成功结果 return FileSummaryOutput( successTrue, summarysummary, error_message ) except Exception as e: # 捕获所有未预料的异常 return FileSummaryOutput( successFalse, summary, error_messagef技能执行过程中发生错误: {str(e)} )关键点解释继承与装饰器Skill 类继承BaseSkill核心方法用skill(task...)装饰这是 Nobody 框架识别 Skill 的标准方式。异步方法Skill 的执行方法通常是async的以支持 I/O 密集型操作如网络请求、文件读写。强类型参数方法参数input_data的类型就是我们在 Task 中定义的FileSummaryInput框架会确保传入的数据符合模型规范。结构化返回返回值必须是FileSummaryOutput类型这保证了输出格式的确定性。错误处理在 Skill 内部进行细致的错误处理如文件不存在、内容为空、API调用失败并返回结构化的错误信息这是生产级应用的关键。5.4 步骤三与四注册并执行 Skill现在我们在main.py中把所有的部分组装起来。# file: main.py import asyncio import os from edg.nobody import Executor from skills.file_summary_skill import FileSummarySkill from tasks.file_tasks import FileSummaryInput async def main(): # 1. 创建 Executor 实例 executor Executor() # 2. 创建我们的 Skill 实例 file_summary_skill FileSummarySkill() # 3. 将 Skill 注册到 Executor # Executor 会自动发现 Skill 中用 skill 装饰的方法并将其关联的 Task 注册进来。 executor.register_skill(file_summary_skill) # 4. 准备输入参数 input_params FileSummaryInput( input_file_path./input/meeting_notes_20240415.txt, output_file_path./output/summary_20240415.txt, languagezh ) # 5. 执行 Task # 调用 execute_task 方法传入 Task 名称和参数。 result await executor.execute_task( task_namesummarize_text_file, # 必须与 Task 定义中的 name 完全一致 input_datainput_params ) # 6. 处理结果 print(任务执行完成) print(f成功: {result.success}) if result.success: print(f总结内容: {result.summary}) print(f总结已保存至: {input_params.output_file_path}) else: print(f错误信息: {result.error_message}) if __name__ __main__: # 设置你的 OpenAI API Key (示例请使用安全的方式) os.environ[OPENAI_API_KEY] your-openai-api-key-here # 运行异步主函数 asyncio.run(main())5.5 准备输入文件在input/meeting_notes_20240415.txt中放入一些文本内容例如项目周会纪要 (2024-04-15) 参会人员张三、李四、王五、赵六。 议题 1. 后端API性能优化方案讨论。李四提出引入缓存层预计可将平均响应时间从 120ms 降低至 40ms。王五建议对数据库慢查询进行索引优化。 2. 前端组件库升级计划。赵六展示了新组件库的Demo一致同意在下个迭代周期开始迁移。 3. 下阶段里程碑确定。目标是在五月底完成核心功能闭环并发布内部测试版。 后续行动 - 李四负责编写缓存方案详细设计文档截止04-18。 - 王五负责分析并优化前10条慢查询SQL截止04-19。 - 全体成员审阅前端迁移计划截止04-17。6. 运行结果与效果验证6.1 运行程序在项目根目录下确保虚拟环境已激活且依赖已安装edg-nobody,openai,pydantic然后运行python main.py6.2 预期输出如果一切顺利你将在控制台看到类似以下输出任务执行完成 成功: True 总结内容: 本次项目周会主要讨论了后端API性能优化、前端组件库升级以及下阶段里程碑规划。后端方面计划引入缓存层和优化数据库索引以提升性能前端决定在下一迭代周期迁移至新组件库。会议确定了五月底完成核心功能并发布内部测试版的目标并分配了相应的后续行动任务包括编写缓存设计文档、优化慢查询SQL等。 总结已保存至: ./output/summary_20240415.txt同时检查output/目录会发现新生成了summary_20240415.txt文件内容就是上面的总结文本。6.3 如何验证成功控制台输出success字段为True且输出了总结内容。文件系统在指定的输出路径生成了新文件且内容非空、符合总结预期。错误处理验证你可以尝试修改main.py中的输入文件路径为一个不存在的文件程序会捕获错误并返回successFalse以及相应的错误信息而不会崩溃。这证明了 Skill 内部错误处理的有效性。7. 常见问题与排查思路在开发和运行过程中你可能会遇到以下问题问题现象可能原因排查方式解决方案导入错误ModuleNotFoundError: No module named ‘edg‘1.edg-nobody未安装。2. 在错误的 Python 环境未激活虚拟环境中运行。1. 运行pip list | grep edg-nobody。2. 检查终端提示符是否在虚拟环境中。1. 在正确的虚拟环境中执行pip install edg-nobody。2. 确保激活了项目专用的虚拟环境。执行时报错Task ‘xxx‘ not found1. Task 名称拼写错误。2. Skill 未正确注册到 Executor。3. Skill 中的skill装饰器未正确关联 Task。1. 检查execute_task调用时的task_name字符串。2. 检查executor.register_skill()是否被调用。3. 检查 Skill 方法上的skill(task…)是否引用了正确的 Task 对象。1. 确保task_name与 Task 定义中的name属性完全一致大小写敏感。2. 确保register_skill在execute_task之前被调用。3. 确保from … import …路径正确Task 对象是同一个。Skill 方法未被执行但也没报错Skill 的执行方法是异步的 (async def)但被同步调用了反之亦然或者 Executor 运行在错误的事件循环中。检查main()函数是否用asyncio.run()运行。检查 Skill 方法是否定义为async。1. 主入口使用asyncio.run(main())。2. 确保 Skill 中实现 Task 的方法是async的。3. 确保execute_task被await。OpenAI API 调用失败1. API Key 未设置或错误。2. 网络问题。3. 额度不足或模型不可用。1. 检查环境变量OPENAI_API_KEY。2. 尝试用curl或简单脚本测试 API。3. 查看 OpenAI 控制台账单和状态。1. 使用os.environ[“OPENAI_API_KEY”] “sk-…”或在.env文件中设置。2. 检查代理或防火墙设置。3. 更换 API Key 或检查额度。输入参数验证失败传入execute_task的input_data字典或对象不符合 Task 输入模型 (FileSummaryInput) 的定义。查看 Executor 返回的错误信息通常会指出哪个字段缺失或类型错误。确保传入的参数是FileSummaryInput的实例或者是一个完全匹配的字典。使用 Pydantic 模型的dict()或json()方法确保格式正确。文件操作权限错误程序对指定的输入文件没有读取权限或对输出目录没有写入权限。检查文件路径是否存在以及 Python 进程的用户权限。使用绝对路径并确保程序有足够的权限。在代码中添加更详细的路径检查和创建目录的逻辑。8. 最佳实践与工程建议将 EDG-Nobody 用于实际项目时遵循以下建议可以避免很多坑Task 设计要“高内聚、低耦合”一个 Task 应该只做一件明确的事情。例如“总结文件”是一个 Task“总结文件并发送邮件”最好是两个独立的 Task。这样每个 Skill 更简单也更容易复用和组合。Skill 实现要健壮充分的错误处理如示例所示对文件是否存在、内容是否为空、API 调用是否成功等都要进行检查和容错。资源管理确保打开的文件、网络连接等资源被正确关闭可以使用with语句或try…finally。超时控制对于网络请求等可能长时间阻塞的操作设置合理的超时时间。安全管理是重中之重永远不要硬编码密钥将 API Key、数据库密码等敏感信息存储在环境变量或专业的密钥管理服务中。限制 Skill 权限在 Docker 容器或具有最小权限的系统用户下运行 Executor。Skill 能执行系统命令这意味着如果 Skill 代码被恶意注入或利用风险很高。对输入进行消毒如果 Task 参数涉及文件路径、系统命令等必须进行严格的验证防止路径遍历../../../etc/passwd或命令注入攻击。配置化管理将 Task 名称、文件路径、模型参数等提取到配置文件如config.yaml或.env中而不是散落在代码里。日志与监控在 Skill 和 Executor 中添加详细的日志记录包括任务开始、结束、耗时、输入参数脱敏后和结果状态。这对于调试和运维至关重要。可以考虑将执行记录成功/失败、耗时上报到监控系统。测试策略单元测试 Skill像测试普通函数一样测试每个 Skill 的逻辑可以 Mock 掉外部依赖如文件系统、API 客户端。集成测试 Task测试从 Executor 调用 Task 的完整流程使用测试专用的配置和资源。与 LLM 协作的模式工具描述生成可以利用 Task 的name、description和输入模型的字段描述自动生成供 LLM 使用的“工具描述”如 OpenAI 的 Function Calling 格式。这是 Nobody 架构与 LLM 联动的天然优势。参数校验前置在将 LLM 生成的参数交给 Nobody 执行前可以先用 Pydantic 模型进行预校验给出更友好的错误提示而不是等到 Skill 执行时才失败。EDG-Nobody 代表的是一种“确定性执行层”的思想。它并不取代 LLM 的创造性而是为 LLM 的决策提供一个安全、可靠、可预测的执行环境。当你需要构建严肃的、生产级的 AI 应用时这种将“思考”与“行动”分离的架构往往是更稳健的选择。它让 AI 更像一个拥有标准操作手册的助手而不是一个行为不可预测的黑盒。