Nacos Python SDK 强势来袭，动态管理大模型Prompt！

# 导入Flask库，用于构建web应用
from flask import Flask

# 导入NacosClient，用于与Nacos服务器交互，实现配置管理和服务发现功能
from nacos import NacosClient

# 初始化Flask应用实例
app = Flask(__name__)

# 设置Nacos服务器地址，请替换为实际的Nacos服务器地址
SERVER_ADDRESSES = "Your nacos server address"

# 设置Nacos命名空间ID，请替换为实际的命名空间ID
NAMESPACE = "Your nacos namespace"

# 设置Nacos用户名和密码，用于认证访问Nacos服务器
USERNAME = 'Your user name'
PASSWORD = 'Your password'

client = NacosClient(server_addresses=SERVER_ADDRESSES, namespace=NAMESPACE, username=USERNAME, password=PASSWORD,
                     log_level='INFO')


# 定义路由，访问根路径'/'时返回的消息，展示来自Nacos的配置信息
@app.route('/')
def hello_world():
    # 使用flask的config对象获取名为"config"的配置项，展示配置内容
    return f'Hello, World! Config from Nacos: {app.config.get("config")}'


def init():
    # 服务注册：让Flask应用在启动时自动注册到Nacos，实现服务发现的自动化。heartbeat_interval可以调整后台心跳间隔时间，默认为5秒。
    client.add_naming_instance(service_name='my-flask-service', ip='localhost', port=5000, heartbeat_interval=5)

    # 设置Nacos中配置数据的数据ID和分组，默认分组为'DEFAULT_GROUP'
    data_id = 'test'
    group = 'DEFAULT_GROUP'

    # 从Nacos获取配置，并更新到Flask应用的config对象中，以便在应用中使用这些配置
    app.config.update(config=client.get_config(data_id=data_id, group=group))

    # 添加配置监听器，当Nacos中的配置发生变化时，自动更新Flask应用的config对象
    client.add_config_watcher(data_id=data_id, group=group,
                              cb=lambda cfg: app.config.update(config=cfg['content']))


if __name__ == '__main__':
    init()
    app.run()

# 导入FastAPI库，用于构建API服务
from fastapi import FastAPI
from nacos import NacosClient

app = FastAPI()

# 设置Nacos服务器地址，请替换为实际的Nacos服务器地址
SERVER_ADDRESSES = "Your nacos server address"

# 设置Nacos命名空间ID，请替换为实际的命名空间ID
NAMESPACE = "Your nacos namespace"

# 设置Nacos用户名和密码，用于认证访问Nacos服务器
USERNAME = 'Your user name'
PASSWORD = 'Your password'

client = NacosClient(server_addresses=SERVER_ADDRESSES, namespace=NAMESPACE, username=USERNAME, password=PASSWORD,
                     log_level='INFO')


# 定义一个异步函数，在FastAPI应用启动时执行
@app.on_event("startup")
async def startup_event():
    # 在启动时创建一个任务来初始化配置
    asyncio.create_task(init())


# 通过NacosClient获取配置，并存储在应用的状态(state)中，以便后续使用
async def load_config(data_id, group):
    app.state = {'config_data': client.get_config(data_id=data_id, group=group)}


# 异步函数，用于初始化应用所需的各种配置和监听器
async def init():
    data_id = 'test'
    group = 'DEFAULT_GROUP'
    await load_config(data_id, group)

    def on_config_change(cfg):
        # 当Nacos中的配置发生变化时，更新应用状态中的配置数据
        app.state = {'config_data': cfg['content']}

    client.add_config_watcher(data_id, group, cb=on_config_change)
    client.add_naming_instance(service_name='my-flask-service', ip='localhost', port=8000, heartbeat_interval=5)

# 定义一个GET请求的路由，返回简单的欢迎信息及当前从Nacos获取的配置数据
@app.get("/")
def hello_world():
    return f'Hello, World! Config from Nacos: {app.state["config_data"]}'


if __name__ == '__main__':
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)

# 定义一个名为NacosConfigManager的类，用于管理Nacos配置
class NacosConfigManager:
    def __init__(self):
        # 初始化方法，创建一个空字典用于缓存配置信息
        self.config_cache = {}

    def get_config(self, data_id, group="DEFAULT_GROUP", need_watch=False):
        """
        获取配置，如果缓存中没有则从Nacos获取并缓存。
        如果need_watch为True，则为该配置添加监听器以便在配置变化时自动更新。
        """
        if data_id not in self.config_cache:
            config = client.get_config(data_id=data_id, group=group)
            self.config_cache[data_id] = config

        if need_watch:
            client.add_config_watcher(data_id, group, cb=self.on_config_change)
        return self.config_cache[data_id]

    def on_config_change(self, cfg):
        self.config_cache[cfg['data_id']] = cfg['content']


config_manager = NacosConfigManager()

# 初始化异步函数
async def init():
    group = 'DEFAULT_GROUP'
    config_manager.get_config(promot_data_id, group, True)
    config_manager.get_config(app_id_data_id, group)
    config_manager.get_config(app_key_id, group)

@app.get("/")
def hello_world():
    # 通过百炼Application.call方法来调用模型，传递从Nacos获取的配置参数
    response = Application.call(app_id=config_manager.get_config(app_id_data_id),
                                prompt=config_manager.get_config(promot_data_id),
                                api_key=config_manager.get_config(app_key_id),
                                )
    if response.status_code != HTTPStatus.OK:
        print('request_id=%s, code=%s, message=%s\n' % (response.request_id, response.status_code, response.message))
    else:
        print('request_id=%s\n output=%s\n usage=%s\n' % (response.request_id, response.output, response.usage))

    return response.output.text