492 lines
17 KiB
Python
492 lines
17 KiB
Python
"""
|
||
ACP Gateway - Agent Communication Protocol 内部实现
|
||
Phase G3: 多 Agent 协作编排层
|
||
|
||
将平台已有的 Agent（MarketApp type='agent'）封装为可编排调用的"Agent 节点"。
|
||
每个 Agent 节点执行时，通过本 Gateway 创建会话、发送提示、接收流式响应。
|
||
支持通过 Redis EventBus 发布节点进度通知，供 WebSocket 实时推送到前端。
|
||
"""
|
||
import json
|
||
import uuid
|
||
import time
|
||
import logging
|
||
import asyncio
|
||
import threading
|
||
from typing import Optional, Dict, Any, Callable, AsyncIterator
|
||
from dataclasses import dataclass, field
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ============================================================
|
||
# Data Models
|
||
# ============================================================
|
||
|
||
@dataclass
|
||
class AcpSessionState:
|
||
"""ACP 会话状态"""
|
||
session_id: str
|
||
agent_id: str # MarketApp id
|
||
agent_name: str = ""
|
||
status: str = "idle" # idle / running / completed / failed / cancelled
|
||
prompt: str = ""
|
||
created_at: float = field(default_factory=time.time)
|
||
completed_at: Optional[float] = None
|
||
output: str = ""
|
||
error: Optional[str] = None
|
||
tool_calls: list = field(default_factory=list)
|
||
chain_of_thought: list = field(default_factory=list) # 思考过程
|
||
|
||
def to_dict(self) -> dict:
|
||
return {
|
||
"session_id": self.session_id,
|
||
"agent_id": self.agent_id,
|
||
"agent_name": self.agent_name,
|
||
"status": self.status,
|
||
"prompt": self.prompt[:200],
|
||
"created_at": self.created_at,
|
||
"completed_at": self.completed_at,
|
||
"output": self.output[:500] if self.output else "",
|
||
"error": self.error,
|
||
"tool_calls": self.tool_calls,
|
||
"chain_of_thought": self.chain_of_thought,
|
||
}
|
||
|
||
|
||
@dataclass
|
||
class AgentConfig:
|
||
"""从 MarketApp.skill_content 解析出的 Agent 配置"""
|
||
agent_id: str = ""
|
||
agent_name: str = ""
|
||
system_prompt: str = ""
|
||
instructions: str = ""
|
||
model_id: Optional[str] = None
|
||
temperature: float = 0.7
|
||
max_tokens: int = 4096
|
||
skills: list = field(default_factory=list)
|
||
knowledge_base_ids: list = field(default_factory=list)
|
||
tools: list = field(default_factory=list)
|
||
|
||
|
||
# ============================================================
|
||
# Agent Resolver
|
||
# ============================================================
|
||
|
||
def resolve_agent_config(db, agent_id: str) -> Optional[AgentConfig]:
|
||
"""从 MarketApp 解析 Agent 配置"""
|
||
from app.models import MarketApp
|
||
|
||
agent = db.query(MarketApp).filter(
|
||
MarketApp.id == agent_id,
|
||
MarketApp.type == "agent"
|
||
).first()
|
||
|
||
if not agent:
|
||
return None
|
||
|
||
config_json = {}
|
||
if agent.skill_content:
|
||
try:
|
||
config_json = json.loads(agent.skill_content) if isinstance(agent.skill_content, str) else agent.skill_content
|
||
except Exception:
|
||
pass
|
||
|
||
return AgentConfig(
|
||
agent_id=agent.id,
|
||
agent_name=agent.name,
|
||
system_prompt=config_json.get("system_prompt", config_json.get("role", agent.description or "AI助手")),
|
||
instructions=config_json.get("instructions", ""),
|
||
model_id=config_json.get("model_id"),
|
||
temperature=float(config_json.get("temperature", 0.7)),
|
||
max_tokens=int(config_json.get("max_tokens", 4096)),
|
||
skills=config_json.get("skills", []),
|
||
knowledge_base_ids=config_json.get("knowledge_base_ids", []),
|
||
tools=config_json.get("tools", []),
|
||
)
|
||
|
||
|
||
def resolve_model_config(db, model_id: str) -> dict:
|
||
"""获取模型的 API 配置"""
|
||
from app.models import AIModel
|
||
|
||
model = db.query(AIModel).filter(AIModel.id == model_id).first()
|
||
if not model:
|
||
# Fallback to default model
|
||
default_model = db.query(AIModel).filter(AIModel.is_enabled == True).first()
|
||
if default_model:
|
||
model = default_model
|
||
else:
|
||
return {
|
||
"api_key": "",
|
||
"api_base": "https://api.siliconflow.cn/v1",
|
||
"model_name": "deepseek-ai/DeepSeek-V4-Flash",
|
||
}
|
||
|
||
return {
|
||
"api_key": model.api_key or "",
|
||
"api_base": model.api_base or "https://api.siliconflow.cn/v1",
|
||
"model_name": model.model_id or model.name,
|
||
}
|
||
|
||
|
||
# ============================================================
|
||
# ACP Gateway
|
||
# ============================================================
|
||
|
||
class AcpGateway:
|
||
"""Agent 通信网关 — 平台内部多 Agent 编排核心"""
|
||
|
||
def __init__(self):
|
||
self._sessions: Dict[str, AcpSessionState] = {}
|
||
self._lock = threading.Lock()
|
||
self._on_notification: Optional[Callable] = None
|
||
|
||
def set_notification_handler(self, handler: Callable[[dict], None]):
|
||
"""设置通知回调 — 用于发布到 Redis EventBus"""
|
||
self._on_notification = handler
|
||
|
||
def _notify(self, event_type: str, data: dict):
|
||
"""发送通知"""
|
||
if self._on_notification:
|
||
try:
|
||
self._on_notification({
|
||
"type": event_type,
|
||
"timestamp": time.time(),
|
||
**data,
|
||
})
|
||
except Exception as e:
|
||
logger.error(f"[AcpGateway] 通知回调异常: {e}")
|
||
|
||
# ---- Session Management ----
|
||
|
||
def create_session(self, agent_id: str, agent_name: str = "") -> str:
|
||
"""创建新的 Agent 会话"""
|
||
session_id = f"acp-{uuid.uuid4().hex[:12]}"
|
||
with self._lock:
|
||
self._sessions[session_id] = AcpSessionState(
|
||
session_id=session_id,
|
||
agent_id=agent_id,
|
||
agent_name=agent_name or agent_id,
|
||
status="idle",
|
||
)
|
||
logger.info(f"[AcpGateway] 创建会话: {session_id} (agent={agent_name or agent_id})")
|
||
return session_id
|
||
|
||
def get_session(self, session_id: str) -> Optional[AcpSessionState]:
|
||
"""获取会话"""
|
||
with self._lock:
|
||
return self._sessions.get(session_id)
|
||
|
||
def cancel_session(self, session_id: str) -> bool:
|
||
"""取消会话"""
|
||
with self._lock:
|
||
sess = self._sessions.get(session_id)
|
||
if sess and sess.status == "running":
|
||
sess.status = "cancelled"
|
||
sess.completed_at = time.time()
|
||
self._notify("agent_cancelled", sess.to_dict())
|
||
return True
|
||
return False
|
||
|
||
# ---- Agent Invocation ----
|
||
|
||
async def invoke_agent(
|
||
self,
|
||
agent_id: str,
|
||
prompt: str,
|
||
system_prompt: str = "",
|
||
model_id: Optional[str] = None,
|
||
temperature: float = 0.7,
|
||
max_tokens: int = 4096,
|
||
context: Optional[dict] = None,
|
||
db_session=None,
|
||
) -> dict:
|
||
"""调用 Agent — 创建会话、发送提示、等待完成
|
||
|
||
这是一个高级封装，内部自动创建和管理会话生命周期。
|
||
返回完整结果字典，包含 output, tool_calls, chain_of_thought 等。
|
||
"""
|
||
import httpx
|
||
|
||
# Resolve agent config from DB
|
||
agent_name = agent_id
|
||
if db_session:
|
||
config = resolve_agent_config(db_session, agent_id)
|
||
if config:
|
||
agent_name = config.agent_name
|
||
system_prompt = system_prompt or config.system_prompt
|
||
model_id = model_id or config.model_id
|
||
temperature = config.temperature
|
||
max_tokens = config.max_tokens
|
||
|
||
# Create session
|
||
session_id = self.create_session(agent_id, agent_name)
|
||
sess = self.get_session(session_id)
|
||
|
||
# Resolve model
|
||
if db_session and model_id:
|
||
model_cfg = resolve_model_config(db_session, model_id)
|
||
else:
|
||
model_cfg = {"api_key": "", "api_base": "https://api.siliconflow.cn/v1", "model_name": "deepseek-ai/DeepSeek-V4-Flash"}
|
||
|
||
# Update status
|
||
with self._lock:
|
||
sess.status = "running"
|
||
sess.prompt = prompt
|
||
|
||
self._notify("agent_running", {"session_id": session_id, "agent_name": agent_name, "prompt": prompt[:200]})
|
||
|
||
try:
|
||
# Build messages
|
||
messages = []
|
||
if system_prompt:
|
||
messages.append({"role": "system", "content": system_prompt})
|
||
messages.append({"role": "user", "content": prompt})
|
||
|
||
# Call LLM
|
||
headers = {
|
||
"Authorization": f"Bearer {model_cfg['api_key']}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
payload = {
|
||
"model": model_cfg["model_name"],
|
||
"messages": messages,
|
||
"temperature": temperature,
|
||
"max_tokens": max_tokens,
|
||
"stream": False,
|
||
}
|
||
|
||
async with httpx.AsyncClient(timeout=120.0) as client:
|
||
resp = await client.post(
|
||
f"{model_cfg['api_base']}/chat/completions",
|
||
headers=headers,
|
||
json=payload,
|
||
)
|
||
resp.raise_for_status()
|
||
data = resp.json()
|
||
|
||
output = data["choices"][0]["message"]["content"]
|
||
usage = data.get("usage", {})
|
||
|
||
with self._lock:
|
||
sess.status = "completed"
|
||
sess.output = output
|
||
sess.completed_at = time.time()
|
||
|
||
result = {
|
||
"session_id": session_id,
|
||
"status": "completed",
|
||
"output": output,
|
||
"agent_name": agent_name,
|
||
"usage": usage,
|
||
"tool_calls": sess.tool_calls,
|
||
"chain_of_thought": sess.chain_of_thought,
|
||
}
|
||
self._notify("agent_completed", {"session_id": session_id, "output_preview": output[:200]})
|
||
return result
|
||
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
with self._lock:
|
||
sess.status = "failed"
|
||
sess.error = error_msg
|
||
sess.completed_at = time.time()
|
||
|
||
result = {
|
||
"session_id": session_id,
|
||
"status": "failed",
|
||
"output": f"[Agent] 调用失败: {error_msg}",
|
||
"agent_name": agent_name,
|
||
"error": error_msg,
|
||
}
|
||
self._notify("agent_failed", {"session_id": session_id, "error": error_msg})
|
||
return result
|
||
|
||
async def invoke_agent_streaming(
|
||
self,
|
||
agent_id: str,
|
||
prompt: str,
|
||
system_prompt: str = "",
|
||
model_id: Optional[str] = None,
|
||
temperature: float = 0.7,
|
||
max_tokens: int = 4096,
|
||
context: Optional[dict] = None,
|
||
db_session=None,
|
||
) -> AsyncIterator[dict]:
|
||
"""流式调用 Agent — 逐步返回 LLM 生成的 token chunks
|
||
|
||
每个 yield 的 dict 格式:
|
||
{"type": "token"|"done"|"error", "content": str, ...}
|
||
"""
|
||
import httpx
|
||
|
||
# Resolve agent config
|
||
agent_name = agent_id
|
||
if db_session:
|
||
config = resolve_agent_config(db_session, agent_id)
|
||
if config:
|
||
agent_name = config.agent_name
|
||
system_prompt = system_prompt or config.system_prompt
|
||
model_id = model_id or config.model_id
|
||
temperature = config.temperature
|
||
max_tokens = config.max_tokens
|
||
|
||
session_id = self.create_session(agent_id, agent_name)
|
||
sess = self.get_session(session_id)
|
||
|
||
if db_session and model_id:
|
||
model_cfg = resolve_model_config(db_session, model_id)
|
||
else:
|
||
model_cfg = {"api_key": "", "api_base": "https://api.siliconflow.cn/v1", "model_name": "deepseek-ai/DeepSeek-V4-Flash"}
|
||
|
||
with self._lock:
|
||
sess.status = "running"
|
||
sess.prompt = prompt
|
||
|
||
self._notify("agent_running", {"session_id": session_id, "agent_name": agent_name})
|
||
|
||
try:
|
||
messages = []
|
||
if system_prompt:
|
||
messages.append({"role": "system", "content": system_prompt})
|
||
messages.append({"role": "user", "content": prompt})
|
||
|
||
headers = {
|
||
"Authorization": f"Bearer {model_cfg['api_key']}",
|
||
"Content-Type": "application/json",
|
||
}
|
||
payload = {
|
||
"model": model_cfg["model_name"],
|
||
"messages": messages,
|
||
"temperature": temperature,
|
||
"max_tokens": max_tokens,
|
||
"stream": True,
|
||
}
|
||
|
||
collected = []
|
||
async with httpx.AsyncClient(timeout=120.0) as client:
|
||
async with client.stream("POST", f"{model_cfg['api_base']}/chat/completions",
|
||
headers=headers, json=payload) as resp:
|
||
resp.raise_for_status()
|
||
async for line in resp.aiter_lines():
|
||
if line.startswith("data: "):
|
||
data_str = line[6:]
|
||
if data_str.strip() == "[DONE]":
|
||
break
|
||
try:
|
||
chunk = json.loads(data_str)
|
||
delta = chunk.get("choices", [{}])[0].get("delta", {})
|
||
content = delta.get("content", "")
|
||
if content:
|
||
collected.append(content)
|
||
yield {"type": "token", "content": content, "session_id": session_id}
|
||
# Notify periodically (every ~20 tokens)
|
||
if len(collected) % 20 == 0:
|
||
self._notify("agent_token", {
|
||
"session_id": session_id,
|
||
"partial": "".join(collected[-20:]),
|
||
})
|
||
except json.JSONDecodeError:
|
||
continue
|
||
|
||
full_output = "".join(collected)
|
||
with self._lock:
|
||
sess.status = "completed"
|
||
sess.output = full_output
|
||
sess.completed_at = time.time()
|
||
|
||
self._notify("agent_completed", {"session_id": session_id, "output_preview": full_output[:200]})
|
||
yield {"type": "done", "session_id": session_id, "output": full_output, "status": "completed"}
|
||
|
||
except Exception as e:
|
||
error_msg = str(e)
|
||
with self._lock:
|
||
sess.status = "failed"
|
||
sess.error = error_msg
|
||
sess.completed_at = time.time()
|
||
|
||
self._notify("agent_failed", {"session_id": session_id, "error": error_msg})
|
||
yield {"type": "error", "session_id": session_id, "error": error_msg, "status": "failed"}
|
||
|
||
# ---- Multi-Agent Parallel Execution ----
|
||
|
||
async def invoke_agents_parallel(
|
||
self,
|
||
tasks: list,
|
||
db_session=None,
|
||
) -> list:
|
||
"""并行调用多个 Agent
|
||
|
||
tasks = [
|
||
{"agent_id": "xxx", "prompt": "...", "system_prompt": "..."},
|
||
{"agent_id": "yyy", "prompt": "...", "system_prompt": "..."},
|
||
]
|
||
|
||
Returns list of results in same order.
|
||
"""
|
||
async def _invoke_one(task: dict) -> dict:
|
||
try:
|
||
return await self.invoke_agent(
|
||
agent_id=task["agent_id"],
|
||
prompt=task.get("prompt", ""),
|
||
system_prompt=task.get("system_prompt", ""),
|
||
model_id=task.get("model_id"),
|
||
temperature=task.get("temperature", 0.7),
|
||
max_tokens=task.get("max_tokens", 4096),
|
||
context=task.get("context"),
|
||
db_session=db_session,
|
||
)
|
||
except Exception as e:
|
||
return {
|
||
"status": "failed",
|
||
"output": f"[Agent {task.get('agent_id', 'unknown')}] 执行异常: {str(e)}",
|
||
"error": str(e),
|
||
}
|
||
|
||
return await asyncio.gather(*[_invoke_one(t) for t in tasks])
|
||
|
||
|
||
# ============================================================
|
||
# Global Singleton
|
||
# ============================================================
|
||
|
||
_acp_gateway_instance: Optional[AcpGateway] = None
|
||
_lock_gateway = threading.Lock()
|
||
|
||
|
||
def get_acp_gateway() -> AcpGateway:
|
||
"""获取全局 ACP Gateway 单例"""
|
||
global _acp_gateway_instance
|
||
if _acp_gateway_instance is None:
|
||
with _lock_gateway:
|
||
if _acp_gateway_instance is None:
|
||
_acp_gateway_instance = AcpGateway()
|
||
return _acp_gateway_instance
|
||
|
||
|
||
# ============================================================
|
||
# Redis Notification Helper
|
||
# ============================================================
|
||
|
||
def publish_workflow_event(workflow_id: str, node_id: str, event: dict):
|
||
"""发布工作流执行事件到 Redis EventBus"""
|
||
try:
|
||
from app.event_bus import get_event_bus, MessageEvent
|
||
|
||
bus = get_event_bus()
|
||
msg = MessageEvent(
|
||
platform="workflow",
|
||
channel_id=f"{workflow_id}:{node_id}",
|
||
content=json.dumps(event, ensure_ascii=False),
|
||
content_type="workflow_event",
|
||
raw={
|
||
"workflow_id": workflow_id,
|
||
"node_id": node_id,
|
||
**event,
|
||
},
|
||
)
|
||
bus.publish(msg)
|
||
return True
|
||
except Exception as e:
|
||
logger.error(f"[AcpGateway] Redis 发布失败: {e}")
|
||
return False
|