第 6 篇模型接入让 Agent 使用不同的模型提供者6.1 模型是什么在 Norp 里模型是 Agent 的“大脑”——它接收对话历史和工具列表决定下一步说什么、调用哪个工具、什么时候结束。对于NorpAgent模型只是一个可选项并非核心成分。一个模型提供者就是一个实现了ModelProvider协议的对象from norpagent.protocols.model import ModelProvider class MyProvider: model_id my_model def generate(self, messages, tools, params): # 接收对话历史 工具列表 → 返回模型输出 return ModelOutput(content...) def stream(self, messages, tools, params): # 可选逐 chunk 返回输出 yield ModelStreamChunk(delta_content...)核心是generate()方法输入messages对话历史和tools工具 Schema输出ModelOutput文本 可能的工具调用。实现了stream()的模型框架会优先走流式路径逐 chunk 广播给前端没实现则走一次性generate()。6.2 ModelOutput 长什么样generate()返回一个ModelOutput对象字段类型说明contentstr模型生成的文本内容reasoningstr可选思维链内容tool_callsList[ToolCallSpec]可选模型想调用的工具列表usageModelUsage可选token 消耗统计finish_reasonstrstop / tool_calls / length 等模型最核心的决策是返回content→ 直接回答用户返回tool_calls→ 调用工具然后继续6.3 内置模型适配器Norp 默认注册了三个模型适配器mocknpa(modelmock)确定性输出内置问答对 引导语零依赖。用于测试、基准、无网络环境。没有 API Key 时自动回落。openai_compatnpa(modelopenai_compat, model_namedeepseek-v4-flash, base_urlhttps://api.deepseek.com/v1, api_keysk-xxx)OpenAI 兼容协议覆盖 DeepSeek / OpenAI / Qwen / vLLM / Ollama。依赖norpagent[openai]扩展包安装openaiSDK。自动识别 DeepSeek V4 并正确处理思维链格式。anthropicnpa(modelanthropic, model_nameclaude-3-5-sonnet-20241022, api_keysk-xxx)Anthropic 协议适配器。依赖norpagent[anthropic]扩展包。6.4 凭据的三种传递方式模型可能需要 API Key。有三种方式传递方式一启动参数npa(modelopenai_compat, model_namedeepseek-v4-flash, base_urlhttps://api.deepseek.com/v1, api_keysk-xxx)model_name、base_url、api_key是模型快捷参数只在模型是内置适配器名时生效。方式二环境变量export OPENAI_API_KEYsk-xxx npa(modelopenai_compat)框架按以下顺序检查环境变量OPENAI_API_KEY→DEEPSEEK_API_KEY→ANTHROPIC_API_KEY→DASHSCOPE_API_KEY→NORPAGENT_API_KEY。方式三构造好的实例from norpagent.builtin.models.openai_compat import OpenAICompatProvider provider OpenAICompatProvider( model_namedeepseek-v4-flash, base_urlhttps://api.deepseek.com/v1, api_keysk-xxx ) npa(modelprovider)完全控制构造过程适合复杂场景。凭据回落规则没检测到任何 Key 时模型自动回落mock嵌入式场景开箱可用。若模型不需要key可瞎填一个。6.5 实现一个自定义模型提供者对接任意 HTTP API 的模型适配器示例# myapp/models/http_json.py import json from urllib.request import Request, urlopen from norpagent.protocols.model import ( ModelOutput, ModelProvider, ModelUsage, ToolCallSpec ) class HttpJsonModel: model_id http-json def __init__(self, endpoint: str, api_key: str ): self._endpoint endpoint self._api_key api_key def _payload(self, messages, tools, params): body { messages: [m.to_openai() for m in messages], temperature: params.get(temperature, 0.7), } if tools: body[tools] tools return body def generate(self, messages, tools, params): body self._payload(messages, tools, params) req Request( self._endpoint, datajson.dumps(body).encode(utf-8), headers{Content-Type: application/json} ) if self._api_key: req.add_header(Authorization, fBearer {self._api_key}) with urlopen(req, timeout60) as resp: data json.load(resp) choice data[choices][0] msg choice[message] tool_calls None if msg.get(tool_calls): tool_calls [ ToolCallSpec( idtc[id], nametc[function][name], argumentsjson.loads(tc[function][arguments]) ) for tc in msg[tool_calls] ] usage data.get(usage, {}) return ModelOutput( contentmsg.get(content, ), tool_callstool_calls, usageModelUsage( input_tokensusage.get(prompt_tokens, 0), output_tokensusage.get(completion_tokens, 0), total_tokensusage.get(total_tokens, 0), ), finish_reasonchoice.get(finish_reason, stop), ) def create(configNone, **kw): return HttpJsonModel( endpoint(config or {}).get(endpoint, http://localhost:8000/v1), api_key(config or {}).get(api_key, ), )使用npa(modelmyapp.models.http_json:create;endpointhttp://127.0.0.1:8000/v1)6.6 流式输出实现stream()让模型输出逐词显示def stream(self, messages, tools, params): from norpagent.loops.cancel import cancel_requested body self._payload(messages, tools, params) body[stream] True with urlopen(Request(...), timeout120) as resp: for line in resp: if cancel_requested(): # CtrlC / 引擎停止 → 尽早退出 return line line.decode().strip() if not line.startswith(data:): continue chunk json.loads(line[5:]) delta chunk[choices][0].get(delta, {}) yield ModelStreamChunk( delta_contentdelta.get(content, ) )流式路径中必须检查取消信号params[_cancel_event]在call_timeout0时同样注入引擎停止 / CtrlC 时置位。实现stream后框架优先走流式路径逐 chunk 广播on_content未实现则走generate。6.7 注册和使用注册自定义模型from norpagent import Registry reg Registry() reg.register_model(http_json, HttpJsonModel(endpoint...))启动时引用npa(modelhttp_json)或直接按地址加载npa(modelmyapp.models.http_json:create;endpoint...)或传实例npa(modelHttpJsonModel(endpoint...))6.8 模型热重载模型是组件槽位热重载下一次run()生效npa.remount(modelmyapp.models.http_json:create;endpointhttp://10.0.0.1:8000/v1)修改模型代码后用地址形态重挂载模块缓存自动失效新代码被重新导入。6.9 取消支持模型适配器必须支持取消params[_cancel_event]是内核注入的取消事件流式路径每 chunk 检查cancel_requested()非流式路径请在内部循环或分片路径上检查不检查取消的长模型调用会占住工作池后续任务堆积。这是硬性要求。6.10 你现在能做的事情读完这篇你已经能理解ModelProvider协议generate 可选stream使用内置模型mock、openai_compat、anthropic通过三种方式传递 API Key参数、环境变量、实例实现自定义 HTTP 模型适配器实现流式输出并在流中检查取消信号注册自定义模型并通过名字/地址/实例使用用npa.remount(model...)热重载模型下一篇你会学到预设模式——如何用预设组合管理不同的配置方案。
元框架系列教程第六篇——接入模型
第 6 篇模型接入让 Agent 使用不同的模型提供者6.1 模型是什么在 Norp 里模型是 Agent 的“大脑”——它接收对话历史和工具列表决定下一步说什么、调用哪个工具、什么时候结束。对于NorpAgent模型只是一个可选项并非核心成分。一个模型提供者就是一个实现了ModelProvider协议的对象from norpagent.protocols.model import ModelProvider class MyProvider: model_id my_model def generate(self, messages, tools, params): # 接收对话历史 工具列表 → 返回模型输出 return ModelOutput(content...) def stream(self, messages, tools, params): # 可选逐 chunk 返回输出 yield ModelStreamChunk(delta_content...)核心是generate()方法输入messages对话历史和tools工具 Schema输出ModelOutput文本 可能的工具调用。实现了stream()的模型框架会优先走流式路径逐 chunk 广播给前端没实现则走一次性generate()。6.2 ModelOutput 长什么样generate()返回一个ModelOutput对象字段类型说明contentstr模型生成的文本内容reasoningstr可选思维链内容tool_callsList[ToolCallSpec]可选模型想调用的工具列表usageModelUsage可选token 消耗统计finish_reasonstrstop / tool_calls / length 等模型最核心的决策是返回content→ 直接回答用户返回tool_calls→ 调用工具然后继续6.3 内置模型适配器Norp 默认注册了三个模型适配器mocknpa(modelmock)确定性输出内置问答对 引导语零依赖。用于测试、基准、无网络环境。没有 API Key 时自动回落。openai_compatnpa(modelopenai_compat, model_namedeepseek-v4-flash, base_urlhttps://api.deepseek.com/v1, api_keysk-xxx)OpenAI 兼容协议覆盖 DeepSeek / OpenAI / Qwen / vLLM / Ollama。依赖norpagent[openai]扩展包安装openaiSDK。自动识别 DeepSeek V4 并正确处理思维链格式。anthropicnpa(modelanthropic, model_nameclaude-3-5-sonnet-20241022, api_keysk-xxx)Anthropic 协议适配器。依赖norpagent[anthropic]扩展包。6.4 凭据的三种传递方式模型可能需要 API Key。有三种方式传递方式一启动参数npa(modelopenai_compat, model_namedeepseek-v4-flash, base_urlhttps://api.deepseek.com/v1, api_keysk-xxx)model_name、base_url、api_key是模型快捷参数只在模型是内置适配器名时生效。方式二环境变量export OPENAI_API_KEYsk-xxx npa(modelopenai_compat)框架按以下顺序检查环境变量OPENAI_API_KEY→DEEPSEEK_API_KEY→ANTHROPIC_API_KEY→DASHSCOPE_API_KEY→NORPAGENT_API_KEY。方式三构造好的实例from norpagent.builtin.models.openai_compat import OpenAICompatProvider provider OpenAICompatProvider( model_namedeepseek-v4-flash, base_urlhttps://api.deepseek.com/v1, api_keysk-xxx ) npa(modelprovider)完全控制构造过程适合复杂场景。凭据回落规则没检测到任何 Key 时模型自动回落mock嵌入式场景开箱可用。若模型不需要key可瞎填一个。6.5 实现一个自定义模型提供者对接任意 HTTP API 的模型适配器示例# myapp/models/http_json.py import json from urllib.request import Request, urlopen from norpagent.protocols.model import ( ModelOutput, ModelProvider, ModelUsage, ToolCallSpec ) class HttpJsonModel: model_id http-json def __init__(self, endpoint: str, api_key: str ): self._endpoint endpoint self._api_key api_key def _payload(self, messages, tools, params): body { messages: [m.to_openai() for m in messages], temperature: params.get(temperature, 0.7), } if tools: body[tools] tools return body def generate(self, messages, tools, params): body self._payload(messages, tools, params) req Request( self._endpoint, datajson.dumps(body).encode(utf-8), headers{Content-Type: application/json} ) if self._api_key: req.add_header(Authorization, fBearer {self._api_key}) with urlopen(req, timeout60) as resp: data json.load(resp) choice data[choices][0] msg choice[message] tool_calls None if msg.get(tool_calls): tool_calls [ ToolCallSpec( idtc[id], nametc[function][name], argumentsjson.loads(tc[function][arguments]) ) for tc in msg[tool_calls] ] usage data.get(usage, {}) return ModelOutput( contentmsg.get(content, ), tool_callstool_calls, usageModelUsage( input_tokensusage.get(prompt_tokens, 0), output_tokensusage.get(completion_tokens, 0), total_tokensusage.get(total_tokens, 0), ), finish_reasonchoice.get(finish_reason, stop), ) def create(configNone, **kw): return HttpJsonModel( endpoint(config or {}).get(endpoint, http://localhost:8000/v1), api_key(config or {}).get(api_key, ), )使用npa(modelmyapp.models.http_json:create;endpointhttp://127.0.0.1:8000/v1)6.6 流式输出实现stream()让模型输出逐词显示def stream(self, messages, tools, params): from norpagent.loops.cancel import cancel_requested body self._payload(messages, tools, params) body[stream] True with urlopen(Request(...), timeout120) as resp: for line in resp: if cancel_requested(): # CtrlC / 引擎停止 → 尽早退出 return line line.decode().strip() if not line.startswith(data:): continue chunk json.loads(line[5:]) delta chunk[choices][0].get(delta, {}) yield ModelStreamChunk( delta_contentdelta.get(content, ) )流式路径中必须检查取消信号params[_cancel_event]在call_timeout0时同样注入引擎停止 / CtrlC 时置位。实现stream后框架优先走流式路径逐 chunk 广播on_content未实现则走generate。6.7 注册和使用注册自定义模型from norpagent import Registry reg Registry() reg.register_model(http_json, HttpJsonModel(endpoint...))启动时引用npa(modelhttp_json)或直接按地址加载npa(modelmyapp.models.http_json:create;endpoint...)或传实例npa(modelHttpJsonModel(endpoint...))6.8 模型热重载模型是组件槽位热重载下一次run()生效npa.remount(modelmyapp.models.http_json:create;endpointhttp://10.0.0.1:8000/v1)修改模型代码后用地址形态重挂载模块缓存自动失效新代码被重新导入。6.9 取消支持模型适配器必须支持取消params[_cancel_event]是内核注入的取消事件流式路径每 chunk 检查cancel_requested()非流式路径请在内部循环或分片路径上检查不检查取消的长模型调用会占住工作池后续任务堆积。这是硬性要求。6.10 你现在能做的事情读完这篇你已经能理解ModelProvider协议generate 可选stream使用内置模型mock、openai_compat、anthropic通过三种方式传递 API Key参数、环境变量、实例实现自定义 HTTP 模型适配器实现流式输出并在流中检查取消信号注册自定义模型并通过名字/地址/实例使用用npa.remount(model...)热重载模型下一篇你会学到预设模式——如何用预设组合管理不同的配置方案。