Skip to content

FEAT: 作业提交,自动对话10次 #98

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: main
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions selected_homework/role-play/README.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
### 实现 role-play 对话数据生成工具,要求包含下列功能:

+ 给定两个角色的人设,调用 CharacterGLM 交替生成他们的回复。

streamlit run --server.address 127.0.0.1 characterglm_api_demo_streamlit.py
183 changes: 183 additions & 0 deletions selected_homework/role-play/api.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,183 @@
import requests
import time
import os
from typing import Generator

import jwt

from data_types import TextMsg, ImageMsg, TextMsgList, MsgList, CharacterMeta


# 智谱开放平台API key,参考 https://open.bigmodel.cn/usercenter/apikeys
os.environ["API_KEY"] = "04559ca20c345d282e76d07238d03ad9.PfCA5xBWPWBGHnzY"
API_KEY: str = os.getenv("API_KEY", "c175ae7f0ba6a73eb22e4ec0cddcdcc9.55fbjTiX2iacE9wG")


class ApiKeyNotSet(ValueError):
pass


def verify_api_key_not_empty():
if not API_KEY:
raise ApiKeyNotSet


def generate_token(apikey: str, exp_seconds: int) -> str:
# reference: https://open.bigmodel.cn/dev/api#nosdk
try:
id, secret = apikey.split(".")
except Exception as e:
raise Exception("invalid apikey", e)

payload = {
"api_key": id,
"exp": int(round(time.time() * 1000)) + exp_seconds * 1000,
"timestamp": int(round(time.time() * 1000)),
}

return jwt.encode(
payload,
secret,
algorithm="HS256",
headers={"alg": "HS256", "sign_type": "SIGN"},
)


def get_characterglm_response(messages: TextMsgList, meta: CharacterMeta) -> Generator[str, None, None]:
""" 通过http调用characterglm """
# Reference: https://open.bigmodel.cn/dev/api#characterglm
verify_api_key_not_empty()
url = "https://open.bigmodel.cn/api/paas/v3/model-api/charglm-3/sse-invoke"
resp = requests.post(
url,
headers={"Authorization": generate_token(API_KEY, 1800)},
json=dict(
model="charglm-3",
meta=meta,
prompt=messages,
incremental=True)
)
resp.raise_for_status()

# 解析响应(非官方实现)
sep = b':'
last_event = None
for line in resp.iter_lines():
if not line or line.startswith(sep):
continue
field, value = line.split(sep, maxsplit=1)
if field == b'event':
last_event = value
elif field == b'data' and last_event == b'add':
yield value.decode()


def get_characterglm_response_via_sdk(messages: TextMsgList, meta: CharacterMeta) -> Generator[str, None, None]:
""" 通过旧版sdk调用characterglm """
# 与get_characterglm_response等价
# Reference: https://open.bigmodel.cn/dev/api#characterglm
# 需要安装旧版sdk,zhipuai==1.0.7
import zhipuai
verify_api_key_not_empty()
zhipuai.api_key = API_KEY
response = zhipuai.model_api.sse_invoke(
model="charglm-3",
meta= meta,
prompt= messages,
incremental=True
)
for event in response.events():
if event.event == 'add':
yield event.data


def get_chatglm_response_via_sdk(messages: TextMsgList) -> Generator[str, None, None]:
""" 通过sdk调用chatglm """
# reference: https://open.bigmodel.cn/dev/api#glm-3-turbo `GLM-3-Turbo`相关内容
# 需要安装新版zhipuai
from zhipuai import ZhipuAI
verify_api_key_not_empty()
client = ZhipuAI(api_key=API_KEY) # 请填写您自己的APIKey
response = client.chat.completions.create(
model="glm-3-turbo", # 填写需要调用的模型名称
messages=messages,
stream=True,
)
for chunk in response:
yield chunk.choices[0].delta.content


def generate_role_appearance(role_profile: str) -> Generator[str, None, None]:
""" 用chatglm生成角色的外貌描写 """

instruction = f"""
请从下列文本中,抽取人物的外貌描写。若文本中不包含外貌描写,请你推测人物的性别、年龄,并生成一段外貌描写。要求:
1. 只生成外貌描写,不要生成任何多余的内容。
2. 外貌描写不能包含敏感词,人物形象需得体。
3. 尽量用短语描写,而不是完整的句子。
4. 不要超过50字
文本:
{role_profile}
"""
return get_chatglm_response_via_sdk(
messages=[
{
"role": "user",
"content": instruction.strip()
}
]
)


def generate_chat_scene_prompt(messages: TextMsgList, meta: CharacterMeta) -> Generator[str, None, None]:
""" 调用chatglm生成cogview的prompt,描写对话场景 """
instruction = f"""
阅读下面的角色人设与对话,生成一段文字描写场景。
{meta['bot_name_1']}的人设:
{meta['bot_info_1']}
""".strip()

