人工智能AI Agent多模态语音AI 应用【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址https://gitcode.com/TEN-framework/ten-framework点击查看免费下载导读aio_http_server_python是 TEN Framework 官方提供的一个 Python 扩展示例它基于 aiohttp 构建了一个异步 HTTP 服务器并展示了如何把外部 HTTP 请求桥接进 TEN 运行时外部客户端通过 HTTP/WebSocket 访问该扩展扩展将请求转换为 TEN 命令Cmd并投递给图Graph中的其他扩展再把命令结果以 HTTP 响应的形式返回给客户端。读完本文你将掌握在 TEN Framework 中编写 Python 异步扩展的完整套路包括扩展注册、生命周期回调、属性读取、命令收发以及如何用一个可复现的集成测试验证整条链路。扩展包概览依据扩展包自带的 README.en-US.md该包的核心定位是Asynchronous HTTP server extension example written in Python for TEN Framework用 Python 为 TEN Framework 编写的异步 HTTP 服务器扩展示例它的关键特性即Python async HTTP server extension也就是说它不只是一个玩具 Hello World而是一个可复用的 HTTP 服务接入范式通过一个AsyncExtension子类在 TEN 运行时内启动 aiohttp 服务并作为图中的一个普通扩展节点与其他扩展协作。从包结构看该扩展由以下文件构成见 包目录文件作用main.py扩展核心实现HTTP 服务启动、请求处理、命令桥接manifest.json包元数据类型、版本、显示名、运行时依赖property.json包默认属性当前为空对象实际运行参数由宿主 App 注入requirements.txt第三方依赖声明aiohttp3.10.2pyproject.tomlPython 工程元数据要求 Python 3.10BUILD.gnTEN 的 GN 构建描述负责打包 extension 及文档资源快速开始前置条件按照 manifest.json 的声明本扩展只有一项必需依赖ten_runtime_pythonsystem 类型版本0.11.73即 TEN Framework 的 Python 运行时绑定。此外Python 环境要求 3.10见 pyproject.toml第三方依赖仅有 requirements.txt 中锁定的aiohttp3.10.2依赖面非常精简。安装安装方式遵循 TEN Framework 的标准包安装流程通过tman工具Ten Manager将扩展包安装到目标应用的ten_packages/extension/目录下或通过tgn配合 BUILD.gn 直接参与构建打包。BUILD.gn 中定义了ten_package(aio_http_server_python)其resources清单会将main.py、manifest.json、property.json、pyproject.toml、requirements.txt、LICENSE以及docs/**下的多语言文档一并打入最终包当开启 ten_manager 时还会生成upload_aio_http_server_python_to_server目标用于发布。使用方式此包需要根据框架规范集成进 TEN 应用在应用级property.json的predefined_graphs中把aio_http_server_python声明为一个扩展节点并配置好属性与命令连接随后该扩展便随应用启动作为一个常驻的异步 HTTP 服务运行。核心实现解析从 HTTP 请求到 TEN 命令扩展的全部核心逻辑都在 main.py 中代码虽短却完整覆盖了 TEN Python 异步扩展的所有关键机制。下面分层解读。1. 扩展类与注册register_addon_as_extension(aio_http_server_python) class DefaultExtensionAddon(Addon): def on_create_instance(self, ten_env, name, context): ten_env.on_create_instance_done(HttpServerExtension(name), context)HttpServerExtension继承自AsyncExtension是异步扩展的入口DefaultExtensionAddon通过register_addon_as_extension(aio_http_server_python)装饰器完成注册注册名与 manifest.json 中的name保持一致并在on_create_instance中把实例创建完成的信号通过on_create_instance_done回传给运行时。这是每个 TEN Python 扩展都必须遵循的注册范式。2. 生命周期回调扩展实现了五个生命周期方法构成完整的异步生命周期闭环on_init保存ten_env句柄供后续所有异步操作使用on_start读取server_port属性随后调用start_server(0.0.0.0, self.server_port)启动 HTTP 服务on_cmd作为命令接收方时返回StatusCode.ERROR本扩展默认不支持被投递命令职责是发起方而非消费方on_stop依次停掉tcpSite与runner释放 aiohttp 资源on_deinit记录日志完成反初始化。在on_start中端口读取使用了 TEN 异步环境的标准属性 APIself.server_port, err await ten_env.get_property_int(server_port) if err is not None: ten_env.log(LogLevel.ERROR, Could not read server_port from properties. str(err)) self.server_port 8002注意其健壮性设计当属性读取失败时会记录错误日志并回退到默认端口8002保证扩展在任何环境下都不会因为缺配置而崩溃。3. HTTP 服务启动def create_runner(self): self.webApp web.Application() self.webApp.add_routes([ web.post(/, self.default_handler), web.get(/ws, self.default_ws_handler), ]) return web.AppRunner(self.webApp) async def start_server(self, host, port): self.runner self.create_runner() await self.runner.setup() self.tcpSite web.TCPSite(self.runner, host, port) await self.tcpSite.start()扩展对外暴露两个端点POST /JSON 请求入口将 HTTP 请求转发为 TEN 命令GET /wsWebSocket 端点用于双向文本消息。绑定地址为0.0.0.0监听端口来自应用注入的server_port属性默认8002。使用AppRunnerTCPSite而非简单的web.run_app是为了把服务器的启停完全收归到扩展自身的on_start/on_stop生命周期内便于 TEN 运行时精确控制资源。4. 请求处理HTTP 到 Cmd 的桥接default_handler是整个扩展的精华它演示了外部 HTTP 世界与TEN 消息世界之间的翻译层data await request.json() method request.method url request.path if ten not in data: return web.Response(status404, textNot found)协议约定请求体必须包含ten字段其内部有两种指令指令一关闭应用if type in data[ten] and data[ten][type] close_app: close_app_cmd Cmd.create(ten:close_app) close_app_cmd.set_dests([Loc()]) asyncio.create_task(self.ten_env.send_cmd(close_app_cmd)) return web.Response(status200, textOK)发送内置命令ten:close_app目标Loc()表示应用根节点。这里使用asyncio.create_task异步投递避免阻塞当前请求协程。指令二转发任意命名命令cmd Cmd.create(data[ten][name]) cmd.set_property_string(method, method) cmd.set_property_string(url, url) cmd_result, _ await self.ten_env.send_cmd(cmd)这是最核心的动态路由能力客户端在 JSON 的ten.name中指定任意命令名扩展动态创建对应Cmd并把 HTTP 的method与url作为命令属性一并携带然后通过await self.ten_env.send_cmd(cmd)投递给图中连接的扩展并等待结果——send_cmd返回(cmd_result, error)二元组。结果回写if cmd_result is not None and cmd_result.get_status_code() StatusCode.OK: detail, err cmd_result.get_property_string(detail) if err is not None: return web.Response(status500, textInternal server error) return web.Response(textdetail) else: return web.Response(status500, textInternal server error)若下游扩展以StatusCode.OK返回命令结果扩展取出结果字符串属性detail作为 HTTP 响应正文返回任何异常结果为空、状态码非 OK、属性读取失败统一映射为 HTTP 500。整个处理过程完全异步不会阻塞 aiohttp 的事件循环。5. WebSocket 处理async def default_ws_handler(self, request): ws web.WebSocketResponse() await ws.prepare(request) async for msg in ws: if msg.type WSMsgType.TEXT: if msg.data close: await ws.close() else: await ws.send_str(some websocket message payload) elif msg.type WSMsgType.ERROR: self.ten_env.log(LogLevel.ERROR, ws connection closed with exception %s % ws.exception()) return wsWebSocket 端点演示了长连接场景收到文本close时主动关闭连接其余文本回送一条示例负载连接异常时通过ten_env.log以LogLevel.ERROR记录错误。配置属性与图的集成方式server_port 属性扩展运行期依赖的唯一属性是server_port监听端口。它并不写在扩展包自身的 property.json该文件目前为空对象{}里而是由宿主应用的图配置注入体现了包属性 vs 实例属性的职责划分。在图中的声明与连接以仓库自带的集成测试应用为例见 测试应用 property.json一个最小可运行的集成配置如下{ ten: { predefined_graphs: [ { name: default, auto_start: true, graph: { nodes: [ { type: extension, name: aio_http_server_python, addon: aio_http_server_python, extension_group: test, property: { server_port: 8002 } }, { type: extension, name: simple_echo_cpp, addon: simple_echo_cpp, extension_group: default_extension_group } ], connections: [ { extension: aio_http_server_python, cmd: [ { name: test, dest: [{ extension: simple_echo_cpp }] } ] } ] } } ] } }该配置揭示了三件事节点声明以addon: aio_http_server_python实例化扩展节点并通过节点级property传入server_port: 8002这正是on_start中get_property_int(server_port)能读到的值命令连接通过connections将本扩展发出的名为test的Cmd路由到simple_echo_cpp扩展——对应default_handler中Cmd.create(data[ten][name])动态创建命令后由运行时按图路由的过程自动启动auto_start: true表示图随应用启动自动拉起。端到端验证集成测试实测链路仓库在 tests/ten_runtime/integration/python/aio_http_server_python/ 提供了完整的集成测试可直接作为如何运行与验证的可执行说明书。核心流程见 test_case.py构建应用build_pkg.prepare_and_build_app依据aio_http_server_python_app目录组装出可运行的应用包引导依赖bootstrap_python_dependencies同步 Python 依赖pyproject.toml 与 requirements启动服务通过uv run main.py启用 sanitizer 时直接使用.venv/bin/python main.py绕过uv拉起应用健康探测http.is_app_started(127.0.0.1, 8002, 30)轮询等待服务在 30 秒内就绪发送请求def http_request(): return http.post(http://127.0.0.1:8002/, {ten: {name: test}})断言与清理断言响应状态码不为 500即证明POST /→ 创建test命令 → 路由到simple_echo_cpp→ 结果回写 HTTP 响应的完整链路畅通随后调用http.stop_app(127.0.0.1, 8002, 30)通过ten:close_app命令优雅关闭应用并断言退出码为 0。这条测试链路同时验证了default_handler的两条指令路径{ten: {name: test}}走命令转发stop_app走ten:close_app关闭流程。典型应用场景从实现机制可以推断该扩展适合作为 TEN 应用对外暴露的HTTP 网关或命令入口将外部系统的 REST 调用转换为 TEN 图中的内部命令例如让外部 Webhook、管理后台直接触发图内扩展的业务逻辑借助cmd.set_property_string(method, method)与cmd.set_property_string(url, url)携带 HTTP 上下文让下游扩展感知请求来源提供 WebSocket 长连接通道用于实时控制或事件推送作为 voice agent 类应用的本地控制面与仓库中 voice-assistant-sip-plivo、voice-assistant-sip-twilio 等示例中server_port的用法一脉相承在 8002 端口对外提供 HTTP API。许可证本扩展包是 TEN Framework 项目的一部分源码采用 Apache License 2.0 许可见 LICENSE与整个仓库的授权方式一致。赞分享人工智能AI Agent多模态语音AI 应用【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址https://gitcode.com/TEN-framework/ten-framework点击查看免费下载相关推荐TEN Framework 异步 HTTP 服务器 Python 扩展实战指南aio_http_server_python 的集成、协议与源码解析TEN Framework 异步 HTTP 服务器 Python 扩展实战指南aio_http_server_python 的集成、协议与源码解析 本指南以人工智能AI Agent多模态语音AI 应用在 TEN Framework 中用 Node.js 构建 HTTP 服务器扩展http_server_extension_nodejs 实战指南在 TEN Framework 中用 Node.js 构建 HTTP 服务器扩展http_server_extension_nodejs 实战指南 导读 本文人工智能AI Agent多模态语音AI 应用TEN Framework Ollama Python 扩展实战异步 LLM 扩展示例与源码解析TEN Framework Ollama Python 扩展实战异步 LLM 扩展示例与源码解析 导读 ollama_python 是 TEN Framewo人工智能AI Agent多模态语音AI 应用上一篇React-media性能优化避免响应式渲染中的常见陷阱下一篇PyGame-Learning-Environment 使用教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 在 TEN Framework 中构建异步 HTTP 服务器 Python 扩展：aio_http_server_python 源码解析与实战
人工智能AI Agent多模态语音AI 应用【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址https://gitcode.com/TEN-framework/ten-framework点击查看免费下载导读aio_http_server_python是 TEN Framework 官方提供的一个 Python 扩展示例它基于 aiohttp 构建了一个异步 HTTP 服务器并展示了如何把外部 HTTP 请求桥接进 TEN 运行时外部客户端通过 HTTP/WebSocket 访问该扩展扩展将请求转换为 TEN 命令Cmd并投递给图Graph中的其他扩展再把命令结果以 HTTP 响应的形式返回给客户端。读完本文你将掌握在 TEN Framework 中编写 Python 异步扩展的完整套路包括扩展注册、生命周期回调、属性读取、命令收发以及如何用一个可复现的集成测试验证整条链路。扩展包概览依据扩展包自带的 README.en-US.md该包的核心定位是Asynchronous HTTP server extension example written in Python for TEN Framework用 Python 为 TEN Framework 编写的异步 HTTP 服务器扩展示例它的关键特性即Python async HTTP server extension也就是说它不只是一个玩具 Hello World而是一个可复用的 HTTP 服务接入范式通过一个AsyncExtension子类在 TEN 运行时内启动 aiohttp 服务并作为图中的一个普通扩展节点与其他扩展协作。从包结构看该扩展由以下文件构成见 包目录文件作用main.py扩展核心实现HTTP 服务启动、请求处理、命令桥接manifest.json包元数据类型、版本、显示名、运行时依赖property.json包默认属性当前为空对象实际运行参数由宿主 App 注入requirements.txt第三方依赖声明aiohttp3.10.2pyproject.tomlPython 工程元数据要求 Python 3.10BUILD.gnTEN 的 GN 构建描述负责打包 extension 及文档资源快速开始前置条件按照 manifest.json 的声明本扩展只有一项必需依赖ten_runtime_pythonsystem 类型版本0.11.73即 TEN Framework 的 Python 运行时绑定。此外Python 环境要求 3.10见 pyproject.toml第三方依赖仅有 requirements.txt 中锁定的aiohttp3.10.2依赖面非常精简。安装安装方式遵循 TEN Framework 的标准包安装流程通过tman工具Ten Manager将扩展包安装到目标应用的ten_packages/extension/目录下或通过tgn配合 BUILD.gn 直接参与构建打包。BUILD.gn 中定义了ten_package(aio_http_server_python)其resources清单会将main.py、manifest.json、property.json、pyproject.toml、requirements.txt、LICENSE以及docs/**下的多语言文档一并打入最终包当开启 ten_manager 时还会生成upload_aio_http_server_python_to_server目标用于发布。使用方式此包需要根据框架规范集成进 TEN 应用在应用级property.json的predefined_graphs中把aio_http_server_python声明为一个扩展节点并配置好属性与命令连接随后该扩展便随应用启动作为一个常驻的异步 HTTP 服务运行。核心实现解析从 HTTP 请求到 TEN 命令扩展的全部核心逻辑都在 main.py 中代码虽短却完整覆盖了 TEN Python 异步扩展的所有关键机制。下面分层解读。1. 扩展类与注册register_addon_as_extension(aio_http_server_python) class DefaultExtensionAddon(Addon): def on_create_instance(self, ten_env, name, context): ten_env.on_create_instance_done(HttpServerExtension(name), context)HttpServerExtension继承自AsyncExtension是异步扩展的入口DefaultExtensionAddon通过register_addon_as_extension(aio_http_server_python)装饰器完成注册注册名与 manifest.json 中的name保持一致并在on_create_instance中把实例创建完成的信号通过on_create_instance_done回传给运行时。这是每个 TEN Python 扩展都必须遵循的注册范式。2. 生命周期回调扩展实现了五个生命周期方法构成完整的异步生命周期闭环on_init保存ten_env句柄供后续所有异步操作使用on_start读取server_port属性随后调用start_server(0.0.0.0, self.server_port)启动 HTTP 服务on_cmd作为命令接收方时返回StatusCode.ERROR本扩展默认不支持被投递命令职责是发起方而非消费方on_stop依次停掉tcpSite与runner释放 aiohttp 资源on_deinit记录日志完成反初始化。在on_start中端口读取使用了 TEN 异步环境的标准属性 APIself.server_port, err await ten_env.get_property_int(server_port) if err is not None: ten_env.log(LogLevel.ERROR, Could not read server_port from properties. str(err)) self.server_port 8002注意其健壮性设计当属性读取失败时会记录错误日志并回退到默认端口8002保证扩展在任何环境下都不会因为缺配置而崩溃。3. HTTP 服务启动def create_runner(self): self.webApp web.Application() self.webApp.add_routes([ web.post(/, self.default_handler), web.get(/ws, self.default_ws_handler), ]) return web.AppRunner(self.webApp) async def start_server(self, host, port): self.runner self.create_runner() await self.runner.setup() self.tcpSite web.TCPSite(self.runner, host, port) await self.tcpSite.start()扩展对外暴露两个端点POST /JSON 请求入口将 HTTP 请求转发为 TEN 命令GET /wsWebSocket 端点用于双向文本消息。绑定地址为0.0.0.0监听端口来自应用注入的server_port属性默认8002。使用AppRunnerTCPSite而非简单的web.run_app是为了把服务器的启停完全收归到扩展自身的on_start/on_stop生命周期内便于 TEN 运行时精确控制资源。4. 请求处理HTTP 到 Cmd 的桥接default_handler是整个扩展的精华它演示了外部 HTTP 世界与TEN 消息世界之间的翻译层data await request.json() method request.method url request.path if ten not in data: return web.Response(status404, textNot found)协议约定请求体必须包含ten字段其内部有两种指令指令一关闭应用if type in data[ten] and data[ten][type] close_app: close_app_cmd Cmd.create(ten:close_app) close_app_cmd.set_dests([Loc()]) asyncio.create_task(self.ten_env.send_cmd(close_app_cmd)) return web.Response(status200, textOK)发送内置命令ten:close_app目标Loc()表示应用根节点。这里使用asyncio.create_task异步投递避免阻塞当前请求协程。指令二转发任意命名命令cmd Cmd.create(data[ten][name]) cmd.set_property_string(method, method) cmd.set_property_string(url, url) cmd_result, _ await self.ten_env.send_cmd(cmd)这是最核心的动态路由能力客户端在 JSON 的ten.name中指定任意命令名扩展动态创建对应Cmd并把 HTTP 的method与url作为命令属性一并携带然后通过await self.ten_env.send_cmd(cmd)投递给图中连接的扩展并等待结果——send_cmd返回(cmd_result, error)二元组。结果回写if cmd_result is not None and cmd_result.get_status_code() StatusCode.OK: detail, err cmd_result.get_property_string(detail) if err is not None: return web.Response(status500, textInternal server error) return web.Response(textdetail) else: return web.Response(status500, textInternal server error)若下游扩展以StatusCode.OK返回命令结果扩展取出结果字符串属性detail作为 HTTP 响应正文返回任何异常结果为空、状态码非 OK、属性读取失败统一映射为 HTTP 500。整个处理过程完全异步不会阻塞 aiohttp 的事件循环。5. WebSocket 处理async def default_ws_handler(self, request): ws web.WebSocketResponse() await ws.prepare(request) async for msg in ws: if msg.type WSMsgType.TEXT: if msg.data close: await ws.close() else: await ws.send_str(some websocket message payload) elif msg.type WSMsgType.ERROR: self.ten_env.log(LogLevel.ERROR, ws connection closed with exception %s % ws.exception()) return wsWebSocket 端点演示了长连接场景收到文本close时主动关闭连接其余文本回送一条示例负载连接异常时通过ten_env.log以LogLevel.ERROR记录错误。配置属性与图的集成方式server_port 属性扩展运行期依赖的唯一属性是server_port监听端口。它并不写在扩展包自身的 property.json该文件目前为空对象{}里而是由宿主应用的图配置注入体现了包属性 vs 实例属性的职责划分。在图中的声明与连接以仓库自带的集成测试应用为例见 测试应用 property.json一个最小可运行的集成配置如下{ ten: { predefined_graphs: [ { name: default, auto_start: true, graph: { nodes: [ { type: extension, name: aio_http_server_python, addon: aio_http_server_python, extension_group: test, property: { server_port: 8002 } }, { type: extension, name: simple_echo_cpp, addon: simple_echo_cpp, extension_group: default_extension_group } ], connections: [ { extension: aio_http_server_python, cmd: [ { name: test, dest: [{ extension: simple_echo_cpp }] } ] } ] } } ] } }该配置揭示了三件事节点声明以addon: aio_http_server_python实例化扩展节点并通过节点级property传入server_port: 8002这正是on_start中get_property_int(server_port)能读到的值命令连接通过connections将本扩展发出的名为test的Cmd路由到simple_echo_cpp扩展——对应default_handler中Cmd.create(data[ten][name])动态创建命令后由运行时按图路由的过程自动启动auto_start: true表示图随应用启动自动拉起。端到端验证集成测试实测链路仓库在 tests/ten_runtime/integration/python/aio_http_server_python/ 提供了完整的集成测试可直接作为如何运行与验证的可执行说明书。核心流程见 test_case.py构建应用build_pkg.prepare_and_build_app依据aio_http_server_python_app目录组装出可运行的应用包引导依赖bootstrap_python_dependencies同步 Python 依赖pyproject.toml 与 requirements启动服务通过uv run main.py启用 sanitizer 时直接使用.venv/bin/python main.py绕过uv拉起应用健康探测http.is_app_started(127.0.0.1, 8002, 30)轮询等待服务在 30 秒内就绪发送请求def http_request(): return http.post(http://127.0.0.1:8002/, {ten: {name: test}})断言与清理断言响应状态码不为 500即证明POST /→ 创建test命令 → 路由到simple_echo_cpp→ 结果回写 HTTP 响应的完整链路畅通随后调用http.stop_app(127.0.0.1, 8002, 30)通过ten:close_app命令优雅关闭应用并断言退出码为 0。这条测试链路同时验证了default_handler的两条指令路径{ten: {name: test}}走命令转发stop_app走ten:close_app关闭流程。典型应用场景从实现机制可以推断该扩展适合作为 TEN 应用对外暴露的HTTP 网关或命令入口将外部系统的 REST 调用转换为 TEN 图中的内部命令例如让外部 Webhook、管理后台直接触发图内扩展的业务逻辑借助cmd.set_property_string(method, method)与cmd.set_property_string(url, url)携带 HTTP 上下文让下游扩展感知请求来源提供 WebSocket 长连接通道用于实时控制或事件推送作为 voice agent 类应用的本地控制面与仓库中 voice-assistant-sip-plivo、voice-assistant-sip-twilio 等示例中server_port的用法一脉相承在 8002 端口对外提供 HTTP API。许可证本扩展包是 TEN Framework 项目的一部分源码采用 Apache License 2.0 许可见 LICENSE与整个仓库的授权方式一致。赞分享人工智能AI Agent多模态语音AI 应用【免费下载链接】ten-frameworkOpen-source framework for conversational voice AI agents项目地址https://gitcode.com/TEN-framework/ten-framework点击查看免费下载相关推荐TEN Framework 异步 HTTP 服务器 Python 扩展实战指南aio_http_server_python 的集成、协议与源码解析TEN Framework 异步 HTTP 服务器 Python 扩展实战指南aio_http_server_python 的集成、协议与源码解析 本指南以人工智能AI Agent多模态语音AI 应用在 TEN Framework 中用 Node.js 构建 HTTP 服务器扩展http_server_extension_nodejs 实战指南在 TEN Framework 中用 Node.js 构建 HTTP 服务器扩展http_server_extension_nodejs 实战指南 导读 本文人工智能AI Agent多模态语音AI 应用TEN Framework Ollama Python 扩展实战异步 LLM 扩展示例与源码解析TEN Framework Ollama Python 扩展实战异步 LLM 扩展示例与源码解析 导读 ollama_python 是 TEN Framewo人工智能AI Agent多模态语音AI 应用上一篇React-media性能优化避免响应式渲染中的常见陷阱下一篇PyGame-Learning-Environment 使用教程创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考