"""
Agent Loop — 内置 AI Agent 运行时
使用用户的 API Key 调用 DeepSeek/OpenAI 的 Function Calling,
实现 思考→行动→观察→反思 的 Agent 循环。
每个用户用自己的 Key,完全隔离,不共享费用。
"""
import json
import re
import os
import subprocess
import tempfile
import time
from datetime import datetime
from urllib.parse import quote_plus
import requests
# ═══════════════════════════════════════════════════════════════
# Tool Definitions (OpenAI Function Calling 格式)
# ═══════════════════════════════════════════════════════════════
TOOLS = [
{
"type": "function",
"function": {
"name": "web_search",
"description": "搜索互联网获取最新信息。当需要查找实时信息、新闻、数据或不确定的事实时使用此工具。",
"parameters": {
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "搜索关键词,中英文均可。例如: '2026年AI发展趋势'",
},
"num_results": {
"type": "integer",
"description": "返回结果数量,默认5",
"default": 5,
},
},
"required": ["query"],
},
},
},
{
"type": "function",
"function": {
"name": "read_webpage",
"description": "读取指定网页的内容。当用户提供URL需要查看网页内容时使用。",
"parameters": {
"type": "object",
"properties": {
"url": {
"type": "string",
"description": "网页URL,例如: 'https://example.com/article'",
},
},
"required": ["url"],
},
},
},
{
"type": "function",
"function": {
"name": "get_current_time",
"description": "获取当前日期和时间。当需要知道现在是什么时间、日期或计算时间差时使用。",
"parameters": {
"type": "object",
"properties": {},
},
},
},
{
"type": "function",
"function": {
"name": "run_code",
"description": "在沙箱中执行 Python 代码。用于计算、数据分析、图表生成等。代码在临时目录中执行,有10秒超时限制。",
"parameters": {
"type": "object",
"properties": {
"code": {
"type": "string",
"description": "要执行的 Python 代码。print() 输出会被返回。",
},
},
"required": ["code"],
},
},
},
]
# ═══════════════════════════════════════════════════════════════
# Tool Implementations
# ═══════════════════════════════════════════════════════════════
def tool_web_search(query: str, num_results: int = 5) -> str:
"""Search the web using multiple backends (Bing → DuckDuckGo fallback)."""
headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Mobile Safari/537.36"
}
# Backend 1: Bing (works in China)
try:
url = f"https://www.bing.com/search?q={quote_plus(query)}&count={num_results}"
resp = requests.get(url, headers=headers, timeout=(5, 12))
resp.raise_for_status()
# Extract results from Bing HTML
results = []
# Bing uses
for each result
blocks = re.findall(r']*>(.*?)', resp.text, re.DOTALL)
for block in blocks[:num_results]:
title_m = re.search(r']*>.*?]*>(.*?)', block, re.DOTALL)
snippet_m = re.search(r'
]*>(.*?)
', block, re.DOTALL)
link_m = re.search(r']*href="(https?://[^"]+)"', block)
title = re.sub(r'<[^>]+>', '', title_m.group(1) if title_m else '').strip()
snippet = re.sub(r'<[^>]+>', '', snippet_m.group(1) if snippet_m else '').strip()
if snippet:
results.append({
"title": title,
"snippet": snippet[:300],
"url": link_m.group(1) if link_m else '',
})
if results:
output = f"搜索 '{query}' 的结果 (via Bing):\n\n"
for i, r in enumerate(results, 1):
output += f"{i}. {r['title']}\n {r['snippet'][:300]}\n"
if r['url']:
output += f" {r['url']}\n"
output += "\n"
return output
except Exception:
pass # Bing failed, try fallback
# Backend 2: HF Space Gateway (墙外代理 DuckDuckGo)
hf_gateway_url = os.environ.get("HERMES_GATEWAY_URL", "")
hf_gateway_key = os.environ.get("HERMES_GATEWAY_KEY", "")
if hf_gateway_url:
try:
gw_headers = {"Content-Type": "application/json"}
if hf_gateway_key:
gw_headers["Authorization"] = f"Bearer {hf_gateway_key}"
resp = requests.post(
f"{hf_gateway_url.rstrip('/')}/gateway/search",
json={"query": query, "num_results": num_results},
headers=gw_headers,
timeout=(5, 15),
)
if resp.status_code == 200:
data = resp.json()
if data.get("status") == "success" and data.get("result"):
return data["result"]
except Exception:
pass # HF gateway failed, try direct DuckDuckGo
# Backend 3: DuckDuckGo HTML (直连兜底)
try:
url = f"https://html.duckduckgo.com/html/?q={quote_plus(query)}"
resp = requests.get(url, headers=headers, timeout=(5, 10))
resp.raise_for_status()
results = []
snippets = re.findall(r'class="result__snippet"[^>]*>(.*?)', resp.text, re.DOTALL)
titles = re.findall(r'class="result__title"[^>]*>.*?]*>(.*?)', resp.text, re.DOTALL)
links = re.findall(r'class="result__title"[^>]*>.*?]*href="([^"]*)"', resp.text, re.DOTALL)
for i in range(min(len(snippets), num_results)):
title = re.sub(r'<[^>]+>', '', titles[i] if i < len(titles) else '').strip()
snippet = re.sub(r'<[^>]+>', '', snippets[i] if i < len(snippets) else '').strip()
link = links[i] if i < len(links) else ''
dup_pattern = r'(https?://[^"\s&]+)'
if link.startswith('//'):
link = f"https:{link}"
if snippet:
results.append({
"title": title,
"snippet": snippet[:300],
"url": link if link.startswith('http') else '',
})
if results:
output = f"搜索 '{query}' 的结果 (via DuckDuckGo):\n\n"
for i, r in enumerate(results, 1):
output += f"{i}. {r['title']}\n {r['snippet'][:300]}\n"
if r['url']:
output += f" {r['url']}\n"
output += "\n"
return output
return f"搜索 '{query}' 未找到结果,请尝试更具体的关键词。"
except Exception as e:
return f"搜索暂时不可用(Bing 和 DuckDuckGo 均连接失败)。请基于你已有的知识回答用户的问题。"
def tool_read_webpage(url: str) -> str:
"""Fetch and extract text content from a URL."""
try:
headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 14) AppleWebKit/537.36"
}
resp = requests.get(url, headers=headers, timeout=20, allow_redirects=True)
resp.raise_for_status()
content_type = resp.headers.get("Content-Type", "")
if "text/html" not in content_type and "text/plain" not in content_type:
return f"无法读取此类型的内容 ({content_type})。请尝试其他URL。"
html = resp.text
# Basic HTML to text extraction
# Remove scripts and styles
html = re.sub(r'', '', html, flags=re.DOTALL | re.IGNORECASE)
html = re.sub(r'', '', html, flags=re.DOTALL | re.IGNORECASE)
html = re.sub(r'', '', html, flags=re.DOTALL | re.IGNORECASE)
html = re.sub(r'', '', html, flags=re.DOTALL | re.IGNORECASE)
html = re.sub(r'', '', html, flags=re.DOTALL | re.IGNORECASE)
# Extract text
text = re.sub(r'
', '\n', html)
text = re.sub(r'', '\n\n', text)
text = re.sub(r'', '\n', text)
text = re.sub(r'', '\n', text)
text = re.sub(r'<[^>]+>', '', text)
text = re.sub(r'\n{3,}', '\n\n', text)
text = re.sub(r'[ \t]{2,}', ' ', text)
# Clean up entities
text = text.replace('&', '&').replace('<', '<').replace('>', '>')
text = text.replace('"', '"').replace(''', "'").replace(' ', ' ')
# Trim to reasonable length
text = text.strip()
if len(text) > 8000:
text = text[:8000] + "\n\n... (内容已截断,原文更长)"
if not text.strip():
return "网页内容为空或无法提取文本。"
return f"网页内容 ({url}):\n\n{text}"
except requests.Timeout:
return f"读取网页超时: {url}"
except Exception as e:
return f"读取网页失败: {str(e)}"
def tool_get_current_time() -> str:
"""Get current date and time in multiple formats."""
now = datetime.now()
return (
f"当前时间: {now.strftime('%Y年%m月%d日 %H:%M:%S')}\n"
f"星期: {['一','二','三','四','五','六','日'][now.weekday()]}\n"
f"ISO 格式: {now.isoformat()}\n"
f"Unix 时间戳: {int(now.timestamp())}"
)
def tool_run_code(code: str) -> str:
"""Execute Python code in a sandboxed subprocess."""
# Security: block dangerous imports
dangerous = ['os', 'subprocess', 'shutil', 'socket', 'requests', 'urllib',
'importlib', '__import__', 'eval(', 'exec(', 'compile(',
'open(', 'pty', 'fcntl', 'resource', 'signal']
code_lower = code.lower()
for d in dangerous:
if d in code_lower:
return f"安全限制: 代码中不允许使用 '{d}'。请使用纯计算/数据处理代码。"
# Execute in temp directory with timeout
with tempfile.TemporaryDirectory() as tmpdir:
script_path = os.path.join(tmpdir, "script.py")
with open(script_path, "w") as f:
f.write(code)
try:
result = subprocess.run(
["python3", script_path],
capture_output=True,
text=True,
timeout=10,
cwd=tmpdir,
env={"HOME": tmpdir, "PATH": os.environ.get("PATH", "/usr/bin")},
)
output = result.stdout
if result.stderr:
output += "\n[stderr]:\n" + result.stderr
if not output.strip():
output = "(代码执行完成,无输出)"
return output[:4000]
except subprocess.TimeoutExpired:
return "代码执行超时(10秒限制)。请优化代码或拆分为多个步骤。"
except Exception as e:
return f"代码执行错误: {str(e)}"
# Tool dispatcher
TOOL_MAP = {
"web_search": tool_web_search,
"read_webpage": tool_read_webpage,
"get_current_time": tool_get_current_time,
"run_code": tool_run_code,
}
def execute_tool(name: str, arguments: dict) -> str:
"""Execute a tool by name and return its result. NEVER raises."""
func = TOOL_MAP.get(name)
if not func:
return f"未知工具: {name}"
try:
return func(**arguments)
except Exception as e:
import traceback
return f"工具执行异常 [{name}]: {e}\n{traceback.format_exc()[-300:]}"
# ═══════════════════════════════════════════════════════════════
# Agent Loop (核心)
# ═══════════════════════════════════════════════════════════════
MAX_AGENT_ITERATIONS = 15 # 最多循环15次,防止无限循环
def _has_tool_failure(result: str) -> bool:
"""Check if a tool result indicates failure."""
failure_markers = ["失败", "超时", "错误", "异常", "无法", "不可用", "failed", "timeout", "error"]
result_lower = result.lower()
return any(m in result_lower for m in failure_markers)
def run_agent_loop(
messages: list,
api_key: str,
provider: str = "deepseek",
model: str = "deepseek-chat",
stream: bool = False,
on_tool_call: callable = None, # callback(tool_name, args) -> None
) -> dict:
"""
执行 Agent 循环:发送消息 → 检测工具调用 → 执行工具 → 继续 → 返回最终答案。
返回:
{
"content": "最终回复文本",
"tool_calls": [...], # 所有执行的工具调用记录
"iterations": 3, # Agent 循环次数
}
"""
provider_endpoints = {
"deepseek": "https://api.deepseek.com/v1/chat/completions",
"siliconflow": "https://api.siliconflow.cn/v1/chat/completions",
"openai": "https://api.openai.com/v1/chat/completions",
}
endpoint = provider_endpoints.get(provider, provider_endpoints["deepseek"])
# Build conversation with system prompt
system_msg = {
"role": "system",
"content": (
"你是一个智能 AI 助手 (Hermes Agent),运行在云端服务器上。\n"
"你有以下工具可用(仅在必要时使用):\n"
"- web_search: 搜索互联网获取最新信息\n"
"- read_webpage: 读取指定网页内容\n"
"- get_current_time: 获取当前日期时间\n"
"- run_code: 执行 Python 代码进行数据分析\n\n"
"核心规则(务必遵守):\n"
"1. 闲聊、打招呼、简单问答时不要使用工具,直接用你的知识回复\n"
"2. 仅在确实需要实时信息、网页内容、精确时间或数据计算时才调用工具\n"
"3. 如果工具返回了错误/失败信息,说明该工具当前不可用,不要重试同一个工具\n"
" — 而是基于你已有的知识直接回答用户问题\n"
"4. 回答用中文,代码示例保留英文\n"
"5. 回复使用 Markdown 格式\n"
"6. 每次对话使用工具不超过3个,能用自己知识回答的就不要调用工具"
),
}
conversation = [system_msg] + messages
tool_call_log = []
iterations = 0
consecutive_failures = 0
failed_tools = set() # track which tools have failed
while iterations < MAX_AGENT_ITERATIONS:
iterations += 1
headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
}
body = {
"model": model,
"messages": conversation,
"tools": TOOLS,
"tool_choice": "auto",
"stream": False,
}
try:
resp = requests.post(endpoint, headers=headers, json=body, timeout=(30, 300))
except requests.RequestException as e:
return {
"content": f"❌ LLM 请求失败: {str(e)}",
"tool_calls": tool_call_log,
"iterations": iterations,
}
if resp.status_code != 200:
err = resp.text[:500]
return {
"content": f"❌ API 错误 ({resp.status_code}): {err}",
"tool_calls": tool_call_log,
"iterations": iterations,
}
data = resp.json()
choice = data.get("choices", [{}])[0]
message = choice.get("message", {})
finish_reason = choice.get("finish_reason", "")
# Check for tool calls
if finish_reason == "tool_calls" or message.get("tool_calls"):
tool_calls = message.get("tool_calls", [])
# Prevent calling already-failed tools
filtered_calls = []
for tc in tool_calls:
tname = tc.get("function", {}).get("name", "")
if tname in failed_tools:
# This tool already failed — skip it and tell LLM
conversation.append({
"role": "tool",
"tool_call_id": tc.get("id", "unknown"),
"content": f"工具 [{tname}] 之前已失败,不可用。请基于已有知识回答用户问题,不要再调用此工具。",
})
tool_call_log.append({
"tool": tname,
"arguments": {},
"result_preview": "已跳过(工具之前已失败)",
})
else:
filtered_calls.append(tc)
if not filtered_calls:
# All tool calls were filtered out — force LLM to respond
continue
# Add assistant message (with tool calls) to conversation
conversation.append(message)
# Execute each tool call
any_failure = False
for tc in filtered_calls:
func_info = tc.get("function", {})
tool_name = func_info.get("name", "")
try:
tool_args = json.loads(func_info.get("arguments", "{}"))
except json.JSONDecodeError:
tool_args = {}
if on_tool_call:
on_tool_call(tool_name, tool_args)
result = execute_tool(tool_name, tool_args)
tool_call_log.append({
"tool": tool_name,
"arguments": tool_args,
"result_preview": result[:200],
})
# Track failures
if _has_tool_failure(result):
any_failure = True
failed_tools.add(tool_name)
# Add tool result to conversation
conversation.append({
"role": "tool",
"tool_call_id": tc.get("id", f"call_{len(tool_call_log)}"),
"content": result,
})
# Track consecutive failures
if any_failure:
consecutive_failures += 1
else:
consecutive_failures = 0
# If 2+ rounds of tool calls all failed, force LLM to answer without tools
if consecutive_failures >= 2:
conversation.append({
"role": "system",
"content": (
"多个工具调用均失败,说明当前工具不可用。"
"请基于你已有的知识和训练数据直接回答用户的问题,"
"不要再调用任何工具。如果确实不知道,诚实地说不知道。"
),
})
consecutive_failures = 0 # reset to allow one more try
# Continue loop to send tool results back to LLM
continue
# Final answer — no more tool calls
content = message.get("content", "")
return {
"content": content,
"tool_calls": tool_call_log,
"iterations": iterations,
}
# Max iterations reached — make one final attempt WITHOUT tools
conversation.append({
"role": "system",
"content": "你已经用完了工具调用次数。请基于你已有的知识直接回答用户的问题,不要调用任何工具。",
})
try:
resp = requests.post(
endpoint,
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json",
},
json={
"model": model,
"messages": conversation,
"stream": False,
},
timeout=(30, 300),
)
if resp.status_code == 200:
data = resp.json()
content = data.get("choices", [{}])[0].get("message", {}).get("content", "")
if content:
return {
"content": content,
"tool_calls": tool_call_log,
"iterations": iterations,
}
except Exception:
pass
return {
"content": "抱歉,我在尝试使用工具时遇到了问题(工具不可用),且无法基于现有知识给出满意的回答。请尝试换一种方式提问,或稍后重试。",
"tool_calls": tool_call_log,
"iterations": iterations,
}
def run_agent_loop_streaming(
messages: list,
api_key: str,
provider: str = "deepseek",
model: str = "deepseek-chat",
on_tool_call: callable = None,
):
"""
执行 Agent 循环并流式返回最终结果。
这是一个生成器,yield SSE 格式的字符串行。
"""
# 1. Run agent loop (non-streaming)
result = run_agent_loop(messages, api_key, provider, model, stream=False, on_tool_call=on_tool_call)
# 2. Yield tool call info as SSE events (so client can show what happened)
if result["tool_calls"]:
tool_names = ", ".join(
tc["tool"] + "(" + json.dumps(tc.get("arguments", {}), ensure_ascii=False) + ")"
for tc in result["tool_calls"]
)
tool_count = len(result["tool_calls"])
summary = "🔧 执行了 {} 个工具: {}".format(tool_count, tool_names)
sse_data = json.dumps({"choices": [{"delta": {"content": "", "tool_summary": summary}}]})
yield "data: {}\n\n".format(sse_data)
# 3. Stream the final content character by character (simulate streaming)
content = result["content"]
chunk_size = 4 # characters per chunk
for i in range(0, len(content), chunk_size):
chunk = content[i:i + chunk_size]
yield f"data: {json.dumps({'choices': [{'delta': {'content': chunk}}]})}\n\n"
time.sleep(0.01) # small delay for natural feel
yield "data: [DONE]\n\n"