if meta["bot_info_2"]:
instruction += f"""
{meta["bot_name_2"]}的人设:
{meta["bot_info_2"]}
""".rstrip()

if messages:
instruction += "\n\n对话:" + '\n'.join((meta['bot_name_1'] if msg['role'] == "user_1" else meta['bot_name_2']) + ':' + msg['content'].strip() for msg in messages)

instruction += """
要求如下:
1. 只生成场景描写,不要生成任何多余的内容
2. 描写不能包含敏感词,人物形象需得体
3. 尽量用短语描写,而不是完整的句子
4. 不要超过50字
""".rstrip()
print(instruction)

return get_chatglm_response_via_sdk(
messages=[
{
"role": "user",
"content": instruction.strip()
}
]
)


def generate_cogview_image(prompt: str) -> str:
""" 调用cogview生成图片,返回url """
# reference: https://open.bigmodel.cn/dev/api#cogview
from zhipuai import ZhipuAI
client = ZhipuAI(api_key=API_KEY) # 请填写您自己的APIKey

response = client.images.generations(
model="cogview-3", #填写需要调用的模型名称
prompt=prompt
)
return response.data[0].url

215 changes: 215 additions & 0 deletions selected_homework/role-play/characterglm_api_demo_streamlit.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,215 @@
"""
一个简单的demo,调用CharacterGLM实现角色扮演,调用CogView生成图片,调用ChatGLM生成CogView所需的prompt。
依赖:
pyjwt
requests
streamlit
zhipuai
python-dotenv
运行方式:
```bash
streamlit run characterglm_api_demo_streamlit.py
```
"""
import os
import itertools
from typing import Iterator, Optional

import streamlit as st
from dotenv import load_dotenv

# 通过.env文件设置环境变量
# reference: https://github.com/theskumar/python-dotenv
load_dotenv()

import api
from api import get_characterglm_response
from data_types import TextMsg, filter_text_msg

st.set_page_config(page_title="CharacterGLM API Demo", page_icon="🤖", layout="wide")
debug = os.getenv("DEBUG", "").lower() in ("1", "yes", "y", "true", "t", "on")


def update_api_key(key: Optional[str] = None):
if debug:
print(f'update_api_key. st.session_state["API_KEY"] = {st.session_state["API_KEY"]}, key = {key}')
key = key or st.session_state["API_KEY"]
if key:
api.API_KEY = key


# 设置API KEY
api_key = st.sidebar.text_input("API_KEY", value=os.getenv("API_KEY", ""), key="API_KEY", type="password",
on_change=update_api_key)
update_api_key(api_key)

# 初始化
if "history" not in st.session_state:
st.session_state["history"] = []
if "meta" not in st.session_state:
st.session_state["meta"] = {
"user_info": "",
"bot_info": "",
"bot_name": "",
"user_name": ""
}


def init_session():
st.session_state["history"] = []


# 4个输入框,设置meta的4个字段
meta_labels = {
"bot_name": "角色名",
"user_name": "用户名",
"bot_info": "角色人设",
"user_info": "用户人设"
}


# 2x2 layout
with st.container():
col1, col2 = st.columns(2)
with col1:
st.text_input(label="角色名", value="小白", key="bot_name",
on_change=lambda: st.session_state["meta"].update(bot_name=st.session_state["bot_name"]),
help="模型所扮演的角色的名字,不可以为空")
st.text_area(label="角色人设", value="善良", key="bot_info",
on_change=lambda: st.session_state["meta"].update(bot_info=st.session_state["bot_info"]),
help="角色的详细人设信息,不可以为空")

with col2:
st.text_input(label="用户名", value="小黑", key="user_name",
on_change=lambda: st.session_state["meta"].update(user_name=st.session_state["user_name"]),
help="用户的名字,默认为小黑")
st.text_area(label="用户人设", value="内向", key="user_info",
on_change=lambda: st.session_state["meta"].update(user_info=st.session_state["user_info"]),
help="用户的详细人设信息,可以为空")


def verify_meta() -> bool:
# 检查`角色名`和`角色人设`是否空,若为空,则弹出提醒
if st.session_state["meta"]["bot_name"] == "" or st.session_state["meta"]["bot_info"] == "":
st.session_state["meta"]["bot_name"] = "小白"
st.session_state["meta"]["bot_info"] = "善良"
return True
else:
return True


button_labels = {
"clear_meta": "清空人设",
"clear_history": "清空对话历史",
"gen_picture": "生成图片"
}
if debug:
button_labels.update({
"show_api_key": "查看API_KEY",
"show_meta": "查看meta",
"show_history": "查看历史"
})

# 在同一行排列按钮
with st.container():
n_button = len(button_labels)
cols = st.columns(n_button)
button_key_to_col = dict(zip(button_labels.keys(), cols))

with button_key_to_col["clear_meta"]:
clear_meta = st.button(button_labels["clear_meta"], key="clear_meta")
if clear_meta:
st.session_state["meta"] = {
"user_info": "",
"bot_info": "",
"bot_name": "",
"user_name": ""
}
st.rerun()

