feat: 添加联网搜索、地理位置查询、天气查询

This commit is contained in:
JiajunLI
2026-03-03 17:20:54 +08:00
parent 9065ac77d6
commit c86e2458ef
3 changed files with 523 additions and 51 deletions

203
main.py
View File

@@ -1,12 +1,50 @@
import asyncio
import json
import sqlite3
import sys
from pathlib import Path
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
from autogen_ext.models.openai import _openai_client as openai_client_module
from autogen_ext.tools.mcp import StdioServerParams, mcp_server_tools
BASE_DIR = Path(__file__).resolve().parent
USER_DB_PATH = BASE_DIR / "users.db"
MODEL_CALL_TIMEOUT_SECONDS = 45
# --- 第一部分:本地工具(面部 + 语音,以后接硬件)---
def _patch_autogen_tool_schema_for_vllm() -> None:
"""
vLLM 目前会对 OpenAI 工具定义中的 `strict` 字段告警(即便 strict=False
这里做最小补丁:保留工具定义,移除该字段,避免无意义警告。
"""
if getattr(openai_client_module.convert_tools, "_strict_removed_patch", False):
return
original_convert_tools = openai_client_module.convert_tools
def convert_tools_without_strict(tools):
converted = original_convert_tools(tools)
for tool in converted:
fn = tool.get("function")
if isinstance(fn, dict):
fn.pop("strict", None)
return converted
convert_tools_without_strict._strict_removed_patch = True
openai_client_module.convert_tools = convert_tools_without_strict
async def _async_console_input(prompt: str) -> str:
"""在线程中执行阻塞 input避免阻塞事件循环。"""
return await asyncio.to_thread(input, prompt)
# --- 第一部分:工具定义 ---
# 以后接上机器人时,把 print 替换成串口指令 / TTS 调用即可
async def set_expression(
expression: Annotated[str, "机器人要展示的表情,如:开心、疑惑、难过、待机"],
@@ -16,15 +54,47 @@ async def set_expression(
print(f"\n>>>>>> 🤖 表情更新: 【{expression}】 (强度: {intensity}/10) <<<<<<")
return f"已切换到【{expression}】表情。"
async def speak(
text: Annotated[str, "机器人要说的话,简短自然"]
) -> str:
"""[模拟 TTS] 机器人开口说话。以后接 TTS 引擎播放语音"""
print(f">>>>>> 🔊 机器人说: {text} <<<<<<\n")
return "语音已播放。"
# --- 第二部分:直接读取用户档案(不经过 MCP避免多轮工具调用---
def _load_user_profile(user_name: str, db_path: str | Path = USER_DB_PATH) -> str:
"""在 Python 层直接读档案,注入到消息上下文,模型无需主动调用 get_user_profile"""
try:
with sqlite3.connect(db_path) as conn:
conn.row_factory = sqlite3.Row
user = conn.execute(
"SELECT * FROM users WHERE name = ?", (user_name,)
).fetchone()
if not user:
return f"用户 {user_name} 尚无历史记录,这是第一次见面。"
prefs = conn.execute(
"SELECT category, content FROM preferences WHERE user_name = ?",
(user_name,)
).fetchall()
conn.execute(
"UPDATE users SET last_seen = datetime('now') WHERE name = ?",
(user_name,)
)
return json.dumps({
"基本信息": {"姓名": user["name"], "年龄": user["age"], "上次见面": user["last_seen"]},
"偏好习惯": {p["category"]: p["content"] for p in prefs},
}, ensure_ascii=False)
except Exception as e:
return f"档案读取失败({e}),当作第一次见面。"
# --- 第三部分:启动大脑 ---
# --- 第二部分:配置 VLM 大脑 ---
async def start_simulated_head():
_patch_autogen_tool_schema_for_vllm()
# 连接 MCP Server只取写入类工具读档案由 Python 层直接处理)
server_params = StdioServerParams(
command=sys.executable,
args=["robot_mcp_server.py"],
)
all_mcp_tools = await mcp_server_tools(server_params)
# 过滤掉 get_user_profile模型无需主动调用它
mcp_tools = [t for t in all_mcp_tools if getattr(t, "name", "") != "get_user_profile"]
model_client = OpenAIChatCompletionClient(
model="Qwen/Qwen3-VL-8B-Instruct",
base_url="http://localhost:8000/v1",
@@ -33,70 +103,101 @@ async def start_simulated_head():
"vision": True,
"function_calling": True,
"json_output": True,
"structured_output": True,
# 关闭 structured_output,避免向 vLLM 发送 strict 字段告警。
"structured_output": False,
"family": "unknown",
"context_length": 4096,
"context_length": 32000,
}
)
brain = AssistantAgent(
name="RobotBrain",
model_client=model_client,
tools=[set_expression, speak],
system_message="""你是一个有情感的机器人伙伴,能感知用户状态并进行语言交流。
每次收到输入时,你必须
1. 综合视觉信息和用户说的话,理解当前情境和用户的情绪/需求。
2. 调用 set_expression 展示合适的表情。
3. 调用 speak 用简短、温暖、自然的语言回应用户。
回应风格:像和老朋友聊天,不要太正式,有点个性和幽默感。"""
tools=[set_expression] + mcp_tools,
reflect_on_tool_use=True,
system_message="""你是一个有情感的机器人伙伴。每条消息包含
[用户档案] 历史记录(已由系统预加载)
[视觉状态] 当前摄像头观察
[用户说] 用户这句话
回应规则:
1. 同一轮内调用set_expression表情+ 所有需要的查询工具get_weather/get_location/web_search+ 需要的用户信息工具upsert_user/set_preference
2. 工具执行完毕后,用简短、温暖、自然的语言直接回答用户——这段文字就是你的语音输出。
3. 不要说"我去查一下"之类的过渡语,直接完成任务并给出结果。""",
)
# --- 第部分:交互循环 ---
# 模拟视觉上下文(真实项目中由摄像头实时提供)
visual_context = "视觉输入:用户坐在电脑前,表情平静,看着屏幕。"
# --- 第部分:交互循环 ---
print("=" * 50)
print(" 机器人已上线!输入 'quit' 退出")
print("=" * 50)
print(f"[当前视觉状态]: {visual_context}")
try:
user_name = (await _async_console_input("请输入你的名字: ")).strip() or "用户"
except (EOFError, KeyboardInterrupt):
print("\n机器人下线,再见!")
return
visual_context = "视觉输入:用户坐在电脑前,表情平静,看着屏幕。"
print(f"\n[当前视觉状态]: {visual_context}")
print("提示:输入 'v <描述>' 可以更新视觉状态,例如: v 用户在笑\n")
history = [] # 维护完整对话历史,让机器人记住上下文
history = []
while True:
try:
user_input = input("你说: ").strip()
except (EOFError, KeyboardInterrupt):
print("\n机器人下线,再见!")
break
try:
while True:
try:
user_input = (await _async_console_input("你说: ")).strip()
except (EOFError, KeyboardInterrupt):
print("\n机器人下线,再见!")
break
if not user_input:
continue
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()
if user_input.lower() in ("quit", "exit", "退出"):
print("机器人下线,再见!")
break
if user_input.lower().startswith("v "):
visual_context = f"视觉输入:{user_input[2:].strip()}"
print(f"[视觉状态已更新]: {visual_context}\n")
continue
# Python 层直接读取档案并注入消息,模型无需发起额外工具调用
profile = _load_user_profile(user_name)
combined_input = (
f"[用户档案]\n{profile}\n\n"
f"[视觉状态] {visual_context}\n"
f"[用户说] {user_input}"
)
print("\n机器人下线,再见!")
break
history.append(TextMessage(content=combined_input, source="user"))
# 支持临时更新视觉状态
if user_input.lower().startswith("v "):
visual_context = f"视觉输入:{user_input[2:].strip()}"
print(f"[视觉状态已更新]: {visual_context}\n")
continue
# 只保留最近 6 条消息3轮对话防止超出 token 上限
# 用户档案每轮从数据库重新注入,不依赖长历史
if len(history) > 6:
history = history[-6:]
# 合并视觉 + 语言输入
combined_input = f"{visual_context}\n用户说:「{user_input}"
history.append(TextMessage(content=combined_input, source="user"))
try:
response = await asyncio.wait_for(
brain.on_messages(history, CancellationToken()),
timeout=MODEL_CALL_TIMEOUT_SECONDS,
)
except asyncio.TimeoutError:
print(">>>>>> ⚠️ 请求超时,请稍后重试或简化问题。 <<<<<<\n")
continue
except Exception as e:
print(f">>>>>> ⚠️ 本轮处理失败:{e} <<<<<<\n")
continue
response = await brain.on_messages(history, CancellationToken())
# 模型的文字回复就是语音输出reflect_on_tool_use=True 保证这里是 TextMessage
speech = response.chat_message.content
if speech and isinstance(speech, str):
print(f">>>>>> 🔊 机器人说: {speech} <<<<<<\n")
# 把本轮所有消息(工具调用、工具结果、最终回复加入历史
if response.inner_messages:
history.extend(response.inner_messages)
history.append(response.chat_message)
# 只把最终回复加入历史inner_messages 是事件对象不能序列化回模型
history.append(response.chat_message)
finally:
model_client.close()
if __name__ == "__main__":
asyncio.run(start_simulated_head())

358
robot_mcp_server.py Normal file
View File

@@ -0,0 +1,358 @@
"""
机器人用户档案 MCP Server
存储并维护用户基本信息、偏好习惯。
"""
import json
import logging
import sqlite3
from pathlib import Path
# 压制 mcp 库的 INFO 日志,只保留 WARNING 及以上
logging.basicConfig(level=logging.WARNING)
from mcp.server.fastmcp import FastMCP
DB_PATH = Path(__file__).parent / "users.db"
mcp = FastMCP("robot-user-db")
def _db_connect() -> sqlite3.Connection:
"""统一数据库连接入口,确保启用外键约束。"""
conn = sqlite3.connect(DB_PATH)
conn.execute("PRAGMA foreign_keys = ON")
return conn
def _create_preferences_table(conn: sqlite3.Connection) -> None:
conn.execute("""
CREATE TABLE preferences (
user_name TEXT NOT NULL,
category TEXT NOT NULL,
content TEXT NOT NULL,
updated_at TEXT DEFAULT (datetime('now')),
PRIMARY KEY (user_name, category),
FOREIGN KEY (user_name) REFERENCES users(name)
ON UPDATE CASCADE
ON DELETE CASCADE
)
""")
# --- 初始化数据库 ---
def _init_db():
with _db_connect() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS users (
name TEXT PRIMARY KEY,
age INTEGER,
created_at TEXT DEFAULT (datetime('now')),
last_seen TEXT
)
""")
preferences_exists = conn.execute(
"SELECT 1 FROM sqlite_master WHERE type='table' AND name='preferences'"
).fetchone() is not None
if not preferences_exists:
_create_preferences_table(conn)
_init_db()
# --- MCP 工具定义 ---
@mcp.tool()
def get_user_profile(user_name: str) -> str:
"""
获取用户档案(基本信息、所有偏好)。
在每次对话的第一轮调用,用于了解用户背景。
"""
with _db_connect() as conn:
conn.row_factory = sqlite3.Row
conn.execute(
"UPDATE users SET last_seen = datetime('now') WHERE name = ?",
(user_name,)
)
user = conn.execute(
"SELECT * FROM users WHERE name = ?", (user_name,)
).fetchone()
if not user:
return json.dumps(
{"found": False, "message": f"用户 {user_name} 尚无档案,这是第一次见面。"},
ensure_ascii=False
)
prefs = conn.execute(
"SELECT category, content FROM preferences WHERE user_name = ?",
(user_name,)
).fetchall()
return json.dumps({
"found": True,
"basic": {
"name": user["name"],
"age": user["age"],
"last_seen": user["last_seen"],
},
"preferences": {p["category"]: p["content"] for p in prefs},
}, ensure_ascii=False, indent=2)
@mcp.tool()
def upsert_user(user_name: str, age: int = None) -> str:
"""
创建或更新用户基本信息。
当得知用户姓名、年龄等基本信息时调用。
"""
with _db_connect() as conn:
existing = conn.execute(
"SELECT 1 FROM users WHERE name = ?",
(user_name,),
).fetchone() is not None
conn.execute(
"""INSERT INTO users (name, age, last_seen)
VALUES (?, ?, datetime('now'))
ON CONFLICT(name) DO UPDATE SET
age = COALESCE(excluded.age, users.age),
last_seen = datetime('now')""",
(user_name, age),
)
if existing:
return f"已更新用户 {user_name} 的档案。"
return f"已为 {user_name} 创建新档案。"
@mcp.tool()
def set_preference(user_name: str, category: str, content: str) -> str:
"""
更新用户的某项偏好或习惯(同一 category 只保留最新值)。
category 示例:'话题喜好''沟通风格''工作习惯''忌讳''饮食偏好'
在对话中发现新偏好时调用。
"""
with _db_connect() as conn:
# 保证 users 中存在主档,满足 preferences 外键约束。
conn.execute(
"""INSERT INTO users (name, last_seen)
VALUES (?, datetime('now'))
ON CONFLICT(name) DO UPDATE SET last_seen = datetime('now')""",
(user_name,),
)
conn.execute(
"""INSERT INTO preferences (user_name, category, content)
VALUES (?, ?, ?)
ON CONFLICT(user_name, category)
DO UPDATE SET content = excluded.content, updated_at = datetime('now')""",
(user_name, category, content)
)
return f"已更新 {user_name} 的偏好 [{category}]{content}"
# ============================================================
# 联网工具:定位 / 天气 / 搜索
# ============================================================
import requests
# WMO 天气代码 → 中文描述
_WMO = {
0: "晴天", 1: "基本晴朗", 2: "局部多云", 3: "阴天",
45: "", 48: "冻雾",
51: "轻微毛毛雨", 53: "毛毛雨", 55: "密集毛毛雨",
61: "小雨", 63: "中雨", 65: "大雨",
71: "小雪", 73: "中雪", 75: "大雪", 77: "冰粒",
80: "阵雨", 81: "中等阵雨", 82: "强阵雨",
85: "阵雪", 86: "强阵雪",
95: "雷阵雨", 96: "伴有冰雹的雷阵雨", 99: "强雷阵雨",
}
def _lookup_location_cn(ip: str = None) -> dict | None:
"""
使用国内 IP 归属地接口查询地理信息。
优先返回城市/省份;该接口不提供经纬度。
"""
params = {"json": "true"}
if ip:
params["ip"] = ip
resp = requests.get(
"https://whois.pconline.com.cn/ipJson.jsp",
params=params,
timeout=8,
)
# 该接口常见返回为 gbk 编码
resp.encoding = resp.apparent_encoding or "gbk"
raw = resp.text.strip()
if raw.startswith("var returnCitySN"):
raw = raw.split("=", 1)[-1].strip().rstrip(";")
data = json.loads(raw)
city = (data.get("city") or "").strip()
region = (data.get("pro") or "").strip()
ip_value = (data.get("ip") or ip or "").strip()
if not (city or region):
return None
return {
"city": city or region,
"region": region,
"country": "中国",
"lat": None,
"lon": None,
"ip": ip_value,
}
def _lookup_location_ipapi() -> dict | None:
"""回退定位:使用 ip-api。"""
data = requests.get(
"http://ip-api.com/json/",
params={"lang": "zh-CN", "fields": "status,city,regionName,country,lat,lon,query"},
timeout=8,
).json()
if data.get("status") != "success":
return None
return {
"city": data.get("city") or "",
"region": data.get("regionName") or "",
"country": data.get("country") or "",
"lat": data.get("lat"),
"lon": data.get("lon"),
"ip": data.get("query") or "",
}
def _lookup_location() -> dict | None:
"""统一定位入口:中国 IP 接口优先,失败回退 ip-api。"""
return _lookup_location_cn() or _lookup_location_ipapi()
def _geocode_city(city: str) -> tuple[float | None, float | None, str]:
"""根据城市名查经纬度,供天气查询使用。"""
geo = requests.get(
"https://geocoding-api.open-meteo.com/v1/search",
params={"name": city, "count": 1, "language": "zh"},
timeout=8,
).json()
results = geo.get("results")
if not results:
return None, None, city
return (
results[0].get("latitude"),
results[0].get("longitude"),
results[0].get("name", city),
)
def _resolve_weather_target(
city: str | None, lat: float | None, lon: float | None
) -> tuple[float | None, float | None, str | None, str | None]:
"""统一解析天气查询目标,减少重复分支。"""
auto_locate_error = "自动定位失败,请手动传入城市名。"
if lat is not None and lon is not None:
return lat, lon, city, None
if city:
lat, lon, city = _geocode_city(city)
if lat is None or lon is None:
return None, None, city, f"找不到城市:{city}"
return lat, lon, city, None
loc = _lookup_location()
if not loc:
return None, None, None, auto_locate_error
city = loc["city"] or city
lat, lon = loc.get("lat"), loc.get("lon")
if lat is None or lon is None:
if not city:
return None, None, None, auto_locate_error
lat, lon, city = _geocode_city(city)
if lat is None or lon is None:
return None, None, None, auto_locate_error
return lat, lon, city, None
@mcp.tool()
def get_location() -> str:
"""
通过 IP 地址获取当前地理位置(城市、省份、国家、经纬度)。
在查询天气前,或需要了解用户所在城市时调用。
"""
try:
loc = _lookup_location()
if not loc:
return "定位失败,请稍后再试。"
return json.dumps({
"城市": loc["city"],
"省份": loc["region"],
"国家": loc["country"],
"纬度": loc["lat"],
"经度": loc["lon"],
"IP": loc["ip"],
}, ensure_ascii=False)
except Exception as e:
return f"定位失败:{e}"
@mcp.tool()
def get_weather(city: str = None, lat: float = None, lon: float = None) -> str:
"""
获取实时天气信息。
可以传入城市名city或经纬度lat/lon若都不传则自动定位。
返回温度、天气状况、风速。
"""
try:
lat, lon, city, err = _resolve_weather_target(city, lat, lon)
if err:
return err
# 查询天气
resp = requests.get(
"https://api.open-meteo.com/v1/forecast",
params={
"latitude": lat, "longitude": lon,
"current": "temperature_2m,apparent_temperature,weather_code,wind_speed_10m,relative_humidity_2m",
"timezone": "auto",
},
timeout=10,
).json()
cur = resp.get("current", {})
code = cur.get("weather_code", -1)
return json.dumps({
"城市": city or f"{lat},{lon}",
"天气": _WMO.get(code, f"未知(code={code})"),
"温度": f"{cur.get('temperature_2m', '?')}°C",
"体感温度": f"{cur.get('apparent_temperature', '?')}°C",
"湿度": f"{cur.get('relative_humidity_2m', '?')}%",
"风速": f"{cur.get('wind_speed_10m', '?')} km/h",
}, ensure_ascii=False)
except Exception as e:
return f"天气查询失败:{e}"
@mcp.tool()
def web_search(query: str, max_results: int = 5) -> str:
"""
联网搜索,获取实时信息(新闻、百科、价格等)。
返回最多 max_results 条结果(标题 + 摘要 + 链接)。
"""
try:
query = query.strip()
if not query:
return "搜索关键词不能为空。"
max_results = max(1, min(max_results, 10))
from ddgs import DDGS
results = DDGS().text(query, max_results=max_results)
if not results:
return "搜索无结果。"
output = []
for i, r in enumerate(results, 1):
output.append(f"{i}. {r['title']}\n {r['body'][:150]}\n {r['href']}")
return "\n\n".join(output)
except Exception as e:
return f"搜索失败:{e}"
if __name__ == "__main__":
mcp.run()

13
start_vllm.sh Normal file
View File

@@ -0,0 +1,13 @@
#!/bin/bash
# 启动 vLLM 服务器脚本
# 用法: bash start_vllm.sh
python -m vllm.entrypoints.openai.api_server \
--model Qwen/Qwen3-VL-8B-Instruct \
--trust-remote-code \
--port 8000 \
--gpu-memory-utilization 0.85 \
--max-model-len 32000 \
--enable-auto-tool-choice \
--tool-call-parser hermes \
--uvicorn-log-level warning