拓冰网站定制 
如何在 pipecat 运行时在多个 LLM provider 之间切换LLMSwitcher【免费下载链接】pipecatOpen Source framework for voice agents, multimodal apps, and realtime AI. Maintained by Daily and the community.项目地址: https://gitcode.com/GitHub_Trending/pi/pipecat在基于 pipecat 构建的语音 Agent 中如果对话中途需要更换 LLM provider——比如从一个供应商切换到另一个或者主供应商故障时自动降级——不需要重建 pipelinepipecat 提供了 LLMSwitcher把多个LLMService实例放进同一个切换器组件在运行中通过控制帧或故障策略切换当前生效的 LLM。本文基于仓库中的两个示例 features-service-switcher.py 和 llm_switching.py 说明如何搭建、触发切换并验证结果。适用前提Python 3.11仓库 README 建议 3.12并已按仓库说明配置好本地开发环境。准备环境安装 pipecat 框架仓库 README.md 给出的方式uv add pipecat-ai复制环境变量模板并填入你计划使用的服务的 API key仓库根目录的 env.examplecp env.example .env两个示例用到的 key 不同features-service-switcher.py 使用OPENAI_API_KEY、GOOGLE_API_KEY、CARTESIA_API_KEY、DEEPGRAM_API_KEY示例中通过os.environ读取缺失会直接报错llm_switching.py 的 docstring 明确列出需要CARTESIA_API_KEYTTS、DEEPGRAM_API_KEYSTT、OPENAI_API_KEY、GOOGLE_API_KEY、ANTHROPIC_API_KEY以及 AWS 凭据AWS Bedrock LLMDAILY_API_KEY为 transport 可选。运行示例时默认使用 SmallWebRTC transport浏览器打开 http://localhost:7860/client/ 点击 Connect 即可与 bot 通话。示例也支持-t daily或-t twilio -x NGROK_HOST_NAME等其他 transport见 examples/README.md。LLMSwitcher 是什么LLMSwitcher是 ServiceSwitcher 的 LLM 专用封装源码位于 src/pipecat/pipeline/llm_switcher.py本质是一条并行 pipeline每个成员 LLM 被一对过滤器包住只有当前激活的那个 LLM 会收到帧。关键事实来自源码 docstring构造签名LLMSwitcher(llms[...], strategy_type...)strategy_type默认是ServiceSwitcherStrategyManual列表中的第一个 LLM 是初始激活的 LLM通过active_llm属性读取当前激活的 LLM切换由ServiceSwitcherFrame派生的控制帧驱动手动切换使用的帧是 ManuallySwitchServiceFrameservice指定目标服务通过on_service_switched事件回调感知切换发生。切换会被拒绝的两种情况ServiceSwitcher._set_active_if_available目标服务不在列表中或者目标服务is_usable为 False此时日志会提示 Not switching to ...: it can no longer do its job需要先恢复该服务。最短主路径手动切换 LLM下面摘录自 examples/features/features-service-switcher.py 的关键部分完整可运行版本在该文件中。它把 STT、LLM、TTS 三个环节都做成了切换器这里只看 LLM 部分from pipecat.frames.frames import LLMRunFrame, ManuallySwitchServiceFrame from pipecat.pipeline.llm_switcher import LLMSwitcher from pipecat.pipeline.pipeline import Pipeline from pipecat.services.google.llm import GoogleLLMService from pipecat.services.openai.llm import OpenAILLMService system_prompt You are a helpful assistant in a voice conversation. ... llm_openai OpenAILLMService( api_keyos.environ[OPENAI_API_KEY], settingsOpenAILLMService.Settings(system_instructionsystem_prompt), ) llm_google GoogleLLMService( api_keyos.environ[GOOGLE_API_KEY], settingsGoogleLLMService.Settings(system_instructionsystem_prompt), ) # 默认使用 ServiceSwitcherStrategyManual llm_switcher LLMSwitcher(llms[llm_openai, llm_google]) pipeline Pipeline( [ transport.input(), # Transport user input stt_switcher, user_aggregator, # User responses llm_switcher, # LLM tts_switcher, # TTS transport.output(), # Transport bot output assistant_aggregator, # Assistant spoken responses ] )pipeline 位置固定在 user aggregator 之后、TTS 之前。llms[llm_openai, llm_google]决定了初始激活的是 OpenAI列表顺序也决定了 failover 策略的备用顺序见下文。触发切换在 transport 事件回调里把ManuallySwitchServiceFrame塞进 worker 队列即可切换以下片段摘自同一示例的on_client_connected处理transport.event_handler(on_client_connected) async def on_client_connected(transport, client): # ... 开始对话 ... await worker.queue_frames([LLMRunFrame()]) await asyncio.sleep(15) print(fSwitching to {llm_google}) await worker.queue_frames([ManuallySwitchServiceFrame(servicellm_google)])queue_frames之后帧沿 pipeline 下行到达llm_switcherServiceSwitcherStrategyManual收到ManuallySwitchServiceFrame后把激活服务切换成frame.service。运行这个示例后客户端连接 15 秒终端会打印Switching to ...示例代码自身的 print 语句随后 bot 的回复改由对应 LLM 生成。可选分支在对话中通过工具切换如果切换时机由用户指令决定换到 Google而不是代码定时触发参考 examples/flows/python/llm_switching.py。它把 4 个 LLMOpenAI、Google、Anthropic、AWS Bedrock放进一个切换器llm_switcher LLMSwitcher( llms[llm_openai, llm_google, llm_anthropic, llm_aws], strategy_typeServiceSwitcherStrategyManual, ) # FlowManager 直接使用 llm_switcher 作为 llm flow_manager FlowManager( workerworker, llmllm_switcher, context_aggregatorcontext_aggregator, )切换逻辑写成一个对话工具用户说话时由 LLM 调用async def switch_llm(flow_manager: FlowManager, llm: str) - tuple[SwitchLLMResult, None]: Switch the current LLM service. Args: llm: The name of the LLM service to switch to. Must be one of OpenAI, Google, Anthropic, or AWS. # 按名称映射到对应的 LLM 实例 ... if llm_switcher.active_llm new_llm: return SwitchLLMResult(statussuccess, messagefAlready using {llm} LLM service.), None # 在工具调用里把切换帧从 aggregator 向上游推 # 保证切换在 LLM 用工具结果继续推理之前发生 await context_aggregator.assistant().push_frame( ManuallySwitchServiceFrame(servicenew_llm), FrameDirection.UPSTREAM ) return SwitchLLMResult(statussuccess, messagefSwitched to {llm} LLM service.), None注意它与定时切换写法的一个差异这里切换帧是从 assistant context aggregator向上游push_frame的。示例注释解释了原因——工具调用会触发 aggregator 的上游更新从 aggregator 推上游可以保证切换先于带工具结果再次运行 LLM发生。切换成功后工具返回statussuccess以及Switched to {llm} LLM service.之类的消息已在当前 LLM 上时返回Already using {llm} LLM service.这些字符串由代码直接构造可以作为判断切换意图是否被执行的依据。可选分支故障时自动 failover手动策略只响应显式的ManuallySwitchServiceFrame。如果希望某个 provider 挂掉时自动切到下一个把策略换成ServiceSwitcherStrategyFailover定义在 src/pipecat/pipeline/service_switcher.pyllm_switcher LLMSwitcher( llms[llm_openai, llm_google], strategy_typeServiceSwitcherStrategyFailover, ) llm_switcher.strategy.event_handler(on_service_switched) async def on_switched(strategy, service): # 应用自己决定何时/如何恢复失败的 LLM ...该策略的行为源码 docstring 描述只有当激活的 LLM 报出一个让它is_usableFalse的错误即再也做不了这份工作才切换它能扛过去的错误不会触发 failover按构造时列表的顺序取下一个仍然可用的 LLM从头尾回绕失败的 LLM 仍保留在列表里之后用FrameProcessor.set_usable把它恢复后就能再切回来恢复/回退策略留给你通过on_service_switched事件自己实现。验证切换是否生效文档给出的可核对点有以下几处active_llm属性任何时候llm_switcher.active_llm返回当前生效的 LLM 实例llm_switching 示例正是用它判断是否已经在目标 LLM 上on_service_switched事件注册在 strategy 上的事件处理器在激活服务变化时被调用这是代码层面确认切换发生的通知点日志与返回消息手动示例在每次切换前打印Switching to ...failover 路径下切走时日志会输出Service {name} reported an error: ...找不到可用备用时输出No other service available to switch to单元测试仓库自带 tests/test_llm_switcher.py覆盖direct function 会注册到每个成员 LLM设置更新会到达未激活的成员等契约。按 README.md 的运行方式只跑这个测试套件uv run pytest tests/test_llm_switcher.py切换过程中工具与设置的保持跨 provider 切换最容易踩的坑是 function callingLLMSwitcher对工具注册做了专门的广播源码中有两个事实LLMContext(tools[...])中列出的 direct functions 会在每个成员 LLM 上注册——无论激活与否。原因是成员 LLM 各自被分支过滤器挡住运行时只有激活的 LLM 能收到LLMContextFrame切换器在收到上下文帧时主动把所有成员的 tool handler 同步一遍llm_switcher.py 的process_frame保证切换后工具继续可用。features-service-switcher 示例就是靠这一机制让get_current_weather等工具在切换前后都工作llm_switcher.register_function(...)同样会转发到所有成员 LLMcancel_on_interruption、timeout_secs等参数按显式参数 tool_options装饰器 默认值在每个成员上解析LLMSwitcher.register_direct_function自 1.4.0 起被标记为 deprecated计划 2.0.0 移除官方建议改用LLMContext(tools[...])或推LLMSetToolsFrame在会话中途变更工具运行时更新 LLM 设置如LLMUpdateSettingsFrame带reach_inactive_servicesTrue时更新会同时送达未激活的成员避免某个 LLM 被激活后缺了配置tests/test_llm_switcher.py 中有对应测试。限制与边界目标 LLM 必须在传给LLMSwitcher的列表里且is_usable为 True否则切换被忽略目标不在列表时可能本意是给 pipeline 里另一个 switcher 的直接被忽略切换器自身不接受set_usable它报告的可用性是所有成员 LLM 的读数把某个成员恢复为 usable 即可带动整个切换器恢复非激活 LLM 的错误会被切换器就地吸收不会再向上传播只有激活 LLM 的错误会交给策略决定是否切换切换成功后该错误也不再向上传播两个示例的切换都是会话内行为LLM 实例在启动时就全部构造好API key 需提前配齐运行中新增成员 LLM 的路径文档未提供需要自行处理。完整代码以仓库内文件为准examples/features/features-service-switcher.py三环节手动切换的最小示例与 examples/flows/python/llm_switching.py对话工具驱动切换 Flows 集成Flows 示例的运行方式见 examples/flows/README.md。【免费下载链接】pipecatOpen Source framework for voice agents, multimodal apps, and realtime AI. Maintained by Daily and the community.项目地址: https://gitcode.com/GitHub_Trending/pi/pipecat创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考 如何在 pipecat 运行时在多个 LLM provider 之间切换（LLMSwitcher）？
如何在 pipecat 运行时在多个 LLM provider 之间切换LLMSwitcher【免费下载链接】pipecatOpen Source framework for voice agents, multimodal apps, and realtime AI. Maintained by Daily and the community.项目地址: https://gitcode.com/GitHub_Trending/pi/pipecat在基于 pipecat 构建的语音 Agent 中如果对话中途需要更换 LLM provider——比如从一个供应商切换到另一个或者主供应商故障时自动降级——不需要重建 pipelinepipecat 提供了 LLMSwitcher把多个LLMService实例放进同一个切换器组件在运行中通过控制帧或故障策略切换当前生效的 LLM。本文基于仓库中的两个示例 features-service-switcher.py 和 llm_switching.py 说明如何搭建、触发切换并验证结果。适用前提Python 3.11仓库 README 建议 3.12并已按仓库说明配置好本地开发环境。准备环境安装 pipecat 框架仓库 README.md 给出的方式uv add pipecat-ai复制环境变量模板并填入你计划使用的服务的 API key仓库根目录的 env.examplecp env.example .env两个示例用到的 key 不同features-service-switcher.py 使用OPENAI_API_KEY、GOOGLE_API_KEY、CARTESIA_API_KEY、DEEPGRAM_API_KEY示例中通过os.environ读取缺失会直接报错llm_switching.py 的 docstring 明确列出需要CARTESIA_API_KEYTTS、DEEPGRAM_API_KEYSTT、OPENAI_API_KEY、GOOGLE_API_KEY、ANTHROPIC_API_KEY以及 AWS 凭据AWS Bedrock LLMDAILY_API_KEY为 transport 可选。运行示例时默认使用 SmallWebRTC transport浏览器打开 http://localhost:7860/client/ 点击 Connect 即可与 bot 通话。示例也支持-t daily或-t twilio -x NGROK_HOST_NAME等其他 transport见 examples/README.md。LLMSwitcher 是什么LLMSwitcher是 ServiceSwitcher 的 LLM 专用封装源码位于 src/pipecat/pipeline/llm_switcher.py本质是一条并行 pipeline每个成员 LLM 被一对过滤器包住只有当前激活的那个 LLM 会收到帧。关键事实来自源码 docstring构造签名LLMSwitcher(llms[...], strategy_type...)strategy_type默认是ServiceSwitcherStrategyManual列表中的第一个 LLM 是初始激活的 LLM通过active_llm属性读取当前激活的 LLM切换由ServiceSwitcherFrame派生的控制帧驱动手动切换使用的帧是 ManuallySwitchServiceFrameservice指定目标服务通过on_service_switched事件回调感知切换发生。切换会被拒绝的两种情况ServiceSwitcher._set_active_if_available目标服务不在列表中或者目标服务is_usable为 False此时日志会提示 Not switching to ...: it can no longer do its job需要先恢复该服务。最短主路径手动切换 LLM下面摘录自 examples/features/features-service-switcher.py 的关键部分完整可运行版本在该文件中。它把 STT、LLM、TTS 三个环节都做成了切换器这里只看 LLM 部分from pipecat.frames.frames import LLMRunFrame, ManuallySwitchServiceFrame from pipecat.pipeline.llm_switcher import LLMSwitcher from pipecat.pipeline.pipeline import Pipeline from pipecat.services.google.llm import GoogleLLMService from pipecat.services.openai.llm import OpenAILLMService system_prompt You are a helpful assistant in a voice conversation. ... llm_openai OpenAILLMService( api_keyos.environ[OPENAI_API_KEY], settingsOpenAILLMService.Settings(system_instructionsystem_prompt), ) llm_google GoogleLLMService( api_keyos.environ[GOOGLE_API_KEY], settingsGoogleLLMService.Settings(system_instructionsystem_prompt), ) # 默认使用 ServiceSwitcherStrategyManual llm_switcher LLMSwitcher(llms[llm_openai, llm_google]) pipeline Pipeline( [ transport.input(), # Transport user input stt_switcher, user_aggregator, # User responses llm_switcher, # LLM tts_switcher, # TTS transport.output(), # Transport bot output assistant_aggregator, # Assistant spoken responses ] )pipeline 位置固定在 user aggregator 之后、TTS 之前。llms[llm_openai, llm_google]决定了初始激活的是 OpenAI列表顺序也决定了 failover 策略的备用顺序见下文。触发切换在 transport 事件回调里把ManuallySwitchServiceFrame塞进 worker 队列即可切换以下片段摘自同一示例的on_client_connected处理transport.event_handler(on_client_connected) async def on_client_connected(transport, client): # ... 开始对话 ... await worker.queue_frames([LLMRunFrame()]) await asyncio.sleep(15) print(fSwitching to {llm_google}) await worker.queue_frames([ManuallySwitchServiceFrame(servicellm_google)])queue_frames之后帧沿 pipeline 下行到达llm_switcherServiceSwitcherStrategyManual收到ManuallySwitchServiceFrame后把激活服务切换成frame.service。运行这个示例后客户端连接 15 秒终端会打印Switching to ...示例代码自身的 print 语句随后 bot 的回复改由对应 LLM 生成。可选分支在对话中通过工具切换如果切换时机由用户指令决定换到 Google而不是代码定时触发参考 examples/flows/python/llm_switching.py。它把 4 个 LLMOpenAI、Google、Anthropic、AWS Bedrock放进一个切换器llm_switcher LLMSwitcher( llms[llm_openai, llm_google, llm_anthropic, llm_aws], strategy_typeServiceSwitcherStrategyManual, ) # FlowManager 直接使用 llm_switcher 作为 llm flow_manager FlowManager( workerworker, llmllm_switcher, context_aggregatorcontext_aggregator, )切换逻辑写成一个对话工具用户说话时由 LLM 调用async def switch_llm(flow_manager: FlowManager, llm: str) - tuple[SwitchLLMResult, None]: Switch the current LLM service. Args: llm: The name of the LLM service to switch to. Must be one of OpenAI, Google, Anthropic, or AWS. # 按名称映射到对应的 LLM 实例 ... if llm_switcher.active_llm new_llm: return SwitchLLMResult(statussuccess, messagefAlready using {llm} LLM service.), None # 在工具调用里把切换帧从 aggregator 向上游推 # 保证切换在 LLM 用工具结果继续推理之前发生 await context_aggregator.assistant().push_frame( ManuallySwitchServiceFrame(servicenew_llm), FrameDirection.UPSTREAM ) return SwitchLLMResult(statussuccess, messagefSwitched to {llm} LLM service.), None注意它与定时切换写法的一个差异这里切换帧是从 assistant context aggregator向上游push_frame的。示例注释解释了原因——工具调用会触发 aggregator 的上游更新从 aggregator 推上游可以保证切换先于带工具结果再次运行 LLM发生。切换成功后工具返回statussuccess以及Switched to {llm} LLM service.之类的消息已在当前 LLM 上时返回Already using {llm} LLM service.这些字符串由代码直接构造可以作为判断切换意图是否被执行的依据。可选分支故障时自动 failover手动策略只响应显式的ManuallySwitchServiceFrame。如果希望某个 provider 挂掉时自动切到下一个把策略换成ServiceSwitcherStrategyFailover定义在 src/pipecat/pipeline/service_switcher.pyllm_switcher LLMSwitcher( llms[llm_openai, llm_google], strategy_typeServiceSwitcherStrategyFailover, ) llm_switcher.strategy.event_handler(on_service_switched) async def on_switched(strategy, service): # 应用自己决定何时/如何恢复失败的 LLM ...该策略的行为源码 docstring 描述只有当激活的 LLM 报出一个让它is_usableFalse的错误即再也做不了这份工作才切换它能扛过去的错误不会触发 failover按构造时列表的顺序取下一个仍然可用的 LLM从头尾回绕失败的 LLM 仍保留在列表里之后用FrameProcessor.set_usable把它恢复后就能再切回来恢复/回退策略留给你通过on_service_switched事件自己实现。验证切换是否生效文档给出的可核对点有以下几处active_llm属性任何时候llm_switcher.active_llm返回当前生效的 LLM 实例llm_switching 示例正是用它判断是否已经在目标 LLM 上on_service_switched事件注册在 strategy 上的事件处理器在激活服务变化时被调用这是代码层面确认切换发生的通知点日志与返回消息手动示例在每次切换前打印Switching to ...failover 路径下切走时日志会输出Service {name} reported an error: ...找不到可用备用时输出No other service available to switch to单元测试仓库自带 tests/test_llm_switcher.py覆盖direct function 会注册到每个成员 LLM设置更新会到达未激活的成员等契约。按 README.md 的运行方式只跑这个测试套件uv run pytest tests/test_llm_switcher.py切换过程中工具与设置的保持跨 provider 切换最容易踩的坑是 function callingLLMSwitcher对工具注册做了专门的广播源码中有两个事实LLMContext(tools[...])中列出的 direct functions 会在每个成员 LLM 上注册——无论激活与否。原因是成员 LLM 各自被分支过滤器挡住运行时只有激活的 LLM 能收到LLMContextFrame切换器在收到上下文帧时主动把所有成员的 tool handler 同步一遍llm_switcher.py 的process_frame保证切换后工具继续可用。features-service-switcher 示例就是靠这一机制让get_current_weather等工具在切换前后都工作llm_switcher.register_function(...)同样会转发到所有成员 LLMcancel_on_interruption、timeout_secs等参数按显式参数 tool_options装饰器 默认值在每个成员上解析LLMSwitcher.register_direct_function自 1.4.0 起被标记为 deprecated计划 2.0.0 移除官方建议改用LLMContext(tools[...])或推LLMSetToolsFrame在会话中途变更工具运行时更新 LLM 设置如LLMUpdateSettingsFrame带reach_inactive_servicesTrue时更新会同时送达未激活的成员避免某个 LLM 被激活后缺了配置tests/test_llm_switcher.py 中有对应测试。限制与边界目标 LLM 必须在传给LLMSwitcher的列表里且is_usable为 True否则切换被忽略目标不在列表时可能本意是给 pipeline 里另一个 switcher 的直接被忽略切换器自身不接受set_usable它报告的可用性是所有成员 LLM 的读数把某个成员恢复为 usable 即可带动整个切换器恢复非激活 LLM 的错误会被切换器就地吸收不会再向上传播只有激活 LLM 的错误会交给策略决定是否切换切换成功后该错误也不再向上传播两个示例的切换都是会话内行为LLM 实例在启动时就全部构造好API key 需提前配齐运行中新增成员 LLM 的路径文档未提供需要自行处理。完整代码以仓库内文件为准examples/features/features-service-switcher.py三环节手动切换的最小示例与 examples/flows/python/llm_switching.py对话工具驱动切换 Flows 集成Flows 示例的运行方式见 examples/flows/README.md。【免费下载链接】pipecatOpen Source framework for voice agents, multimodal apps, and realtime AI. Maintained by Daily and the community.项目地址: https://gitcode.com/GitHub_Trending/pi/pipecat创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考