实战指南AgentScope 自定义模型集成一次讲清 ChatModelBase 契约与完整代码【免费下载链接】agentscopeBuild and run agents you can see, understand and trust.项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope把公司内部 AI 服务接进 AgentScope是让私有模型获得工具调用、上下文压缩等框架能力的常见需求。这篇实战指南按真实代码结构带你从零完成一次自定义模型集成读懂基类契约、写好模型文件、注册导出再用几行 smoke test 确认整条链路跑通。ChatModelBase 契约你必须实现的三样东西AgentScope 里所有对话模型类都继承自ChatModelBasesrc/agentscope/model/_base.py。它是模型层的统一契约上游的 Agent、Agent Service 只认这个抽象不关心背后接的是哪家服务。继承它需要遵守三件事初始化参数credential继承CredentialBase的凭据对象用来挂 API 密钥等敏感信息、model模型名字符串、parameters参数配置以及stream这类行为开关都要透传给父类构造。核心方法公开入口__call__基类已经写好含重试与流式包装你要实现的是它的下层抽象方法_call_api——真正发请求、解析响应的地方。返回值非流式返回一个ChatResponse流式则返回由ChatResponse组成的异步生成器一块块增量最后一块标记结束。契约示意关键签名不超过 8 行class MyChatModel(ChatModelBase): async def __call__(self, messages, toolsNone, tool_choiceNone, **kwargs): ... # 基类实现重试 流式包装 abstractmethod async def _call_api(self, model_name, messages, toolsNone, tool_choiceNone, **kwargs): 必须实现返回 ChatResponse | AsyncGenerator工具调用场景下基类还提供了_validate_tool_choice校验方法用来确认tool_choice指向的工具确实存在避免模型幻觉出不存在的工具名。从零写一个模型文件下面这套代码是放在你本地项目里新建的文件仓库本身保持只读不动。先建文件比如my_model/my_chat_model.py并定义凭据类。凭据类是敏感信息的唯一入口密钥不进模型代码这一步省掉后患from agentscope.credential import CredentialBase from agentscope.model import ChatModelBase from agentscope.model._model_response import ChatResponse from pydantic import Field class MyCredential(CredentialBase): 内部 AI 服务的访问凭据 token: str Field(default, descriptionAccess token) classmethod def get_chat_model_class(cls): from my_model.my_chat_model import MyChatModel return MyChatModel再实现模型类。__init__必须把参数原样透传给父类否则基类里stream、max_retries这些机制会拿不到配置import httpx from datetime import datetime from agentscope.message import Msg, TextBlock class MyChatModel(ChatModelBase): 对接内部 AI 服务的自定义模型 class Parameters(BaseModel): max_tokens: int | None None def __init__(self, credential, model, parametersNone, streamTrue, **kwargs): super().__init__( credentialcredential, modelmodel, parametersparameters or self.Parameters(), streamstream, **kwargs, ) self._http httpx.AsyncClient( base_urlcredential.token, # 简化示意 timeout60.0, )然后是_call_api。这里是自定义模型集成的核心把框架的Msg列表转成你的服务能懂的请求再把响应翻译回ChatResponse。流式分支不能省——Agent 的实时输出全靠它async def _call_api(self, model_name, messages, toolsNone, tool_choiceNone, **kwargs): if tool_choice and tools: self._validate_tool_choice(tool_choice, tools) payload { model: model_name, messages: [m.get_content_blocks() for m in messages], stream: self.stream, } if not self.stream: return await self._parse_completion(payload) # 流式逐块 yield最后一块 is_lastTrue async def _stream(): async with self._http.stream( POST, /v1/chat, jsonpayload, ) as resp: async for line in resp.aiter_lines(): chunk self._parse_chunk(line) yield chunk # is_lastFalse 的增量块 return _stream() async def _parse_completion(self, payload) - ChatResponse: res await self._http.post(/v1/chat, jsonpayload) data res.json() return ChatResponse( content[TextBlock(textdata[message][content])], is_lastTrue, ) def _parse_chunk(self, line) - ChatResponse: data json.loads(line.removeprefix(data: )) last data.get(finish, False) return ChatResponse( content[TextBlock(textdata[delta])] if not last else [], is_lastlast, )两个容易忽略的细节增量块统一is_lastFalse只有终止块置True基类靠它触发完整响应的收尾另外基类__call__会自动重试你通过_get_retryable_exceptions声明的异常类型网络层瞬时错误不用自己处理。模型注册导出 5 行 smoke test类写好后最后一步是导出。这一步的意义在于agentscope.model包名空间的__all__决定了from agentscope.model import MyChatModel这类写法是否可用——漏了它下游项目会直接 ImportError。在你的本地包__init__.py中加两行即可官方各模型在 src/agentscope/model/init.py 中的注册方式就是参照样板from .my_chat_model import MyChatModel __all__ [..., MyChatModel]导出完成后别急着上功能先跑最小闭环。下面这个 smoke test 不超过 5 行核心断言验证的正是集成最脆弱的那一环——消息进去、ChatResponse出来import asyncio async def main(): m MyChatModel(MyCredential(tokenhttp://127.0.0.1:8080), modelinternal-llm, streamFalse) res await m([Msg(user, [TextBlock(textping)], user)]) assert res.is_last and res.content asyncio.run(main())返回了带内容的is_lastTrue响应说明导入、实例化、请求解析整条链路都通了。后续再开流式、加工具出问题时定位范围会小很多。踩坑速查表问题解法 参考路径私有服务的消息结构与框架的Msg对不上格式不兼容用 formatter 模块src/agentscope/formatter/做适配层把Msg列表转成目标格式而不是在_call_api里硬拼 dict流式接入时下游拿不到完整响应、增量块拼接错乱核对每个 chunk 的is_last语义参考 src/agentscope/model/_ollama/_model.py 里_parse_stream_response的写法增量块is_lastFalse终止块才置True高并发下调用慢、偶发连接超时性能瓶颈用长生命周期httpx.AsyncClient复用连接池别每次请求新建客户端异常类型交给基类重试机制兜底上生产前补三件事密钥走凭据对象API 密钥只放进CredentialBase子类并从环境变量读取示例代码里写死 token 仅限本地调试。重试靠基类覆盖_get_retryable_exceptions声明哪些异常值得重试基类会按max_retries/retry_delay自动处理不用手写 sleep 循环。补全用量数据在ChatResponse.usage里填 token 数与耗时配合追踪中间件src/agentscope/middleware/_tracing/记录调用指标线上排查才有据可依。建议的下一步很具体先在本地项目建一个最简模型只跑非流式文本用上面的 smoke test 验证闭环跑稳之后再打开stream分支最后才叠加工具调用与结构化输出。一次加一块每块都先验证——这是流式模型接入里最省时间的做法。【免费下载链接】agentscopeBuild and run agents you can see, understand and trust.项目地址: https://gitcode.com/GitHub_Trending/ag/agentscope创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考