高级场景与工程模板
本页从使用目标出发,给出插件级解决方案的组合方式。
场景 1:自定义模型协议(兼容内部网关)
目标:请求需要特殊签名、响应返回非标准结构。
方案:
- 自定义
ModelRequester负责签名与流式协议 - 自定义
ResponseParser负责 JSON 清洗
python
class MyRequester(ModelRequester):
name = "MyRequester"
DEFAULT_SETTINGS = {"my_gateway": {"timeout": 30}}
def generate_request_data(self):
# 把 prompt 转成网关请求
return {"payload": self.prompt.to_messages(), "sig": "..."}
async def request_model(self, request_data):
# 返回原始 SSE/流
async for item in my_gateway_stream(request_data):
yield item
async def broadcast_response(self, response_generator):
async for event, data in response_generator:
# 标准化输出
yield "delta", data.get("token")
yield "done", data.get("final")场景 2:结构化输出极不稳定
目标:模型输出不稳定,JSON 需要强校验 + 容错。
方案:
- PromptGenerator 输出更强的 schema 约束
- ResponseParser 统一清洗 + 补全 + 兜底
建议做法:
- 在 PromptGenerator 中对
output增加强规则 - 在 ResponseParser 中使用“单一清洗入口”
- 对失败情况提供 fallback(文本解析 / 旧值合并)
场景 3:输出要同时支持流式 UI + 最终结构化结果
目标:UI 要实时显示,后端要拿结构化数据。
方案:
- ModelRequester 持续发出
delta与done - ResponseParser 缓冲 delta,并在 done 时解析结构化结果
这要求你在 ResponseParser 中维护 text_result + parsed_result 两条路径。
场景 4:多模型路由(高成本 / 低成本)
目标:小请求走轻量模型,大请求走高性能模型。
方案:
- ModelRequester 内部路由
- 根据 prompt 长度或 settings 选择路由
python
if self.prompt.get("input") and len(self.prompt.get("input")) < 200:
use_small_model()
else:
use_large_model()场景 5:工具系统要分级授权
目标:某些工具只对部分用户开放。
方案:
- ToolManager 注册时带上 tags
- 调用前先过滤 / 校验
python
tool_manager.tag(["danger_tool"], "admin_only")场景 6:自定义记忆系统(替换默认 memo)
目标:使用外部服务维护记忆、并定期回灌。
方案:
- 自定义
memo_update_handler - 或直接实现新的 Session 插件
python
def memo_update_handler(memo, messages, attachments, settings):
return external_memory_service.update(memo, messages)场景 7:服务端会话管理(本地只做日志)
目标:上下文由服务器维护,本地仅记录。
方案:
- 不 attach Session 到 Agent
- 单独维护 Session 作为日志
场景 8:安全审计与合规记录
目标:记录所有模型输入输出并可追溯。
方案:
- 通过 SessionExtension 的 record 机制捕获输入输出
- 导出 Session JSON 归档
工程落地建议
- 插件最小化:每次只替换一个插件类型
- 可观测性:优先开启
runtime.show_model_logs - 回滚路径:插件失败时应能切回默认插件