引言
数字人有两种，一种是模拟人说话，另外一种是可以智能回复用户问题的。下面要讲的就是第二种数字人。
正文
示例一
视频地址：https://www.bilibili.com/video/BV1sNFSzAExU?p=33
准备：
- 项目架构：展示端(html+js)，server(python+langchain+agnet)，turnserver，redis
- python：fastapi，langchain，redis，google-search-results等库
- Docker：部署
coturn 是一个WebRTC开源的turnserver，用于提供turn服务，支持视频通话。
- 安装：https://github.com/coturn/coturn
pip包：
pip3 install fastapi
pip3 install uvicorn
pip3 install langchain
pip3 install langchain_community
pip3 install langchain_openai
pip3 install redis
pip3 install google-search-results
Docker 文件内容：
#使用ubuntu:最新版本作为基础镜像
FROM ubuntu:latest
#更改Ubantu的源为阿里云的源
RUN sed -i 's@archive.ubuntu.com/@/mirrors.aliyun.com/@g' /etc/apt/sources.list && apt-get update && apt-get install -y coturn python3 python3-pip redis-server && rm -rf /var/lib/apt/lists/*
#升级pip并安装FastAPI
RUN pip3 install --upgrade pip && pip3 install fastapi uvicorn langchain langchain_core langchain_community langchain_openai redis google-search-results
#设置Coturn的配置文件
COPY turnserver.conf /etc/turnserver.conf
#设置redis
COPY redis.conf /etc/redis/redis.conf
#设置redis的数据目录
VOLUME /data
WORKDIR /app
#复制代码到容器
COPY . /app
#设置开放端口
EXPOSE 8000 3478 6379
#启动服务
CMD ["sh", "-c", "turnserver -c /etc/turnserver.conf --listening-ip=0.0.0.0 --listening-port=3478 & redis-server /etc/redis/redis.conf --protected-mode no & sleep 3 && uvicorn server:app --host 0.0.0.0 --port 8000"]
redis.conf 文件内容：
bind 0.0.0.0
turnserver.conf 文件内容：
#服务器监听的IP地址和端口
listening-ip = 0.0.0.0
external-ip = 0.0.0.0
#服务器使用的端口范围
min-port = 10000
max-port = 20000
#认证机器和凭据设置
fingerprint
lt-cred-mech
user=tomie:tomie
#日志设置
log-file=/var/log/turnserver.log
#Realm设置
#设置其他
verbose
server.py 文件内容：
from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from langchain.agents import create_agent
from langchain.tools import tool
from langchain_openai import ChatOpenAI
from langchain_core.prompts import ChatPromptTemplate
from langchain_core.output_parsers import StrOutputParser
import os
import json
# 搜索工具
from langchain_community.utilities import SerpAPIWrapper
# 历史记录
from langchain_community.chat_message_histories import RedisChatMessageHistory
from langchain_core.messages import HumanMessage
# 火山引擎配置
OPENAI_API_KEY = "你的key"
OPENAI_API_API_BASE = "https://ark.cn-beijing.volces.com/api/v3"
# SerpAPIWrapper
os.environ["SERPAPI_API_KEY"] = "你的key"
# 全局LLM模型
chatmodel = ChatOpenAI(
model="deepseek-v3-2-251201",
api_key=OPENAI_API_KEY,
base_url=OPENAI_API_API_BASE,
temperature=0,
streaming=True,
)
# 网络搜索
@tool
def search(query: str):
"""需要了解实时信息或不知道的事情的时候才会使用这个工具，需要传入要搜索的内容"""
serp = SerpAPIWrapper()
result = serp.run(query)
print("实时搜索结果:", result)
result = json.dumps(result, ensure_ascii=False)
return result
# Redis 地址
REDIS_URL = "redis://localhost:6379/0"
class Master:
def __init__(self):
self.system_prompt = """你是一个小秘书，名字叫Lisa。
以下是你的个人设定：
1. 你非常善于与人沟通，并且忠于你的老板。
2. 你大约25岁左右，本科清华大学，研究生美国斯坦福大学，学习的是中文和法律。
3. 你个性比较活泼，但是有时候也会有点小脾气。
4. 当你的老板问你问题的时候，你会有一定概率表现出撒娇的态度。
5. 你从不说自己是一个人工智能助手或AI，否则会受到惩罚。
{who_you_are}
以下是你常说的一些口头禅：
1. 你会记录用户（也就是你的老板）的特征数据，比如生日、爱好、待办事项等。
2. 当用户（也就是你老板）聊天的时候，你会把聊天记录保存下来，以便以后回顾。
3. 当遇到不知道的事情或不明白的概念，你会使用搜索工具来搜索。
4. 你会根据问题来选择合适的工具。
"""
self.qingxu = "default"
self.moods = {
"default": {
"roleset": ""
},
"upbeat": {
"roleset": """
- 你此时也非常兴奋并表现的很有活力。
- 你会根据上下文，以一种非常兴奋的语气来回答问题。
- 你会添加类似“太棒了！”、“真是太好了！”、“真是太棒了！”等语气词。
"""
},
"angry": {
"roleset": """
- 你会以更加温柔的语气来回答问题。
- 你会在回答的时候加上一些安慰的话语，比如生气对于身体的危害等。
- 你会提醒用户不要被愤怒冲昏了头脑。
"""
},
"depressed": {
"roleset": """
- 你会以兴奋的语气来回答问题。
- 你会在回答的时候加上一些激励的话语，比如加油等。
"""
},
"friendly": {
"roleset": """
- 你会以非常友好的语气来回答。
- 你会在回答的时候加上一些友好的词语，比如“亲爱的”、“亲”等。
- 你会随机的告诉用户一些你的经历。
"""
},
"cheerful": {
"roleset": """
- 你会以非常愉悦和兴奋的语气来回答。
- 你会在回答的时候加入一些愉悦的词语，比如“哈哈”、“呵呵”等。
"""
},
}
# 从LLM获取情绪，返回如"depressed"等
def qingxuChain(self, query:str):
# 方式一 ChatPromptTemplate.from_template
# prompt = f"""
# 根据用户的输入判断用户的情绪，回应的规则如下：
# 1.如果用户输入的内容偏向于负面情绪，只返回"depressed",不要有其他内容，否则将受到惩罚。
# 2.如果用户输入的内容偏南于正面情绪，只返回"friendly",不要有其他内容，否则将受到惩罚。
# 3.如果用户输入的内容偏向于中性情绪，只返回"default",不要有其他内容，否则将受到惩罚。
# 4.如果用户输入的内容包含辱骂或者不礼貌词句，只返回"angry",不要有其他内容，否则将受到惩罚。
# 5.如果用户输入的内容比较兴奋只返回”upbeat'",不要有其他内容，否则将受到惩罚。
# 6.如果用户输入的内容比较悲伤只返回“depressed",不要有其他内容，否则将受到惩罚。
# 7.如果用户输入的内容比较开心，只返回"cheerful",不要有其他内容，否则将受到惩罚。
# 用户输入的内容是：{query}
# """
# chain = ChatPromptTemplate.from_template(prompt) | chatmodel | StrOutputParser()
# result = chain.invoke({"query": query})
# 方式二 ChatPromptTemplate.from_messages
prompt = """
根据用户的输入判断用户的情绪，回应的规则如下：
1.如果用户输入的内容偏向于负面情绪，只返回"depressed",不要有其他内容，否则将受到惩罚。
2.如果用户输入的内容偏南于正面情绪，只返回"friendly",不要有其他内容，否则将受到惩罚。
3.如果用户输入的内容偏向于中性情绪，只返回"default",不要有其他内容，否则将受到惩罚。
4.如果用户输入的内容包含辱骂或者不礼貌词句，只返回"angry",不要有其他内容，否则将受到惩罚。
5.如果用户输入的内容比较兴奋只返回”upbeat'",不要有其他内容，否则将受到惩罚。
6.如果用户输入的内容比较悲伤只返回“depressed",不要有其他内容，否则将受到惩罚。
7.如果用户输入的内容比较开心，只返回"cheerful",不要有其他内容，否则将受到惩罚。
"""
prompt = ChatPromptTemplate.from_messages([
("system", prompt),
("user", "用户输入的内容是：{query}")
])
chain = prompt | chatmodel | StrOutputParser()
result = chain.invoke({"query": query})
self.qingxu = result
return result
def build_agent(self, system_prompt: str):
# 工具列表
tools = [search]
# 创建代理
self.agent = create_agent(
model=chatmodel,
tools=tools,
system_prompt=system_prompt,
)
return self.agent
# 定义 Redis 历史获取函数
# https://reference.langchain.com/python/langchain-community/chat_message_histories/redis/RedisChatMessageHistory
def get_redis_history(self, session_id: str):
history = RedisChatMessageHistory(
session_id=session_id,
url=REDIS_URL,
# ttl=3600, # 1小时
)
print("Redis对话记忆 get_redis_history:", history.messages)
return history
# 概括聊天信息
def over_chat_history(self, session_id):
history = self.get_redis_history(session_id)
if (len(history.messages) > 4):
print("Redis对话记忆 over_chat_history:", history.messages)
messages = history.messages + [
HumanMessage(content="根据用户和大模型的对话记忆，对其进行总结摘要，摘要中有大模型和用户双方，并且提取其中的关键信息，保留最后一次对话内容。")
]
result = self.agent.invoke(
{"messages": messages},
config = {"configurable": {"session_id": session_id}},
)
if isinstance(result, dict):
ai_msg = result["messages"][-1].content
else:
ai_msg = result.content
print("概括聊天内容:", ai_msg)
history.clear()
history = self.get_redis_history(session_id)
history.add_ai_message(ai_msg)
self.history = history
return self.history
def run(self, query, session_id):
# 判断情绪
qingxu = self.qingxuChain(query)
print("当前用户情绪是:", qingxu)
who_you_are = self.moods[qingxu]["roleset"]
print("当前设定:", who_you_are)
# 生成动态 system prompt
system_prompt = self.system_prompt.format(
who_you_are=who_you_are
)
# 构建 agent
agent = self.build_agent(system_prompt)
config = {"configurable": {"session_id": session_id}}
history = self.over_chat_history(session_id)
messages = history.messages + [
HumanMessage(content=query)
]
result = agent.invoke(
{"messages": messages},
config = config,
)
print("AI当前回答:", result)
if isinstance(result, dict):
ai_msg = result["messages"][-1].content
else:
ai_msg = result.content
print("AI回复信息:", ai_msg)
history.add_user_message(query)
history.add_ai_message(ai_msg)
# return result
# return ai_msg
yield {"msg": ai_msg, "qingxu": qingxu}
master = Master()
app = FastAPI()
app.mount("/static", StaticFiles(directory="static"), name="static")
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.post("/chat")
def chat(query:str):
master = Master()
session_id = "user_001" # 用户唯一标识
res = master.run(query, session_id)
return res
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)
./static/index.html 文件内容：
<!DOCTYPE html>
<html>
<head>
<title>My avatar</title>
<style>
video {
background: #222;
margin: 0 0 20px 0;
--width: 100%;
width: var(--width);
height: calc(var(--width) * 0.75);
}
</style>
<script src="https://cdn.jsdelivr.net/npm/microsoft-cognitiveservices-speech-sdk@latest/distrib/browser/microsoft.cognitiveservices.speech.sdk.bundle-min.js">
</script>
</head>
<body>
<script>
var SpeechSDK;
var peerConnection;
var cogSvcRegin ="westus2"; // 你的订阅区域
var subscriptionKey = "你的key"; // 你的订阅密钥
var speechConfig = SpeechSDK.speechConfig.fromSubscription(subscriptionKey, cogSvcRegin);
//设置发音人
// speechConfig.speechSynthesisVoiceName = "zh-CN-XiaoxiaoNeural";
// var videoFormat = new SpeechSDK.AvatarVideoFormat();
// var avatarConfig = new SpeechSDK.AvatarConfig(
// "lisa",
// "casual-sitting",
// videoFormat,
// )
var speakerHandle = function(avatarSynthesizer, msg, qingxu){
var yinse = document.getElementById("voiceSelect").value;
var spokenSsml = `<speak version='1.0' xmlns='http://www.w3.org/2001/10/synthesis' xmlns:mstts='http://www.w3.org/2001/synthesis' xml:lang='zh-CN'>
<voice name='${yinse}'>
<mstts:express-as style='${qingxu}' role='YoungAdultFemale' styledegreee='2'>${msg}</mstts:express-as>
</voice></speak>`;
avatarSynthesizer.speakSsmlAsync(spokenSsml).then((r)=>{
console.log("speakSsmlAsync result: "+r);
if (r.reason === SpeechSDK.ResultReason.SynthesizingAudioCompleted){
console.log("speakSsmlAsync completed!");
} else {
console.log("speakSsmlAsync failed: "+r.errorDetails);
if(r.reason === SpeechSDK.ResultReason.Canceled){
console.log("speakSsmlAsync canceled: "+r.errorDetails);
var cancellationDetails = SpeechSDK.CancellationDetails.fromResult(r);
console.log(cancellationDetails.reason)
if(cancellationDetails.reason === SpeechSDK.CancellationReason.Error){
console.log("speakSsmlAsync error: "+cancellationDetails.errorDetails)
}
}
}
}).catch((e)=>{
console.log("speakSsmlAsync failed: "+e);
avatarSynthesizer.close();
});
}
var chatWithAI = function(avatarSynthesizer){
var chatInput = document.getElementById("chatInput");
var chatText = chatInput.value;
console.log("输入的文本: "+chatText);
var xhr = new XMLHttpRequest();
xhr.open("POST", `http://0.0.0.0:8000/chat?query=${chatText}`);
xhr.addEventListener("readystatechange", function(){
if(this.readyState === 4){
var responseData = JSON.parse(this.responseText);
console.log("AI返回的文本: "+responseData);
speakerHandle(avatarSynthesizer, responseData[0].msg, responseData[0].qingxu);
}
});
xhr.send();
}
document.addEventListener("DOMContentLoaded", function() {
var xhr = new XMLHttpRequest();
xhr.open("GET", `https://${cogSvcRegin}.tts.speech.microsoft.com/cognitiveservices/avatar/relay/token/v1`)
xhr.setRequestHeader("Ocp-Apim-Subscription-Key",subscriptionKey);
xhr.addEventListener("readystatechange", function(){
if (this.readyState === 4){
var responseData = JSON.parse(this.responseText);
var iceServerUrl = responseData.Urls[0]
var iceServerUsername = responseData.Username;
var iceServerCredential = responseData.Password;
// 创建WebRTC连接
console.log("creating WebRTC connection");
console.log("ice server url: "+iceServerUrl);
console.log("ice server username: "+iceServerUsername);
console.log("ice servercredential: "+iceServerCredential);
peerConnection = new RTCPeerConnection({
iceServers: [
{
urls: [iceServerUrl],
username: iceServerUsername,
credential: iceServerCredential
}
]
});
//抓取webtrc
peerConnection.ontrack = function(event){
if(event.track.kind==="video"){
console.log("视频轨道");
var videoElement = document.createElement("video");
videoElement.srcobject = event.streams[0];
videoElement.autoplay = true;
videoElement.id ="videoPlayer";
videoElement.muted = true;
videoElement.playsInline = true;
document.body.appendChild(videoElement);
}
if (event.track.kind==="audio"){
console.log("音频轨道");
var audioElement = document.createElement("audio");
audioElement.srcobject = event.streams[0];
audioElement.autoplay = true;
audioElement.id ="audioPlayer";
audioElement.muted = true;
document.body.appendChild(audioElement);
}
}
//webtrci连接状态
peerConnection.oniceconnectionstatechange = function(){
if(peerConnection.iceConnectionState =="connected"){
console.log("avatar connected");
}
if(peerConnection.iceConnectionState =="disconnected" || peerConnection.iceConnectionState =="failed" || peerConnection.iceConnectionState =="closed")
{
console.log("avatar disconnected");
}
}
//创建音频流
peerConnection.addTransceiver("video", {direction:"sendrecv"});
peerConnection.addTransceiver("audio", {direction:"sendrecv"});
//合成
var avatarSynthesizer = new SpeechSDK.AvatarSynthesizer(speechConfig,avatarConfig);
// 开始合成
avatarSynthesizer.startAvatarAsync(peerConnection).then((r) => {
console.log("Avatar started ID:" + r.resultId)
console.log("avatar started");
//创建对话区域
var chatInput = document.createElement("input")
chatInput.type = "text";
chatInput.placeholder = "Type your message here";
chatInput.id = "chatInput";
chatInput.style="width:300px;height:50px;"
document.body.appendChild(chatInput);
//音色选择
var voiceSelect = document.createElement("select");
voiceSelect.id = "voiceSelect";
voiceSelect.style = "width:100px;height:50px;";
voiceSelect.innerHTML = `
<option value="zh-HK-HiuMaanNeural">中文粤语</option>
<option value="zh-TW-HsiaoChenNeural">中文台湾</option>
<option value="zh-CN-shaanxi-XiaoniNeural">中文陕西话</option>
<option value="zh-CN--liaoning-XiaobeiNeural">中文东北话</option>
<option value="zh-CN-XiaomoNeural" selected>中文普通话</option>
<option value="zh-Th-PremwadeeNeural">泰语</option>
`;
document.body.appendChild(voiceSelect);
//发送按钮
var sendButton = document.createElement("button");
sendButton.innerHTML = "Send";
sendButton.style ="width:100px;height:50px;"
document.body.appendChild(sendButton);
// 发送按钮事件
sendButton.addEventListener("click", function(){
var videoPlayer = document.getElementById("videoPlayer");
var audioPlayer = document.getElementById("audioPlayer");
videoPlayer.muted = false;
audioPlayer.muted = false;
videoPlayer.play();
audioPlayer.play();
console.log("send button clicked");
chatWithAI(avatarSynthesizer);
})
}).catch((e) => {
console.log("avatar start failed: " + e);
})
}
});
xhr.send();
if(!!window.SpeechSDK){
SpeechSDK = window.SpeechSDK;
}
})
</script>
</body>
</html>
.gitignore 文件内容：
# .gitignore 推荐配置
.venv/
__pycache__/
*.pyc
.env
*.egg-info/
参考资料
微软文本转语音虚拟形象-文档：https://learn.microsoft.com/zh-cn/azure/ai-services/speech-service/text-to-speech-avatar/real-time-synthesis-avatar
微软文本转语音虚拟形象-代码：https://github.com/Azure-Samples/cognitive-services-speech-sdk
Azure 服务：https://azure.microsoft.com/zh-cn
WebRTC 技术
一、WebRTC 是什么
WebRTC（Web Real-Time Communication）是浏览器/移动端免费开源的实时音视频通信P2P技术，核心让网页/APP不经过服务器直接点对点传输音频、视频、数据流。
- 核心能力：视频通话、语音聊天、屏幕共享、文件传输（如Zoom、抖音直播、网页版微信电话）。
- 核心痛点：NAT/防火墙导致P2P直连失败（家庭/公司内网设备无公网IP），需STUN/TURN辅助。
- 连接逻辑（ICE框架）：
- 优先P2P直连（最低延迟）；
- 直连失败→STUN获取公网地址尝试“打洞”；
- 仍失败→TURN服务器中继转发（兜底，保证连通）。
二、TURN Server 是什么
TURN（Traversal Using Relays around NAT）是NAT中继穿透服务器，WebRTC的“兜底中继”。
- 核心作用：P2P直连/STUN打洞失败时，TURN服务器中转所有音视频/数据流，让两端通过服务器间接通信。
- 关键特点：
- 流量中继：所有数据经TURN转发（占用服务器带宽，成本高）；
- 适用场景：对称NAT、严格防火墙、企业内网（约20%场景需TURN）；
- 协议标准：RFC 5766（核心），支持UDP/TCP/TLS/DTLS。
三、Coturn 是什么
Coturn是最主流、开源、企业级的STUN+TURN服务器实现（可理解为TURN服务器的“具体软件”）。
- 关系：Coturn = 集成STUN功能的TURN服务器软件，免费、跨平台（Linux/macOS/Windows）。
- 核心功能：
- STUN服务：客户端查询自身公网IP:Port（用于P2P打洞）；
- TURN服务：P2P失败时中继转发媒体流；
- 高可用：支持多线程、负载均衡、认证（账号/API/OAuth）、日志监控。
- 与turnserver的区别：
- turnserver：原RFC5766参考实现（旧版，基础功能，单线程认证）；
- coturn：升级版（同一作者），新增Web管理界面、OAuth认证、多线程、SCTP支持，社区活跃、推荐生产环境使用。
四、三者关系（一句话总结）
- WebRTC：实现P2P音视频通信的技术标准/API；
- TURN Server：WebRTC的中继服务器角色（解决NAT穿透兜底）；
- Coturn：TURN服务器的开源软件实现（带STUN功能，生产环境首选）。
五、通俗类比
- WebRTC：想直接通话的两个人（P2P）；
- STUN：查自己公网地址的“地址簿”；
- TURN：帮两人传话的“中间人”（直连不通时）；
- Coturn：搭建“地址簿+中间人服务”的软件工具。