devbox
Gausst松鼠会
这个屌丝很懒，什么也没留下！
关注作者
热门标签
热门文章
当前位置:   article > 正文

1 分钟高效集成 ChatGPT，Stable Diffusion 等 AIGC 模型最强教程

作者：Gausst松鼠会 | 2024-04-01 01:28:58

赞

踩

gpt stablediffusion搭建

1 分钟能做什么？集成 ChatGPT 到自己的公众号，小程序或者 APP？集成各种形式的 Stable Diffusion，让 AIGC 帮助自己的项目更有趣，更生动？本教程将会教大家如何 1 分钟高效集成 ChatGPT，Stable Diffusion 等 AIGC 模型到自己的项目中。(文末有视频和代码案例)

#1 分钟集成 ChatGPT

  1. 1. 注册并登录 https://aigcaas.cn网站

  2. b86760da3c041f7261356df36d361e72.png

  3. 2. 点击左侧 API Explorer，并选择分类“聊天机器人”，选择应用 ChatGPT，并选择Chat Completion：

  4. 21ae5f8e99910213e8d0a5b8e86fc6dd.png

    3. 在中间，按照提示的格式，输入对应的参数信息，例如：[{"role":"user","content":"你好"}]，并点击右侧的在线调用→发送请求按钮：

    c8f0adf75d00327b11798ec3054fe0ff.png

  5. 稍等片刻，即可看到接口完成调用，已经获得请求结果。如果想要进行流式响应，可以在流式响应输入框中输入stream，例如：

  6. 5c75ceb054cdcb4b746f0e95098313f8.png

  7. 4. 点击右侧的案例代码，根据自己需要获得对应的案例代码：0661c61424943a7a00129433b36bee57.png

  8. 5. 此时，可以将代码复制到本地，对代码的第12行-13行进行修改：7e6a15b187a0b0ca34bb208b79bf5bae.png

  9.     点击页面左侧密钥列表→新建，创建密钥信息：c94d09c2cf91b453ec361bf4af00dfeb.png

  10.   将密钥信息对应复制到代码12-13行：97a4423caab2cacc94b31be9991d4091.png

  11. 6. 完成粘贴之后，即可进行代码的测试，例如执行代码（可能涉及到一些依赖的安装，不同语言安装依赖方式略有不同，可以自行安装依赖）：39839278e2fd6c8121892e4a751307ee.png

  12.    完成调试之后，即可和自己的项目做集成。当然也可以根据自己的需求对参数内容进行相对应的优化和调整。

#1 分钟获取 Stable Diffusion

Stable Diffusion 是目前非常火热的文生图工具，根据文字即可生成图片， AIGCaaS 平台拥有数十款 Stable Diffusion 应用，例如以哈士奇作为关键词，进行不同类型，不同风格的图像生成，效果如下：

d818db1fa9750cc6d0b4ca58ca45cea7.png

接口获取的方式与 ChatGPT 基本一致：

  1. 1. 点击左侧 API Explorer，并选择分类“图像生成”，选择一个自己需要的场景，例如“Open Journey”：98d0bb580345671f8a06b47d7bad528c.png

  2. 2. 输入提示词，例如：1girl, white hair, golden eyes, beautiful eyes, detail, flower meadow, cumulonimbus clouds, lighting, detailed sky, garden，并点击发送请求按钮：4856f9de7e15742c7722ac3a78132fce.png

  3.     多次生成，超看效果：ee7bc514d66ab10e75755e1bda9938cb.png

  4.     当然，不同模型的表现可能不同，例如二次元生成的效果：78198c6b3139fe9e10eaa96817f9291e.png

  5. 3. 完成图片的生成，可以点击右侧的案例代码，根据自己需要获得对应的案例代码：827120b228c55e064fee9ce8832a5a31.png

  6. 4. 同样要对13-14行代码进行密钥信息的替换，替换完成即可在本地进行代码的运行：此时，通过返回的链接地址，下载数据即可：3baf1e5b08f439d2c2bfeded1b850588.png

#如何与微信公众号进行集成

