103 lines
3.9 KiB
Python
103 lines
3.9 KiB
Python
import asyncio
|
|
from typing import Annotated
|
|
from autogen_agentchat.agents import AssistantAgent
|
|
from autogen_agentchat.messages import TextMessage
|
|
from autogen_core import CancellationToken
|
|
from autogen_ext.models.openai import OpenAIChatCompletionClient
|
|
|
|
# --- 第一部分：工具定义 ---
|
|
# 以后接上机器人时，把 print 替换成串口指令 / TTS 调用即可
|
|
|
|
async def set_expression(
|
|
expression: Annotated[str, "机器人要展示的表情，如：开心、疑惑、难过、待机"],
|
|
intensity: Annotated[int, "表情强度 1-10"] = 5
|
|
) -> str:
|
|
"""[模拟面部] 控制机器人头部的表情展示。"""
|
|
print(f"\n>>>>>> 🤖 表情更新: 【{expression}】 (强度: {intensity}/10) <<<<<<")
|
|
return f"已切换到【{expression}】表情。"
|
|
|
|
async def speak(
|
|
text: Annotated[str, "机器人要说的话，简短自然"]
|
|
) -> str:
|
|
"""[模拟 TTS] 机器人开口说话。以后接 TTS 引擎播放语音。"""
|
|
print(f">>>>>> 🔊 机器人说: {text} <<<<<<\n")
|
|
return "语音已播放。"
|
|
|
|
# --- 第二部分：配置 VLM 大脑 ---
|
|
async def start_simulated_head():
|
|
model_client = OpenAIChatCompletionClient(
|
|
model="Qwen/Qwen3-VL-8B-Instruct",
|
|
base_url="http://localhost:8000/v1",
|
|
api_key="EMPTY",
|
|
model_info={
|
|
"vision": True,
|
|
"function_calling": True,
|
|
"json_output": True,
|
|
"structured_output": True,
|
|
"family": "unknown",
|
|
"context_length": 4096,
|
|
}
|
|
)
|
|
|
|
brain = AssistantAgent(
|
|
name="RobotBrain",
|
|
model_client=model_client,
|
|
tools=[set_expression, speak],
|
|
system_message="""你是一个有情感的机器人伙伴，能感知用户状态并进行语言交流。
|
|
每次收到输入时，你必须：
|
|
1. 综合视觉信息和用户说的话，理解当前情境和用户的情绪/需求。
|
|
2. 调用 set_expression 展示合适的表情。
|
|
3. 调用 speak 用简短、温暖、自然的语言回应用户。
|
|
回应风格：像和老朋友聊天，不要太正式，有点个性和幽默感。"""
|
|
)
|
|
|
|
# --- 第三部分：交互循环 ---
|
|
# 模拟视觉上下文（真实项目中由摄像头实时提供）
|
|
visual_context = "视觉输入：用户坐在电脑前，表情平静，看着屏幕。"
|
|
|
|
print("=" * 50)
|
|
print(" 机器人已上线！输入 'quit' 退出")
|
|
print("=" * 50)
|
|
print(f"[当前视觉状态]: {visual_context}")
|
|
print("提示：输入 'v <描述>' 可以更新视觉状态，例如: v 用户在笑\n")
|
|
|
|
history = [] # 维护完整对话历史，让机器人记住上下文
|
|
|
|
while True:
|
|
try:
|
|
user_input = input("你说: ").strip()
|
|
except (EOFError, KeyboardInterrupt):
|
|
print("\n机器人下线，再见！")
|
|
break
|
|
|
|
if not user_input:
|
|
continue
|
|
|
|
if user_input.lower() in ("quit", "exit", "退出"):
|
|
await brain.on_messages(
|
|
[*history, TextMessage(content=f"{visual_context}\n用户说：「再见」", source="user")],
|
|
CancellationToken()
|
|
)
|
|
print("\n机器人下线，再见！")
|
|
break
|
|
|
|
# 支持临时更新视觉状态
|
|
if user_input.lower().startswith("v "):
|
|
visual_context = f"视觉输入：{user_input[2:].strip()}。"
|
|
print(f"[视觉状态已更新]: {visual_context}\n")
|
|
continue
|
|
|
|
# 合并视觉 + 语言输入
|
|
combined_input = f"{visual_context}\n用户说：「{user_input}」"
|
|
history.append(TextMessage(content=combined_input, source="user"))
|
|
|
|
response = await brain.on_messages(history, CancellationToken())
|
|
|
|
# 把本轮所有消息（工具调用、工具结果、最终回复）加入历史
|
|
if response.inner_messages:
|
|
history.extend(response.inner_messages)
|
|
history.append(response.chat_message)
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(start_simulated_head())
|