夜雨聆风使用场景：
系统中有多种算法；同一种算法可能有多个版本；业务调用时可以指定版本，也可以走默认版本、最新版本、灰度版本；多个算法之间可以串联形成完整业务流程。
基于 framex-kit 官方 PyPI / GitHub 公开资料整理。framex-kit 当前 PyPI 最新版本显示为 0.3.9，发布时间为 2026-05-18，要求 Python >= 3.11，支持 ray extra。(PyPI) 官方说明它的核心概念包括 Plugin、@on_register()、@on_request(...)、required_remote_apis、call_plugin_api(...)、@remote() 和内置 proxy 插件。(PyPI)
一、先理解整体设计思想
用 framex-kit 做多算法系统时，推荐采用这四层：
业务入口层 └── workflow 插件：负责完整业务流程编排算法路由层 └── algorithm_router 插件：负责选择哪个算法、哪个版本算法能力层 ├── sentiment_v1 插件 ├── sentiment_v2 插件 ├── keyword_v1 插件 ├── risk_rule_v1 插件 └── 其他算法插件基础能力层 ├── preprocess 插件 ├── logging / tracing 插件 └── 其他公共能力插件最重要的原则是：
具体算法插件：只负责一个稳定算法版本算法路由插件：负责版本选择、默认版本、灰度策略业务编排插件：负责把多个算法串起来FrameX：负责插件加载、API 暴露、插件间调用FrameX 官方定位就是把 Python 服务拆成独立插件，同时暴露统一的 HTTP 和内部 API 接口，适合模块边界清晰、多团队并行开发、插件加载、proxy 集成以及从本地执行扩展到 Ray 执行的场景。(GitHub)
二、目标案例
我们做一个“文本智能分析平台”，包含：
1. preprocess 文本预处理插件2. sentiment_v1 情感分类算法 v13. sentiment_v2 情感分类算法 v24. keyword_v1 关键词抽取算法 v15. risk_rule_v1 风险规则判断算法 v16. algorithm_router 算法路由插件，负责选择具体算法版本7. text_workflow 业务编排插件，串联： 文本预处理 → 情感分析 → 关键词抽取 → 风险规则判断外部调用时可以：
直接调用某个算法版本：/api/v1/algorithms/sentiment/v1/predict/api/v1/algorithms/sentiment/v2/predict通过路由插件调用：/api/v1/router/predict通过完整业务流调用：/api/v1/workflow/text/analyze三、创建项目
1. 创建目录
mkdir framex_algorithm_platformcd framex_algorithm_platformpython3.11 -m venv .venvsource .venv/bin/activatepip install framex-kit如果后面你要用 Ray：
pip install "framex-kit[ray]"官方安装说明也是基础包使用 pip install framex-kit，Ray Serve 支持使用 pip install "framex-kit[ray]"。(PyPI)
2. 推荐目录结构
framex_algorithm_platform/├── config.toml├── README.md├── app/│ ├── __init__.py│ ├── schemas.py│ └── utils.py└── plugins/ ├── __init__.py ├── preprocess.py ├── sentiment_v1.py ├── sentiment_v2.py ├── keyword_v1.py ├── risk_rule_v1.py ├── algorithm_router.py └── text_workflow.py创建文件：
mkdir -p app pluginstouch app/__init__.py plugins/__init__.pytouch app/schemas.py app/utils.pytouch plugins/preprocess.pytouch plugins/sentiment_v1.pytouch plugins/sentiment_v2.pytouch plugins/keyword_v1.pytouch plugins/risk_rule_v1.pytouch plugins/algorithm_router.pytouch plugins/text_workflow.pytouch config.toml四、定义统一请求和响应模型
先写 app/schemas.py。
统一模型的意义是：不同算法版本最好保持输入输出兼容。这样 sentiment_v1 升级到 sentiment_v2 时，业务编排插件不用大改。
# app/schemas.pyfrom typing importAny, Literal, Optionalfrom pydantic import BaseModel, FieldclassTextRequest(BaseModel): text: str = Field(..., description="输入文本")classPreprocessResponse(BaseModel): original_text: str clean_text: strclassAlgorithmRequest(BaseModel): text: str algorithm: Optional[str] = None version: Optional[str] = "latest" metadata: dict[str, Any] = Field(default_factory=dict)classAlgorithmResponse(BaseModel): algorithm: str version: str result: dict[str, Any] score: Optional[float] = None metadata: dict[str, Any] = Field(default_factory=dict)classSentimentResult(BaseModel): label: Literal["positive", "neutral", "negative"] score: floatclassKeywordResult(BaseModel): keywords: list[str]classRiskRuleRequest(BaseModel): text: str sentiment: dict[str, Any] keywords: list[str]classRiskRuleResponse(BaseModel): risk_level: Literal["low", "medium", "high"] reasons: list[str]classWorkflowRequest(BaseModel): text: str sentiment_version: Optional[str] = "latest"classWorkflowResponse(BaseModel): clean_text: str sentiment: dict[str, Any] keywords: list[str] risk: dict[str, Any] final_decision: str五、写一个通用工具函数
FrameX 的 HTTP 响应默认可能会包一层统一 JSON envelope；官方说明非 streaming API 默认响应形态类似：
{"status":200,"message":"success","timestamp":"...","data":{}}除非你使用 raw_response=True。(PyPI)
插件内部调用时通常拿到的是目标插件返回值，但为了兼容 Pydantic 对象、dict、HTTP 包装结果，我们写个转换工具。
# app/utils.pyfrom typing importAnydefto_dict(obj: Any) -> dict:""" 将 Pydantic 对象、普通 dict 或 FrameX 包装响应统一转成 dict。 """if obj isNone:return {}ifhasattr(obj, "model_dump"):return obj.model_dump()ifisinstance(obj, dict):# 兼容 HTTP wrapper: {"status": 200, "data": {...}}if"data"in obj and"status"in obj: data = obj.get("data")return data ifisinstance(data, dict) else {"data": data}return objreturn {"value": obj}六、写第一个插件：文本预处理插件
plugins/preprocess.py
# plugins/preprocess.pyfrom typing importAnyfrom framex.consts import VERSIONfrom framex.plugin import BasePlugin, PluginMetadata, on_register, on_requestfrom app.schemas import TextRequest, PreprocessResponse__plugin_meta__ = PluginMetadata( name="preprocess", version=VERSION, description="Text preprocessing plugin", author="your-team", url="https://example.com/preprocess",)@on_register()classPreprocessPlugin(BasePlugin):def__init__(self, **kwargs: Any) -> None:super().__init__(**kwargs) @on_request("/preprocess/normalize", methods=["POST"])asyncdefnormalize(self, request: TextRequest) -> PreprocessResponse: text = request.text clean_text = ( text.strip() .replace("\n", " ") .replace("\t", " ") )while" "in clean_text: clean_text = clean_text.replace(" ", " ")return PreprocessResponse( original_text=text, clean_text=clean_text, )这里的关键点：
@on_register()classPreprocessPlugin(BasePlugin): ...表示注册一个插件类。
@on_request("/preprocess/normalize", methods=["POST"])表示把这个方法暴露成 HTTP API。FrameX 官方 Quick Start 也是通过 @on_register() 注册插件类，并通过 @on_request(...) 暴露 GET / POST 接口。(PyPI)
FrameX 会自动给路径加上 /api/v1 前缀，所以最终接口是：
POST /api/v1/preprocess/normalize七、写算法插件：sentiment_v1
plugins/sentiment_v1.py
# plugins/sentiment_v1.pyfrom typing importAnyfrom framex.plugin import BasePlugin, PluginMetadata, on_register, on_requestfrom app.schemas import TextRequest, AlgorithmResponse__plugin_meta__ = PluginMetadata( name="sentiment_v1", version="1.0.0", description="Sentiment classifier v1", author="your-team", url="https://example.com/sentiment/v1",)@on_register()classSentimentV1Plugin(BasePlugin):def__init__(self, **kwargs: Any) -> None:super().__init__(**kwargs)# 实际项目中可以在这里加载模型# self.model = load_model("models/sentiment/v1")self.positive_words = {"好", "优秀", "满意", "喜欢", "推荐", "不错"}self.negative_words = {"差", "糟糕", "投诉", "失望", "不好", "垃圾"} @on_request("/algorithms/sentiment/v1/predict", methods=["POST"])asyncdefpredict(self, request: TextRequest) -> AlgorithmResponse: text = request.text pos_count = sum(1for word inself.positive_words if word in text) neg_count = sum(1for word inself.negative_words if word in text)if pos_count > neg_count: label = "positive" score = 0.75elif neg_count > pos_count: label = "negative" score = 0.75else: label = "neutral" score = 0.60return AlgorithmResponse( algorithm="sentiment", version="v1", result={"label": label,"score": score, }, score=score, metadata={"method": "rule_based_v1","positive_word_count": pos_count,"negative_word_count": neg_count, }, )八、写算法插件：sentiment_v2
sentiment_v2 可以模拟一个更强的算法版本。实际生产中，这里可以换成深度学习模型、LLM 分类器、BERT 分类器等。
plugins/sentiment_v2.py
# plugins/sentiment_v2.pyfrom typing importAnyfrom framex.plugin import BasePlugin, PluginMetadata, on_register, on_requestfrom app.schemas import TextRequest, AlgorithmResponse__plugin_meta__ = PluginMetadata( name="sentiment_v2", version="2.0.0", description="Sentiment classifier v2", author="your-team", url="https://example.com/sentiment/v2",)@on_register()classSentimentV2Plugin(BasePlugin):def__init__(self, **kwargs: Any) -> None:super().__init__(**kwargs)# 实际项目中可以在这里加载新模型# self.model = load_model("models/sentiment/v2")self.positive_words = {"好", "优秀", "满意", "喜欢", "推荐", "不错", "高效", "专业"}self.negative_words = {"差", "糟糕", "投诉", "失望", "不好", "垃圾", "超时", "敷衍"} @on_request("/algorithms/sentiment/v2/predict", methods=["POST"])asyncdefpredict(self, request: TextRequest) -> AlgorithmResponse: text = request.text pos_count = sum(1for word inself.positive_words if word in text) neg_count = sum(1for word inself.negative_words if word in text)if pos_count > neg_count: label = "positive" score = min(0.95, 0.70 + pos_count * 0.08)elif neg_count > pos_count: label = "negative" score = min(0.95, 0.70 + neg_count * 0.08)else: label = "neutral" score = 0.65return AlgorithmResponse( algorithm="sentiment", version="v2", result={"label": label,"score": score, }, score=score, metadata={"method": "enhanced_rule_based_v2","positive_word_count": pos_count,"negative_word_count": neg_count, }, )到这里，同一个算法已经有两个版本：
/api/v1/algorithms/sentiment/v1/predict/api/v1/algorithms/sentiment/v2/predict这就是最推荐的版本管理方式：
一个重要算法版本 = 一个独立插件优点是：
1. v1 和 v2 可以同时在线2. 出问题时可以快速回滚3. 便于灰度发布4. 便于团队独立维护5. 不同版本之间不会互相污染九、写第二类算法插件：关键词抽取
plugins/keyword_v1.py
# plugins/keyword_v1.pyfrom typing importAnyfrom framex.plugin import BasePlugin, PluginMetadata, on_register, on_requestfrom app.schemas import TextRequest, AlgorithmResponse__plugin_meta__ = PluginMetadata( name="keyword_v1", version="1.0.0", description="Keyword extraction algorithm v1", author="your-team", url="https://example.com/keyword/v1",)@on_register()classKeywordV1Plugin(BasePlugin):def__init__(self, **kwargs: Any) -> None:super().__init__(**kwargs)self.domain_words = {"投诉","退款","物流","客服","订单","发票","超时","服务","质量","价格", } @on_request("/algorithms/keyword/v1/extract", methods=["POST"])asyncdefextract(self, request: TextRequest) -> AlgorithmResponse: text = request.text keywords = [word for word inself.domain_words if word in text]return AlgorithmResponse( algorithm="keyword", version="v1", result={"keywords": keywords, }, metadata={"method": "domain_dictionary_match","keyword_count": len(keywords), }, )十、写风险规则插件
这个插件不一定是机器学习算法，也可以是规则算法。
plugins/risk_rule_v1.py
# plugins/risk_rule_v1.pyfrom typing importAnyfrom framex.plugin import BasePlugin, PluginMetadata, on_register, on_requestfrom app.schemas import RiskRuleRequest, RiskRuleResponse__plugin_meta__ = PluginMetadata( name="risk_rule_v1", version="1.0.0", description="Risk rule engine v1", author="your-team", url="https://example.com/risk-rule/v1",)@on_register()classRiskRuleV1Plugin(BasePlugin):def__init__(self, **kwargs: Any) -> None:super().__init__(**kwargs) @on_request("/algorithms/risk-rule/v1/evaluate", methods=["POST"])asyncdefevaluate(self, request: RiskRuleRequest) -> RiskRuleResponse: reasons: list[str] = [] sentiment_label = request.sentiment.get("label") sentiment_score = float(request.sentiment.get("score", 0.0))if sentiment_label == "negative"and sentiment_score >= 0.8: reasons.append("高置信度负面情绪") risk_keywords = {"投诉", "退款", "超时"} hit_risk_keywords = [word for word in request.keywords if word in risk_keywords]if hit_risk_keywords: reasons.append(f"命中风险关键词：{','.join(hit_risk_keywords)}")iflen(reasons) >= 2: risk_level = "high"eliflen(reasons) == 1: risk_level = "medium"else: risk_level = "low"return RiskRuleResponse( risk_level=risk_level, reasons=reasons, )十一、写算法路由插件
这是多版本协同系统的核心。
它负责：
1. 接收 algorithm + version2. 判断应该调用哪个算法版本3. 调用具体算法插件4. 返回统一结果FrameX 官方说明，一个插件可以声明 required_remote_apis，然后通过 self._call_remote_api(...) 调用其他插件，而不是直接 import 对方实现。(GitHub) 这种方式非常适合算法协同，因为可以把算法能力解耦成稳定服务接口。
plugins/algorithm_router.py
# plugins/algorithm_router.pyimport randomfrom typing importAny, Literal, Optionalfrom pydantic import BaseModelfrom framex.plugin import BasePlugin, PluginMetadata, on_register, on_requestfrom app.schemas import TextRequestfrom app.utils import to_dict__plugin_meta__ = PluginMetadata( name="algorithm_router", version="1.0.0", description="Algorithm version router", author="your-team", url="https://example.com/algorithm-router", required_remote_apis=["/api/v1/algorithms/sentiment/v1/predict","/api/v1/algorithms/sentiment/v2/predict","/api/v1/algorithms/keyword/v1/extract", ],)classRouterRequest(BaseModel): algorithm: Literal["sentiment", "keyword"] text: str version: Optional[str] = "latest" strategy: Optional[Literal["fixed", "latest", "gray"]] = "latest"@on_register()classAlgorithmRouterPlugin(BasePlugin):def__init__(self, **kwargs: Any) -> None:super().__init__(**kwargs)# 实际生产中建议从 config.toml 或配置中心读取self.default_versions = {"sentiment": "v2","keyword": "v1", }# 灰度比例：10% 流量走 v2，90% 走 v1self.gray_policy = {"sentiment": {"v1": 0.90,"v2": 0.10, } } @on_request("/router/predict", methods=["POST"])asyncdefpredict(self, request: RouterRequest) -> dict: target_api = self._select_api( algorithm=request.algorithm, version=request.version, strategy=request.strategy, ) result = awaitself._call_remote_api( target_api, request={"text": request.text, }, ) result_dict = to_dict(result)return {"router": {"algorithm": request.algorithm,"requested_version": request.version,"strategy": request.strategy,"target_api": target_api, },"algorithm_result": result_dict, }def_select_api(self, algorithm: str, version: str | None, strategy: str | None) -> str:if algorithm == "sentiment": selected_version = self._select_sentiment_version(version, strategy)returnf"/api/v1/algorithms/sentiment/{selected_version}/predict"if algorithm == "keyword":return"/api/v1/algorithms/keyword/v1/extract"raise ValueError(f"Unsupported algorithm: {algorithm}")def_select_sentiment_version(self, version: str | None, strategy: str | None) -> str:if strategy == "fixed":if version notin {"v1", "v2"}:raise ValueError(f"Unsupported sentiment version: {version}")return versionif strategy == "gray":returnself._gray_select("sentiment")# latest 或默认情况if version in {None, "latest"}:returnself.default_versions["sentiment"]if version in {"v1", "v2"}:return versionraise ValueError(f"Unsupported sentiment version: {version}")def_gray_select(self, algorithm: str) -> str: policy = self.gray_policy.get(algorithm)ifnot policy:returnself.default_versions[algorithm] r = random.random() cumulative = 0.0for version, ratio in policy.items(): cumulative += ratioif r <= cumulative:return versionreturnself.default_versions[algorithm]这里注意 _call_remote_api 的参数：
awaitself._call_remote_api( target_api, request={"text": request.text},)因为目标插件方法签名是：
asyncdefpredict(self, request: TextRequest)所以这里传的 keyword 名叫 request。官方测试代码里也有类似模式：调用 /api/v1/echo_model 时传入 model={...}，FrameX 会按目标参数名把 dict 转成对应的 Pydantic 模型。(GitHub)
十二、写业务编排插件
这个插件负责“算法协同”。
流程如下：
输入原始文本 ↓调用 preprocess 插件 ↓调用 algorithm_router，选择 sentiment v1 / v2 ↓调用 keyword_v1 ↓调用 risk_rule_v1 ↓生成最终业务决策plugins/text_workflow.py
# plugins/text_workflow.pyfrom typing importAnyfrom framex.plugin import BasePlugin, PluginMetadata, on_register, on_requestfrom app.schemas import WorkflowRequest, WorkflowResponsefrom app.utils import to_dict__plugin_meta__ = PluginMetadata( name="text_workflow", version="1.0.0", description="Text analysis workflow plugin", author="your-team", url="https://example.com/text-workflow", required_remote_apis=["/api/v1/preprocess/normalize","/api/v1/router/predict","/api/v1/algorithms/keyword/v1/extract","/api/v1/algorithms/risk-rule/v1/evaluate", ],)@on_register()classTextWorkflowPlugin(BasePlugin):def__init__(self, **kwargs: Any) -> None:super().__init__(**kwargs) @on_request("/workflow/text/analyze", methods=["POST"])asyncdefanalyze(self, request: WorkflowRequest) -> WorkflowResponse:# 1. 文本预处理 preprocess_result = awaitself._call_remote_api("/api/v1/preprocess/normalize", request={"text": request.text, }, ) preprocess_dict = to_dict(preprocess_result) clean_text = preprocess_dict["clean_text"]# 2. 情感分析：通过 router 选择版本 sentiment_router_result = awaitself._call_remote_api("/api/v1/router/predict", request={"algorithm": "sentiment","text": clean_text,"version": request.sentiment_version,"strategy": "latest"if request.sentiment_version == "latest"else"fixed", }, ) sentiment_router_dict = to_dict(sentiment_router_result) sentiment_algorithm_result = sentiment_router_dict["algorithm_result"] sentiment = sentiment_algorithm_result["result"]# 3. 关键词抽取 keyword_result = awaitself._call_remote_api("/api/v1/algorithms/keyword/v1/extract", request={"text": clean_text, }, ) keyword_dict = to_dict(keyword_result) keywords = keyword_dict["result"]["keywords"]# 4. 风险规则判断 risk_result = awaitself._call_remote_api("/api/v1/algorithms/risk-rule/v1/evaluate", request={"text": clean_text,"sentiment": sentiment,"keywords": keywords, }, ) risk = to_dict(risk_result)# 5. 最终业务决策 final_decision = self._make_decision(risk_level=risk["risk_level"])return WorkflowResponse( clean_text=clean_text, sentiment=sentiment, keywords=keywords, risk=risk, final_decision=final_decision, )def_make_decision(self, risk_level: str) -> str:if risk_level == "high":return"manual_review"if risk_level == "medium":return"need_attention"return"auto_pass"十三、写配置文件
config.toml
load_plugins = [ "plugins.preprocess", "plugins.sentiment_v1", "plugins.sentiment_v2", "plugins.keyword_v1", "plugins.risk_rule_v1", "plugins.algorithm_router", "plugins.text_workflow"][server]host = "127.0.0.1"port = 8080use_ray = falseenable_proxy = false[plugins.preprocess]debug = true[plugins.sentiment_v1]model_path = "models/sentiment/v1"[plugins.sentiment_v2]model_path = "models/sentiment/v2"[plugins.algorithm_router]sentiment_default_version = "v2"FrameX 官方说明配置可以从环境变量、.env、.env.prod、config.toml、pyproject.toml 的 [tool.framex] 加载，并且支持 server.host、server.port、server.use_ray、server.enable_proxy、load_plugins、load_builtin_plugins、plugins.<plugin_name> 等配置项。(PyPI)
十四、启动服务
方式一：使用 config.toml
PYTHONPATH=. framex run方式二：命令行显式加载插件
PYTHONPATH=. framex run \ --load-plugins plugins.preprocess \ --load-plugins plugins.sentiment_v1 \ --load-plugins plugins.sentiment_v2 \ --load-plugins plugins.keyword_v1 \ --load-plugins plugins.risk_rule_v1 \ --load-plugins plugins.algorithm_router \ --load-plugins plugins.text_workflow官方说明 --load-plugins 和 --load-builtin-plugins 是可重复参数，不是逗号分隔列表。(PyPI)
启动成功后，通常可以访问：
http://127.0.0.1:8080/docshttp://127.0.0.1:8080/redochttp://127.0.0.1:8080/api/v1/openapi.json官方 Quick Start 中也说明 FrameX 会生成 /docs、/redoc 和 /api/v1/openapi.json。(PyPI)
十五、测试接口
1. 测试文本预处理
curl -X POST "http://127.0.0.1:8080/api/v1/preprocess/normalize" \ -H "Content-Type: application/json" \ -d '{ "text": " 客服态度不好，物流超时，我要投诉 " }'预期返回类似：
{"status":200,"message":"success","timestamp":"...","data":{"original_text":" 客服态度不好，物流超时，我要投诉 ","clean_text":"客服态度不好，物流超时，我要投诉"}}2. 直接调用情感分析 v1
curl -X POST "http://127.0.0.1:8080/api/v1/algorithms/sentiment/v1/predict" \ -H "Content-Type: application/json" \ -d '{ "text": "客服态度不好，物流超时，我要投诉" }'3. 直接调用情感分析 v2
curl -X POST "http://127.0.0.1:8080/api/v1/algorithms/sentiment/v2/predict" \ -H "Content-Type: application/json" \ -d '{ "text": "客服态度不好，物流超时，我要投诉" }'4. 通过算法路由调用 latest
curl -X POST "http://127.0.0.1:8080/api/v1/router/predict" \ -H "Content-Type: application/json" \ -d '{ "algorithm": "sentiment", "text": "客服态度不好，物流超时，我要投诉", "version": "latest", "strategy": "latest" }'此时会默认走 sentiment_v2。
5. 通过算法路由固定调用 v1
curl -X POST "http://127.0.0.1:8080/api/v1/router/predict" \ -H "Content-Type: application/json" \ -d '{ "algorithm": "sentiment", "text": "客服态度不好，物流超时，我要投诉", "version": "v1", "strategy": "fixed" }'6. 通过算法路由做灰度
curl -X POST "http://127.0.0.1:8080/api/v1/router/predict" \ -H "Content-Type: application/json" \ -d '{ "algorithm": "sentiment", "text": "客服态度不好，物流超时，我要投诉", "version": "latest", "strategy": "gray" }'根据我们写的灰度策略：
self.gray_policy = {"sentiment": {"v1": 0.90,"v2": 0.10, }}大约 90% 请求会走 v1，10% 请求会走 v2。
7. 调用完整业务流程
curl -X POST "http://127.0.0.1:8080/api/v1/workflow/text/analyze" \ -H "Content-Type: application/json" \ -d '{ "text": "客服态度不好，物流超时，我要投诉", "sentiment_version": "latest" }'预期返回类似：
{"status":200,"message":"success","timestamp":"...","data":{"clean_text":"客服态度不好，物流超时，我要投诉","sentiment":{"label":"negative","score":0.94},"keywords":["客服","物流","超时","投诉"],"risk":{"risk_level":"high","reasons":["高置信度负面情绪","命中风险关键词：投诉,超时"]},"final_decision":"manual_review"}}十六、多算法、多版本的推荐规范
1. 插件命名规范
推荐：
sentiment_v1sentiment_v2keyword_v1ocr_v1asr_paraformer_v1asr_whisper_v1rag_retriever_v1rag_retriever_v2llm_answer_v1safety_check_v1不推荐：
algorithm1new_algorithmtest_pluginmy_model原因是生产环境中很容易混乱。
2. API 路径规范
推荐：
/api/v1/algorithms/{algorithm_name}/{version}/{action}例如：
/api/v1/algorithms/sentiment/v1/predict/api/v1/algorithms/sentiment/v2/predict/api/v1/algorithms/keyword/v1/extract/api/v1/algorithms/ocr/v1/recognize/api/v1/algorithms/rag/v1/retrieve路由接口：
/api/v1/router/predict/api/v1/router/execute业务流程接口：
/api/v1/workflow/text/analyze/api/v1/workflow/callcenter/run/api/v1/workflow/document/parse3. 返回结构规范
建议所有算法都返回类似结构：
{"algorithm":"sentiment","version":"v2","result":{},"score":0.91,"metadata":{}}这样业务编排插件可以统一处理不同算法结果。
4. 版本选择策略
你可以支持四种版本策略。
固定版本
{"algorithm":"sentiment","version":"v1","strategy":"fixed"}用于复现实验、指定旧模型、问题排查。
最新版本
{"algorithm":"sentiment","version":"latest","strategy":"latest"}用于普通生产调用。
灰度版本
{"algorithm":"sentiment","strategy":"gray"}用于新模型逐步上线。
A/B 测试版本
可以扩展：
{"algorithm":"sentiment","strategy":"ab_test","metadata":{"user_id":"10001"}}让固定用户稳定命中固定版本。
十七、如何扩展到真实 AI 算法
上面的示例是规则算法。真实项目中，只需要把插件里的规则逻辑替换成模型推理即可。
例如：
classSentimentV2Plugin(BasePlugin):def__init__(self, **kwargs: Any) -> None:super().__init__(**kwargs)self.model = load_model("models/sentiment/v2")self.tokenizer = load_tokenizer("models/sentiment/v2") @on_request("/algorithms/sentiment/v2/predict", methods=["POST"])asyncdefpredict(self, request: TextRequest) -> AlgorithmResponse: inputs = self.tokenizer(request.text) outputs = self.model(inputs) label = outputs.label score = outputs.scorereturn AlgorithmResponse( algorithm="sentiment", version="v2", result={"label": label,"score": score, }, score=score, )注意：
模型应该在 __init__ 中加载不要在每次请求中加载模型否则性能会非常差。
十八、如何用于你的智能客服 / AI 话务机项目
结合你的智能语音客服终端，可以设计成：
callcenter_workflow ↓asr_router ├── asr_paraformer_v1 └── asr_whisper_v1 ↓text_preprocess ↓intent_router ├── intent_cls_v1 └── intent_cls_v2 ↓rag_router ├── rag_retriever_v1 └── rag_retriever_v2 ↓llm_answer_v1 ↓safety_check_v1 ↓tts_router ├── tts_v1 └── tts_v2对应接口：
/api/v1/workflow/callcenter/run/api/v1/algorithms/asr/paraformer/v1/transcribe/api/v1/algorithms/asr/whisper/v1/transcribe/api/v1/algorithms/intent/v1/predict/api/v1/algorithms/intent/v2/predict/api/v1/algorithms/rag/v1/retrieve/api/v1/algorithms/llm/v1/generate/api/v1/algorithms/safety/v1/check/api/v1/algorithms/tts/v1/synthesize一个完整流程可以是：
语音输入 → ASR 插件转文本 → 文本清洗插件 → 意图识别插件 → RAG 检索插件 → 大模型生成插件 → 安全审核插件 → TTS 插件 → 返回语音或文本结果这样做的好处是：
1. ASR 可以从 paraformer 切到 whisper2. 意图识别可以 v1/v2 灰度3. RAG 检索器可以独立升级4. LLM 可以独立更换5. 安全审核可以作为独立插件复用6. 业务流程不需要直接依赖每个算法的内部实现十九、如何引入已有 FastAPI 算法服务
有些算法可能已经是独立 FastAPI 服务，不想重写成 FrameX 插件。
这时可以用 FrameX 的内置 proxy 插件。官方说明 proxy 插件可以把已有 HTTP 服务暴露成 FrameX 统一 API 表面，例如上游服务在 127.0.0.1:9000 提供 /api/v1/chat，FrameX 可以在 127.0.0.1:8080/api/v1/chat 转发请求。(GitHub)
示例配置：
load_builtin_plugins = ["proxy"][server]host = "127.0.0.1"port = 8080enable_proxy = true[plugins.proxy]proxy_urls = ["http://127.0.0.1:9000"]white_list = ["/*"]timeout = 600启动：
framex run --load-builtin-plugins proxy --enable-proxy适合这些场景：
1. 旧算法服务已经在线2. 第三方团队只提供 HTTP API3. 不想立即重构旧服务4. 希望先统一入口，再逐步插件化二十、如何启用 Ray 执行
FrameX 支持本地执行和可选 Ray Serve 执行。官方说明它提供从本地执行到 Ray-backed serving 的路径，并且安装 Ray 支持可以用 pip install "framex-kit[ray]"。(GitHub)
安装：
pip install "framex-kit[ray]"配置：
[server]use_ray = truenum_cpus = 8启动：
PYTHONPATH=. framex run --use-ray --num-cpus 8适合 Ray 的场景：
1. 算法推理耗时较长2. 多个算法需要并发执行3. 模型调用容易阻塞4. 需要更高吞吐5. 后续希望横向扩展不过一开始不建议直接上 Ray。建议先用：
use_ray = false等单机本地模式稳定后，再迁移到 Ray。
二十一、生产环境建议
1. 不要让业务系统直接调用具体算法版本
不推荐：
业务系统 → /api/v1/algorithms/sentiment/v1/predict推荐：
业务系统 → /api/v1/workflow/text/analyze或者：
业务系统 → /api/v1/router/predict这样以后算法升级时，业务系统不用改。
2. 版本回滚必须简单
路由插件中应该有类似配置：
self.default_versions = {"sentiment": "v2",}当 v2 出问题，只需要改成：
self.default_versions = {"sentiment": "v1",}或者放入配置文件：
[plugins.algorithm_router]sentiment_default_version = "v1"3. 所有算法插件都要有健康检查
可以给每个插件加一个 /health 风格接口：
@on_request("/algorithms/sentiment/v2/health", methods=["GET"])asyncdefhealth(self) -> dict:return {"algorithm": "sentiment","version": "v2","status": "ok", }4. 所有算法结果都要带 metadata
例如：
{"algorithm":"sentiment","version":"v2","result":{"label":"negative","score":0.91},"metadata":{"model_name":"sentiment-bert","model_version":"2026-05-01","threshold":0.8,"latency_ms":32}}这样便于排查问题。
5. 插件间调用不要互相 import
不推荐：
from plugins.sentiment_v2 import SentimentV2Plugin推荐：
awaitself._call_remote_api("/api/v1/algorithms/sentiment/v2/predict", request={...})这是 FrameX 的核心设计之一：不同能力之间通过稳定接口调用，而不是依赖对方代码实现。官方文档也强调插件可以通过 FrameX 调用其他插件，而不是直接导入对方实现。(GitHub)
6. 注意 API 稳定性
PyPI 项目页明确提示：FrameX 当前可用，但应视为 actively evolving framework，而不是冻结稳定的平台 API；生产采用前需要测试你依赖的行为，尤其是 proxy、响应包装和 auth 集成。(PyPI)
所以生产使用时建议：
1. 固定 framex-kit 版本2. 不要无脑 pip install -U3. 为每个插件写单元测试4. 为跨插件调用写集成测试5. 上线前测试 response wrapper6. 测试 proxy 模式7. 测试 Ray 模式8. 测试异常处理和超时固定版本示例：
pip install framex-kit==0.3.9二十二、最终项目全貌
完成后你的项目是：
framex_algorithm_platform/├── config.toml├── app/│ ├── __init__.py│ ├── schemas.py│ └── utils.py└── plugins/ ├── __init__.py ├── preprocess.py ├── sentiment_v1.py ├── sentiment_v2.py ├── keyword_v1.py ├── risk_rule_v1.py ├── algorithm_router.py └── text_workflow.py启动：
PYTHONPATH=. framex run查看接口：
http://127.0.0.1:8080/docs测试完整流程：
curl -X POST "http://127.0.0.1:8080/api/v1/workflow/text/analyze" \ -H "Content-Type: application/json" \ -d '{ "text": "客服态度不好，物流超时，我要投诉", "sentiment_version": "latest" }'二十三、总结
用 framex-kit 做多算法、多版本协同系统，建议采用这个模式：
一个算法版本 = 一个插件一个算法族 = 多个版本插件一个 router 插件 = 统一选择版本一个 workflow 插件 = 负责编排多个算法一个 FrameX 服务 = 统一暴露 API 和插件间调用最推荐的架构是：
preprocesssentiment_v1sentiment_v2keyword_v1risk_rule_v1algorithm_routertext_workflow对应到你的 AI 项目，可以变成：
asr_paraformer_v1asr_whisper_v1intent_cls_v1intent_cls_v2rag_retriever_v1llm_answer_v1safety_check_v1tts_v1algorithm_routercallcenter_workflow这样就能实现：
多算法协同同算法多版本共存灰度发布快速回滚统一 API 入口插件独立开发旧服务平滑接入后续扩展 Ray 分布式执行