以 Python 语言为例，通过 Flask 框架作为后端框架：

  1. # coding:utf-8
  2. import os
  3. import time
  4. import json
  5. import redis
  6. import random
  7. import base64
  8. import hashlib
  9. import logging
  10. import requests
  11. import xmltodict
  12. import urllib.parse
  13. import urllib.request
  14. from flask import Flask, request, abort, Response, render_template_string
  15.  
  16. # 获取随机字符串
  17. random_str = lambda count=100: "".join(random.sample('zyxwvutsrqponmlkjihgUIOPLKJHGFDSAZXCVBNM' * 10, count))
  18.  
  19. # 日志配置
  20. logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  21. logger = logging.getLogger(__name__)
  22.  
  23. # 常量
  24. WECHAT_TOKEN = "公众号 Token"
  25. WECHAT_APPID = "公众号 APPID"
  26. WECHAT_APPSECRET = "公众号 APP Secret"
  27. WECHAT_ENCODINGAESKEY = '公众号 Encoding AES Key'
  28. secret_id = 'AIGCaaS 平台密钥'  # 密钥信息
  29. secret_key = 'AIGCaaS 平台密钥'  # 密钥信息
  30.  
  31. db_config = {
  32.     "redis": {
  33.         "server": "192.168.211.1",
  34.         "port": 6379,
  35.     }
  36. }
  37.  
  38. app = Flask(__name__)
  39.  
  40. redis_conn = redis.Redis(host=db_config["redis"]["server"], port=db_config["redis"]["port"])
  41.  
  42. # 签名对象
  43. getSha256 = lambda content: hashlib.sha256(content.encode("utf-8")).hexdigest()
  44.  
  45. getTextResponse = lambda xml_dict, response_data: {
  46.     "xml":
  47.         {
  48.             "ToUserName": xml_dict.get("FromUserName"),
  49.             "FromUserName": xml_dict.get("ToUserName"),
  50.             "CreateTime": int(time.time()),
  51.             "MsgType": "text",
  52.             "Content": response_data
  53.         }
  54. }
  55.  
  56.  
  57. def chatgpt(user, content):
  58.     try:
  59.         messages = redis_conn.get('user-%s' % (user))
  60.         messages = json.loads(messages.decode("utf-8")) if messages else []
  61.         messages.append({"role": "user", "content": content})
  62.         redis_conn.setex('user-%s' % (user), 60 * 60 * 24 * 7, json.dumps(messages))
  63.         url = "https://api.aigcaas.cn/product/%s/api/%s" % ('chatgpt_chat', 'chat_com')
  64.         # 构建请求头
  65.         nonce = str(random.randint(1, 10000))
  66.         timestamp = str(int(time.time()))
  67.         headers = {
  68.             'SecretID': secret_id,
  69.             'Nonce': nonce,
  70.             'Token': getSha256(("%s%s%s" % (timestamp, secret_key, nonce))),
  71.             'Timestamp': timestamp,
  72.             'Content-Type': 'application/json'
  73.         }
  74.         # 构建请求 body
  75.         data = {"messages": messages}
  76.         # 获取响应
  77.         response = requests.request("POST", url, headers=headers, data=json.dumps(data))
  78.         result = json.loads(response.text)
  79.         logger.debug("result: %s" % result)
  80.         if result.get("status") == "Error":
  81.             redis_conn.delete('user-%s' % (user))
  82.             redis_conn.setex('user-last-text-message-%s' % user, 600, "系统检测到敏感信息，已经屏蔽，请重试")
  83.         else:
  84.             messages.append(result["choices"][0]["message"])
  85.             messages = messages[-10:]
  86.             redis_conn.setex('user-%s' % (user), 60 * 60 * 24, json.dumps(messages))
  87.             return result["choices"][0]["message"]["content"]
  88.     except Exception as e:
  89.         logger.error(e)
  90.         return e
  91.  
  92.  
  93. @app.route("/mp", methods=["GET", "POST"])
  94. def wechat():
  95.     # 接收微信服务器发送参数
  96.     signature = request.args.get("signature")
  97.     timestamp = request.args.get("timestamp")
  98.     nonce = request.args.get("nonce")
  99.     logger.debug([signature, timestamp, nonce])
  100.  
  101.     if not all([signature, timestamp, nonce]):
  102.         abort(400)
  103.  
  104.     li = [WECHAT_TOKEN, timestamp, nonce]
  105.     li.sort()
  106.     tmp_str = "".join(li)
  107.     sign = hashlib.sha1(tmp_str.encode("utf-8")).hexdigest()  # 进行sha1加密, 得到正确的签名值
  108.  
  109.     # 将自己计算的签名值, 与请求的签名参数进行对比, 如果相同, 则证明请求来自微信
  110.     if signature != sign:
  111.         abort(403)
  112.  
  113.     if request.method == "GET":
  114.         return request.args.get("echostr")
  115.  
  116.     xml_str = request.data
  117.  
  118.     if not xml_str:
  119.         abort(400)
  120.  
  121.     # 对xml字符串进行解析成字典
  122.     xml_dict = xmltodict.parse(xml_str)
  123.     xml_dict = xml_dict.get("xml")
  124.     logger.debug('xml_str: %s' % xml_str)
  125.     logger.debug('xml_dict: %s' % xml_dict)
  126.     user = xml_dict.get("FromUserName")
  127.  
  128.     # 新的信息处理
  129.     resp_dict = getTextResponse(xml_dict, "感谢您关注 AIGC Hub")
  130.     if xml_dict.get("MsgType") == "text":
  131.         response_data = chatgpt(user, xml_dict.get("Content"))
  132.         resp_dict = getTextResponse(xml_dict, response_data)
  133.     resp_xml_str = xmltodict.unparse(resp_dict)
  134.     logger.debug(resp_xml_str)
  135.     return resp_xml_str
  136.  
  137.  
  138. if __name__ == '__main__':
  139.     app.run(host='0.0.0.0', port=8000, debug=True, threaded=True)

