高级功能
Xyzen 提供了一系列强大的高级功能,使你能够集成多种模型提供商、 管理用户消费记录,以及创建可重用的 Agent 框架。 本文档将详细介绍这些高级功能的使用方法。
1. 多模型提供商管理
概述
Xyzen 支持与多个 LLM 提供商集成,包括:
- OpenAI - GPT-4, GPT-4 Turbo, GPT-3.5 等
- Google Gemini - Gemini Pro 等
- Claude (Anthropic) - Claude 2, Claude Instant 等
- 本地模型 - 通过 Ollama 或其他本地部署
提供商配置
创建新提供商
POST /api/v1/providers
Content-Type: application/json
{
"name": "My OpenAI Account",
"type": "openai",
"config": {
"api_key": "sk-...",
"base_url": "https://api.openai.com/v1",
"model_list": ["gpt-4", "gpt-3.5-turbo"]
},
"is_default": true
}
支持的提供商配置
| 提供商 | 必填字段 | 可选字段 |
|---|---|---|
| OpenAI | api_key | base_url, organization_id |
| Google Gemini | api_key | base_url |
| Claude | api_key | base_url |
| 本地 (Ollama) | base_url | 无 |
动态模型选择
在 Agent 中动态选择模型:
# 服务端示例
from core.providers import get_user_provider_manager
async def execute_with_model_selection(user_id: str, query: str):
provider_manager = await get_user_provider_manager(user_id, db)
# 根据查询复杂度选择模型
if len(query) > 1000:
# 复杂查询使用高级模型
llm = provider_manager.get_llm("gpt-4")
else:
# 简单查询使用低成本模型
llm = provider_manager.get_llm("gpt-3.5-turbo")
response = await llm.apredict(query)
return response
成本优化
Xyzen 支持根据任务成本进行模型的自动选择:
# 按成本优化的 Agent 配置
agent_config = {
"name": "Cost-Aware Agent",
"routing_strategy": "cost_optimized",
"models": [
{
"model": "gpt-3.5-turbo",
"cost_per_1k_tokens": 0.0015,
"suitability": ["simple_qa", "summarization"]
},
{
"model": "gpt-4",
"cost_per_1k_tokens": 0.03,
"suitability": ["complex_reasoning", "code_generation"]
}
]
}
2. MCP(Model Context Protocol)服务器集成
概述
MCP 是一个开放协议,允许 AI 模型通过标准化的接口访问外部工具和数据源。Xyzen 原生支持 MCP 服务器集成。
注册 MCP 服务器
通过 API 注册
POST /api/v1/mcp-servers
Content-Type: application/json
{
"name": "Scientific Tools",
"description": "A collection of scientific computing tools",
"url": "http://localhost:3000/mcp",
"token": "optional-auth-token",
"tags": ["science", "computing"]
}
服务器健康检查
GET /api/v1/mcp-servers/{server_id}/health
Xyzen 会自动:
- 连接到 MCP 服务器
- 列出所有可用工具
- 验证工具的可访问性
- 定期检查服务器状态
在 Agent 中使用 MCP 工具
配置 Agent 使用 MCP
{
"name": "Research Assistant",
"agent_type": "regular",
"mcp_servers": [
{
"mcp_server_id": "server-uuid-1",
"enabled": true
},
{
"mcp_server_id": "server-uuid-2",
"enabled": true
}
]
}
MCP 工具调用流程
工具使用确认机制
为了提高安全性,Xyzen 支持在执行工具前进行确认:
POST /api/v1/chat/messages
Content-Type: application/json
{
"agent_id": "agent-uuid",
"content": "计算 123 + 456",
"require_tool_confirmation": true
}
响应(等待确认):
{
"message_id": "msg-uuid",
"status": "pending_tool_confirmation",
"pending_tools": [
{
"name": "add",
"description": "Add two numbers",
"arguments": {
"a": 123,
"b": 456
}
}
]
}
用户确认后:
POST /api/v1/chat/messages/{message_id}/confirm-tools
Content-Type: application/json
{
"confirmed": true
}
内置 MCP 服务器
Xyzen 预置了几个常用的 MCP 服务器:
| 服务器 | 功能 | 工具示例 |
|---|---|---|
| BioYond | 生物科学工具 | 蛋白质序列分析、基因搜索 |
| Lab | 实验室工具 | 数据处理、可视化 |
| Dify | 工作流集成 | 调用 Dify 工作流 |
3. 内置 Agent 系统
概述
Xyzen 提供了两个预定义的系统 Agent,可供所有用户使用:
随便聊聊 Assistants Agent
- 用途:通用对话和问题解答
- 特点:友好、有用、支持工具调用
- 适用场景:日常提问、信息查询、任务协助
创作工坊 Development Agent
- 用途:AI Agent 设计和优化
- 特点:专业、深入、提供架构建议
- 适用场景:Agent 设计、提示词工程、工作流优化
系统 Agent 配置
# 系统 Agent 配置示例
SYSTEM_AGENTS = {
"chat": {
"name": "随便聊聊",
"capabilities": ["general_chat", "qa", "assistance", "tools"],
"tags": ["助手", "对话", "工具"],
},
"workshop": {
"name": "创作工坊",
"capabilities": ["agent_design", "tool_selection", "prompt_engineering"],
"tags": ["设计", "创作", "优化"],
}
}
创建内置 Graph Agent
Xyzen 支持将 Python 类自动注册为内置 Graph Agent:
# 示例:创建科研论文分析 Agent
from handler.builtin_agents.base_graph_agent import BaseBuiltinGraphAgent
from langgraph.graph import StateGraph
class ScientificFigureAgent(BaseBuiltinGraphAgent):
"""分析科研论文中的图表的内置 Agent"""
def __init__(self):
super().__init__(
name="Scientific Figure Analyzer",
description="分析并解释科研论文中的图表、数据和可视化",
version="1.0.0",
capabilities=[
"figure_analysis",
"data_extraction",
"interpretation"
],
tags=["science", "analysis", "figures"],
author="Xyzen Team",
license_="Apache 2.0"
)
def build_graph(self) -> CompiledStateGraph:
"""构建工作流"""
workflow = StateGraph(GraphState)
# 添加节点
workflow.add_node("extract", self._extract_figure_data)
workflow.add_node("analyze", self._analyze_content)
workflow.add_node("generate_report", self._generate_report)
# 添加边
workflow.add_edge("extract", "analyze")
workflow.add_edge("analyze", "generate_report")
# 设置入出口
workflow.set_entry_point("extract")
workflow.set_finish_point("generate_report")
return workflow.compile()
def get_state_schema(self) -> dict:
"""返回状态模式"""
return {
"image": "bytes",
"extracted_data": "dict",
"analysis": "str",
"report": "str"
}
async def _extract_figure_data(self, state: dict) -> dict:
"""提取图表数据"""
# 实现数据提取逻辑
return {
**state,
"extracted_data": {...}
}
async def _analyze_content(self, state: dict) -> dict:
"""分析内容"""
# 实现分析逻辑
return {
**state,
"analysis": "..."
}
async def _generate_report(self, state: dict) -> dict:
"""生成报告"""
# 实现报告生成逻辑
return {
**state,
"report": "..."
}
内置 Agent 会自动注册到系统中,用户可以直接使用。