LangChain+Gradio+千问大模型实现AI聊天助手
·
1. 非流式输出模式
需要安装依赖:
pip install gradio
pip install langchain
pip install langchain-openai
核心代码:
import gradio as gr
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
# 核心配置
API_KEY = "XXXXXXX" # 阿里百炼 API_KEY
BASE_URL = "XXXXXXXXXX" # 对应模型的 URL
MODEL_NAME = "qwen3.8-max"
TEMPERATURE = 0.7
# 使用init_chat_model初始化模型,兼容OpenAI兼容接口
llm = init_chat_model(
model=MODEL_NAME,
model_provider="openai",
api_key=API_KEY,
base_url=BASE_URL,
temperature=TEMPERATURE
)
def chat_with_ai(message, history):
"""
history:Gradio messages格式:[{"role":"user","content":"xxx"}, {"role":"assistant","content":"xxx"}]
转成langchain消息对象
"""
msg_list = [SystemMessage(content="你是一个 helpful 的助手。")]
# 遍历messages格式历史,转为LangChain对象
for msg in history:
role = msg["role"]
content = msg["content"]
if role == "user":
msg_list.append(HumanMessage(content=content))
elif role == "assistant":
msg_list.append(AIMessage(content=content))
msg_list.append(HumanMessage(content=message))
try:
resp = llm.invoke(msg_list)
return resp.content
except Exception as e:
return f"调用失败:{str(e)}"
with gr.Blocks(title="AI聊天助手") as demo:
gr.Markdown("# AI聊天助手")
chatbot = gr.Chatbot(height=500)
msg_box = gr.Textbox(placeholder="输入你的问题...", label="输入")
clear_btn = gr.Button("清空对话")
def on_submit(user_text, chat_history):
if not user_text.strip():
return "", chat_history
reply = chat_with_ai(user_text, chat_history)
# 追加 messages字典格式,不再用元组!
chat_history.append({"role": "user", "content": user_text})
chat_history.append({"role": "assistant", "content": reply})
return "", chat_history
msg_box.submit(
fn=on_submit,
inputs=[msg_box, chatbot],
outputs=[msg_box, chatbot]
)
clear_btn.click(fn=lambda: [], inputs=None, outputs=chatbot, queue=False)
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", share=False)
效果图:
2. 流式输出模式
import gradio as gr
from langchain.chat_models import init_chat_model
from langchain_core.messages import SystemMessage, HumanMessage, AIMessage
# 核心配置
API_KEY = "xxxx"
BASE_URL = "xxxx"
MODEL_NAME = "qwen3.8-max"
TEMPERATURE = 0.7
# 初始化模型
llm = init_chat_model(
model=MODEL_NAME,
model_provider="openai",
api_key=API_KEY,
base_url=BASE_URL,
temperature=TEMPERATURE
)
def chat_stream_generator(message, history):
"""流式生成器,返回逐块文本"""
msg_list = [SystemMessage(content="你是一个 helpful 的助手。")]
for msg in history:
role = msg["role"]
content = msg["content"]
if role == "user":
msg_list.append(HumanMessage(content=content))
elif role == "assistant":
msg_list.append(AIMessage(content=content))
msg_list.append(HumanMessage(content=message))
full_content = ""
try:
for chunk in llm.stream(msg_list):
full_content += chunk.content
yield full_content
except Exception as e:
yield f"调用失败:{str(e)}"
with gr.Blocks(title="AI聊天助手") as demo:
gr.Markdown("# AI聊天助手")
chatbot = gr.Chatbot(height=500)
msg_box = gr.Textbox(placeholder="输入你的问题...", label="输入")
clear_btn = gr.Button("清空对话")
def on_submit(user_text, chat_history):
if not user_text.strip():
return "", chat_history
# 先把用户消息加入历史
chat_history.append({"role": "user", "content": user_text})
# 初始化空的AI回复
chat_history.append({"role": "assistant", "content": ""})
# 流式迭代
for partial_text in chat_stream_generator(user_text, chat_history[:-2]):
# 更新最后一条assistant消息
chat_history[-1]["content"] = partial_text
yield "", chat_history
msg_box.submit(
fn=on_submit,
inputs=[msg_box, chatbot],
outputs=[msg_box, chatbot]
)
clear_btn.click(fn=lambda: [], inputs=None, outputs=chatbot, queue=False)
if __name__ == "__main__":
demo.launch(server_name="0.0.0.0", share=False)
更多推荐

所有评论(0)