整个项目非常简单：服务端接收到请求，将文字信息提取出来xml_dict.get("Content")，并通过chatgpt方法进行结果的获取，并将结果返回给用户。但是这里遇到一个比较尴尬的事情，因为 AIGC 模型往往运行时间会比较长，可能超过微信公众号的 5 秒钟要求，所以此处可以进行额外的处理：

  1. 1. 服务端接收到请求，将文字信息提取出来xml_dict.get("Content")

  2. 2. 异步调用chatgpt方法，如果 5 秒钟没有响应，则返回一个友好的提示：系统正在进行处理，请稍后回复"继续"查看结果，如果获得到了结果，则进行结果的返回

  3. 3. 如果没有拿到结果，用户稍后发送继续时，系统将结果返回给客户端

整体的实现逻辑为：

  1. # coding:utf-8
  2. import os
  3. import time
  4. import json
  5. import redis
  6. import random
  7. import base64
  8. import hashlib
  9. import logging
  10. import requests
  11. import xmltodict
  12. import urllib.parse
  13. import urllib.request
  14. from flask import Flask, request, abort, Response, render_template_string
  15. from concurrent.futures import ThreadPoolExecutor
  16.  
  17. # 创建线程池执行器
  18. executor = ThreadPoolExecutor(2)
  19.  
  20. # 获取随机字符串
  21. random_str = lambda count=100: "".join(random.sample('zyxwvutsrqponmlkjihgUIOPLKJHGFDSAZXCVBNM' * 10, count))
  22.  
  23. # 日志配置
  24. logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
  25. logger = logging.getLogger(__name__)
  26.  
  27. # 常量
  28. WECHAT_TOKEN = "公众号 Token"
  29. WECHAT_APPID = "公众号 APPID"
  30. WECHAT_APPSECRET = "公众号 APP Secret"
  31. WECHAT_ENCODINGAESKEY = '公众号 Encoding AES Key'
  32. secret_id = 'AIGCaaS 平台密钥'  # 密钥信息
  33. secret_key = 'AIGCaaS 平台密钥'  # 密钥信息
  34.  
  35. db_config = {
  36.     "redis": {
  37.         "server": "192.168.211.1",
  38.         "port": 6379,
  39.     }
  40. }
  41.  
  42. app = Flask(__name__)
  43.  
  44. redis_conn = redis.Redis(host=db_config["redis"]["server"], port=db_config["redis"]["port"])
  45.  
  46. # 签名对象
  47. getSha256 = lambda content: hashlib.sha256(content.encode("utf-8")).hexdigest()
  48.  
  49. getTextResponse = lambda xml_dict, response_data: {
  50.     "xml":
  51.         {
  52.             "ToUserName": xml_dict.get("FromUserName"),
  53.             "FromUserName": xml_dict.get("ToUserName"),
  54.             "CreateTime": int(time.time()),
  55.             "MsgType": "text",
  56.             "Content": response_data
  57.         }
  58. }
  59.  
  60.  
  61. def chatgpt(user, content):
  62.     try:
  63.         messages = redis_conn.get('user-%s' % (user))
  64.         messages = json.loads(messages.decode("utf-8")) if messages else []
  65.         messages.append({"role": "user", "content": content})
  66.         redis_conn.setex('user-%s' % (user), 60 * 60 * 24 * 7, json.dumps(messages))
  67.         url = "https://api.aigcaas.cn/product/%s/api/%s" % ('chatgpt_chat', 'chat_com')
  68.         # 构建请求头
  69.         nonce = str(random.randint(1, 10000))
  70.         timestamp = str(int(time.time()))
  71.         headers = {
  72.             'SecretID': secret_id,
  73.             'Nonce': nonce,
  74.             'Token': getSha256(("%s%s%s" % (timestamp, secret_key, nonce))),
  75.             'Timestamp': timestamp,
  76.             'Content-Type': 'application/json'
  77.         }
  78.         # 构建请求 body
  79.         data = {"messages": messages}
  80.         # 获取响应
  81.         response = requests.request("POST", url, headers=headers, data=json.dumps(data))
  82.         result = json.loads(response.text)
  83.         logger.debug("result: %s" % result)
  84.         if result.get("status") == "Error":
  85.             redis_conn.delete('user-%s' % (user))
  86.             redis_conn.setex('user-last-text-message-%s' % user, 600, "系统检测到敏感信息，已经屏蔽，请重试")
  87.         else:
  88.             messages.append(result["choices"][0]["message"])
  89.             messages = messages[-10:]
  90.             redis_conn.setex('user-%s' % (user), 60 * 60 * 24, json.dumps(messages))
  91.             redis_conn.setex('user-last-text-message-%s' % user, 600, result["choices"][0]["message"]["content"])
  92.     except Exception as e:
  93.         logger.error(e)
  94.         redis_conn.delete('user-%s' % (user))
  95.  
  96.  
  97. def get_access_token():
  98.     url = 'https://api.weixin.qq.com/cgi-bin/token?grant_type=client_credential&appid=%s&secret=%s' % (
  99.     WECHAT_APPID, WECHAT_APPSECRET)
  100.     response_data = urllib.request.urlopen(url).read().decode("utf-8")
  101.     logger.debug("response_data: %s" % response_data)
  102.     return json.loads(response_data)["access_token"]
  103.  
  104.  
  105. @app.route("/mp", methods=["GET", "POST"])
  106. def wechat():
  107.     # 接收微信服务器发送参数
  108.     signature = request.args.get("signature")
  109.     timestamp = request.args.get("timestamp")
  110.     nonce = request.args.get("nonce")
  111.     logger.debug([signature, timestamp, nonce])
  112.  
  113.     if not all([signature, timestamp, nonce]):
  114.         abort(400)
  115.  
  116.     li = [WECHAT_TOKEN, timestamp, nonce]
  117.     li.sort()
  118.     tmp_str = "".join(li)
  119.     sign = hashlib.sha1(tmp_str.encode("utf-8")).hexdigest()  # 进行sha1加密, 得到正确的签名值
  120.  
  121.     # 将自己计算的签名值, 与请求的签名参数进行对比, 如果相同, 则证明请求来自微信
  122.     if signature != sign:
  123.         abort(403)
  124.  
  125.     if request.method == "GET":
  126.         return request.args.get("echostr")
  127.  
  128.     xml_str = request.data
  129.  
  130.     if not xml_str:
  131.         abort(400)
  132.  
  133.     # 对xml字符串进行解析成字典
  134.     xml_dict = xmltodict.parse(xml_str)
  135.     xml_dict = xml_dict.get("xml")
  136.     logger.debug('xml_str: %s' % xml_str)
  137.     logger.debug('xml_dict: %s' % xml_dict)
  138.     user = xml_dict.get("FromUserName")
  139.  
  140.     # 对超时信息进行处理
  141.     resp_dict = None
  142.     if xml_dict.get("MsgType") == "text" and '继续' == xml_dict.get("Content").strip():
  143.         resp_dict = getTextResponse(xml_dict, '系统还在处理中，烦请再等一下发送"继续"查看结果')
  144.         text_status = redis_conn.get('user-last-text-message-status-%s' % user)
  145.         if text_status:
  146.             response_data = redis_conn.get('user-last-text-message-%s' % user)
  147.             if response_data:
  148.                 redis_conn.delete('user-last-text-message-status-%s' % user)
  149.                 redis_conn.delete('user-last-text-message-%s' % user)
  150.                 resp_dict = getTextResponse(xml_dict, response_data.decode("utf-8"))
  151.         else:
  152.             resp_dict = None
  153.  
  154.     # 新的信息处理
  155.     if not resp_dict:
  156.         redis_conn.delete('user-last-text-message-status-%s' % user)
  157.         redis_conn.delete('user-last-text-message-%s' % user)
  158.         # 配置默认信息
  159.         resp_dict = getTextResponse(xml_dict, "感谢您关注 AIGC Hub")
  160.         if xml_dict.get("MsgType") == "text":
  161.             redis_conn.setex('user-last-text-message-status-%s' % user, 600, '1')
  162.             executor.submit(chatgpt, user, xml_dict.get("Content"))
  163.             response_data = '系统正在进行处理，请稍后回复"继续"查看结果'
  164.             for i in range(1, 23):
  165.                 time.sleep(0.2)
  166.                 temp_response_data = redis_conn.get('user-last-text-message-%s' % user)
  167.                 if temp_response_data:
  168.                     response_data = temp_response_data.decode("utf-8")
  169.                     redis_conn.delete('user-last-text-message-status-%s' % user)
  170.                     break
  171.             resp_dict = getTextResponse(xml_dict, response_data)
  172.     resp_xml_str = xmltodict.unparse(resp_dict)
  173.     logger.debug(resp_xml_str)
  174.     return resp_xml_str
  175.  
  176.  
  177. if __name__ == '__main__':
  178.     app.run(host='0.0.0.0', port=8000, debug=True, threaded=True)