with button_key_to_col["clear_history"]:
clear_history = st.button(button_labels["clear_history"], key="clear_history")
if clear_history:
init_session()
st.rerun()

if debug:
with button_key_to_col["show_api_key"]:
show_api_key = st.button(button_labels["show_api_key"], key="show_api_key")
if show_api_key:
print(f"API_KEY = {api.API_KEY}")

with button_key_to_col["show_meta"]:
show_meta = st.button(button_labels["show_meta"], key="show_meta")
if show_meta:
print(f"meta = {st.session_state['meta']}")

with button_key_to_col["show_history"]:
show_history = st.button(button_labels["show_history"], key="show_history")
if show_history:
print(f"history = {st.session_state['history']}")

# 展示对话历史
for msg in st.session_state["history"]:
if msg["role"] == "user":
with st.chat_message(name="user", avatar="user"):
st.markdown(msg["content"])
elif msg["role"] == "assistant":
with st.chat_message(name="assistant", avatar="assistant"):
st.markdown(msg["content"])
else:
raise Exception("Invalid role")


with st.chat_message(name="user", avatar="user"):
message_placeholder_1 = st.empty()
with st.chat_message(name="assistant", avatar="assistant"):
message_placeholder_2 = st.empty()


def output_stream_response(response_stream: Iterator[str], placeholder):
content = ""
for content in itertools.accumulate(response_stream):
placeholder.markdown(content)
return content


def start_chat():
# query = st.chat_input("开始对话吧")
query = '你好, 我是小白'
role_play = "user"
if not query:
return
else:
for i in range(10):
if not verify_meta():
return
if not api.API_KEY:
st.error("未设置API_KEY")

if role_play == "user":
message_placeholder_1.markdown(query)
place_holder = message_placeholder_2
else:
message_placeholder_2.markdown(query)
place_holder = message_placeholder_1
st.session_state["history"].append(TextMsg({"role": role_play, "content": query}))

response_stream = get_characterglm_response(filter_text_msg(st.session_state["history"]),
meta=st.session_state["meta"])
bot_response = output_stream_response(response_stream, place_holder)
print(bot_response)
if not bot_response:
message_placeholder_2.markdown("生成出错")
st.session_state["history"].pop()
# else:
# st.session_state["history"].append(TextMsg({"role": "assistant", "content": bot_response}))
if role_play == "user":
role_play = "assistant"
else:
role_play = "user"
query = bot_response

start_chat()
51 changes: 51 additions & 0 deletions selected_homework/role-play/data_types.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,51 @@
"""
相关数据类型的定义
"""
from typing import Literal, TypedDict, List, Union, Optional, TYPE_CHECKING
if TYPE_CHECKING:
import streamlit.elements.image

class BaseMsg(TypedDict):
pass


class TextMsg(BaseMsg):
"""文本消息"""

# 在类属性标注的下一行用三引号注释,vscode中
role: Literal["user", "assistant"]
"""消息来源"""
content: str
"""消息内容"""


class ImageMsg(BaseMsg):
"""图片消息"""
role: Literal["image"]
image: "streamlit.elements.image.ImageOrImageList"
"""图片内容"""
caption: Optional[Union[str, List[str]]]
"""说明文字"""


Msg = Union[TextMsg, ImageMsg]
TextMsgList = List[TextMsg]
MsgList = List[Msg]


class CharacterMeta(TypedDict):
"""角色扮演设定,它是CharacterGLM API所需的参数"""
user_info: str
user_name: str


def filter_text_msg(messages: MsgList) -> TextMsgList:
return [m for m in messages if m["role"] != "image"]


if __name__ == "__main__":
# 尝试在VSCode等IDE中自己敲一遍下面的代码,观察IDE能提供哪些代码提示
text_msg = TextMsg(role="user")
text_msg["content"] = "42"
print(type(text_msg))
print(text_msg)
Binary file added selected_homework/role-play/后台截图.png
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
Binary file added selected_homework/role-play/界面截图.png
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
8 changes: 8 additions & 0 deletions selected_homework/毕业总结.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,8 @@
目前还是一个普通的后端开发工程师,公司对于我提出的各种技术变化,给出的回应只有会议上的沉默和点头。
所以自身还是求变的我,选择了参加AI训练营,希望借此了解更多更好玩的,带来更多便捷的技术。更希望借此了解的技术,能够突破自己的瓶颈。

通过彭老师的教授,学习了关于openAI、LangChain等相关的知识结构和应用。
这部分虽然还没有在工作中找到实际使用的地方,但相信通过此次学习,能够对我将来的工作带来很多很多帮助。

感谢彭老师、感谢群里的助教豆浆(私下麻烦了他很多次),也很感谢群里热心的小伙伴~
希望将来真的有机会从事到AI应用相关的工作~