#总结

上面为大家介绍了如何用 1 分钟获取多种形式的 AIGC 应用，包括不限于 ChatGPT，Stable Diffusion 等，并通过与微信公众号集成的方式，为大家进行了简单的代码分享。当然，除了与公众号的集成还可以与小程序的集成，还可以与其他更多种形式的应用进行集成，AIGCaaS 平台系统为开发者提供开箱即用的 AIGC 应用 API，可以让开发者更简单，更方便，更快速的使用多种形式的 AIGC 应用。

常见问题：

  1. 1. 为什么我生成的 Stable Diffusion 图片是黑色的？在某些时候，系统进行了敏感检测，如果生成的图片存在敏感信息，系统会自动进行屏蔽；

  2. 2. 这个 AIGCaaS 项目收费么？AIGCaaS 为开发者提供了大量的低价 API，注册即赠送大量的试用与体验机会（以 ChatGPT 为例，可以体验大概2500次），即便超过免费体验此时，开发者也可以申请永久免费计划，进行长期的“薅羊毛”；

  3. 3. 使用过程中出现问题，如何联系技术支持？技术支持微信号：envless，有任何问题随时联系；

AIGC 爱好者交流群

e7653fbfefb9028b2a62d27050332e56.jpeg

------

我们创建了一个高质量的技术交流群，与优秀的人在一起，自己也会优秀起来，赶紧点击加群，享受一起成长的快乐。另外，如果你最近想跳槽的话，年前我花了2周时间收集了一波大厂面经，节后准备跳槽的可以点击这里领取！

推荐阅读

··································

你好，我是程序猿DD，10年开发老司机、阿里云MVP、腾讯云TVP、出过书创过业、国企4年互联网6年。从普通开发到架构师、再到合伙人。一路过来，给我最深的感受就是一定要不断学习并关注前沿。只要你能坚持下来，多思考、少抱怨、勤动手，就很容易实现弯道超车！所以，不要问我现在干什么是否来得及。如果你看好一个事情，一定是坚持了才能看到希望，而不是看到希望才去坚持。相信我，只要坚持下来，你一定比现在更好！如果你还没什么方向，可以先关注我，这里会经常分享一些前沿资讯，帮你积累弯道超车的资本。

声明：本文内容由网友自发贡献，不代表【wpsshop博客】立场，版权归原作者所有，本站不承担相应法律责任。如您发现有侵权的内容，请联系我们。转载请注明出处：https://www.wpsshop.cn/w/Gausst松鼠会/article/detail/346786
推荐阅读
相关标签

Copyright © 2003-2013 www.wpsshop.cn 版权所有，并保留所有权利。

  

闽ICP备14008679号