chore: archive development research artifacts and debug scripts
This commit is contained in:
fujie
@@ -0,0 +1,509 @@
|
||||
# OpenWebUI Function 集成方案
|
||||
|
||||
## 🎯 核心挑战
|
||||
|
||||
在 Copilot Tool Handler 中调用 OpenWebUI Functions 的关键问题：
|
||||
|
||||
**问题：** Copilot SDK 的 Tool Handler 是一个独立的回调函数，如何在这个上下文中访问和执行 OpenWebUI 的 Function？
|
||||
|
||||
---
|
||||
|
||||
## 🔍 OpenWebUI Function 系统分析
|
||||
|
||||
### 1. Function 数据结构
|
||||
|
||||
OpenWebUI 的 Function/Tool 传递格式：
|
||||
|
||||
```python
|
||||
body = {
|
||||
"tools": [
|
||||
{
|
||||
"type": "function",
|
||||
"function": {
|
||||
"name": "get_weather",
|
||||
"description": "Get current weather",
|
||||
"parameters": {
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"location": {"type": "string"}
|
||||
},
|
||||
"required": ["location"]
|
||||
}
|
||||
}
|
||||
}
|
||||
]
|
||||
}
|
||||
```
|
||||
|
||||
### 2. Function 执行机制
|
||||
|
||||
OpenWebUI Functions 的执行方式有几种可能：
|
||||
|
||||
#### 选项 A: 通过 Function ID 调用内部 API
|
||||
|
||||
```python
|
||||
# 假设 OpenWebUI 提供内部 API
|
||||
from open_webui.apps.webui.models.functions import Functions
|
||||
|
||||
function_id = "function_uuid" # 需要从配置中获取
|
||||
result = await Functions.execute_function(
|
||||
function_id=function_id,
|
||||
arguments={"location": "Beijing"}
|
||||
)
|
||||
```
|
||||
|
||||
#### 选项 B: 通过 **event_emitter** 触发
|
||||
|
||||
```python
|
||||
# 通过事件系统触发 function 执行
|
||||
if __event_emitter__:
|
||||
await __event_emitter__({
|
||||
"type": "function_call",
|
||||
"data": {
|
||||
"name": "get_weather",
|
||||
"arguments": {"location": "Beijing"}
|
||||
}
|
||||
})
|
||||
```
|
||||
|
||||
#### 选项 C: 自己实现 Function 逻辑
|
||||
|
||||
```python
|
||||
# 在 Pipe 内部实现常用功能
|
||||
class Pipe:
|
||||
def _builtin_get_weather(self, location: str) -> dict:
|
||||
# 实现天气查询
|
||||
pass
|
||||
|
||||
def _builtin_search_web(self, query: str) -> dict:
|
||||
# 实现网页搜索
|
||||
pass
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 💡 推荐方案：混合架构
|
||||
|
||||
### 架构设计
|
||||
|
||||
```
|
||||
User Message
|
||||
↓
|
||||
OpenWebUI UI (Functions 已配置)
|
||||
↓
|
||||
Pipe.pipe(body) - body 包含 tools[]
|
||||
↓
|
||||
转换为 Copilot Tools + 存储 Function Registry
|
||||
↓
|
||||
Copilot 决定调用 Tool
|
||||
↓
|
||||
Tool Handler 查询 Registry → 执行对应逻辑
|
||||
↓
|
||||
返回结果给 Copilot
|
||||
↓
|
||||
继续生成回答
|
||||
```
|
||||
|
||||
### 核心实现
|
||||
|
||||
#### 1. Function Registry（函数注册表）
|
||||
|
||||
```python
|
||||
class Pipe:
|
||||
def __init__(self):
|
||||
# ...
|
||||
self._function_registry = {} # {function_name: callable}
|
||||
self._function_metadata = {} # {function_name: metadata}
|
||||
```
|
||||
|
||||
#### 2. 注册 Functions
|
||||
|
||||
```python
|
||||
def _register_openwebui_functions(
|
||||
self,
|
||||
owui_functions: List[dict],
|
||||
__event_emitter__=None,
|
||||
__event_call__=None
|
||||
):
|
||||
"""
|
||||
注册 OpenWebUI Functions 到内部 registry
|
||||
|
||||
关键：将 function 定义和执行逻辑关联起来
|
||||
"""
|
||||
for func_def in owui_functions:
|
||||
if func_def.get("type") != "function":
|
||||
continue
|
||||
|
||||
func_info = func_def.get("function", {})
|
||||
func_name = func_info.get("name")
|
||||
|
||||
if not func_name:
|
||||
continue
|
||||
|
||||
# 存储元数据
|
||||
self._function_metadata[func_name] = {
|
||||
"description": func_info.get("description", ""),
|
||||
"parameters": func_info.get("parameters", {}),
|
||||
"original_def": func_def
|
||||
}
|
||||
|
||||
# 创建执行器（关键）
|
||||
executor = self._create_function_executor(
|
||||
func_name,
|
||||
func_def,
|
||||
__event_emitter__,
|
||||
__event_call__
|
||||
)
|
||||
|
||||
self._function_registry[func_name] = executor
|
||||
```
|
||||
|
||||
#### 3. Function Executor 工厂
|
||||
|
||||
```python
|
||||
def _create_function_executor(
|
||||
self,
|
||||
func_name: str,
|
||||
func_def: dict,
|
||||
__event_emitter__=None,
|
||||
__event_call__=None
|
||||
):
|
||||
"""
|
||||
为每个 function 创建执行器
|
||||
|
||||
策略：
|
||||
1. 优先使用内置实现
|
||||
2. 尝试调用 OpenWebUI API
|
||||
3. 返回错误
|
||||
"""
|
||||
|
||||
async def executor(arguments: dict) -> dict:
|
||||
# 策略 1: 检查是否有内置实现
|
||||
builtin_method = getattr(self, f"_builtin_{func_name}", None)
|
||||
if builtin_method:
|
||||
self._emit_debug_log_sync(
|
||||
f"Using builtin implementation for {func_name}",
|
||||
__event_call__
|
||||
)
|
||||
try:
|
||||
result = builtin_method(arguments)
|
||||
if inspect.iscoroutine(result):
|
||||
result = await result
|
||||
return {"success": True, "result": result}
|
||||
except Exception as e:
|
||||
return {"success": False, "error": str(e)}
|
||||
|
||||
# 策略 2: 尝试通过 Event Emitter 调用
|
||||
if __event_emitter__:
|
||||
try:
|
||||
# 尝试触发 function_call 事件
|
||||
response_queue = asyncio.Queue()
|
||||
|
||||
await __event_emitter__({
|
||||
"type": "function_call",
|
||||
"data": {
|
||||
"name": func_name,
|
||||
"arguments": arguments,
|
||||
"response_queue": response_queue # 回调队列
|
||||
}
|
||||
})
|
||||
|
||||
# 等待结果（带超时）
|
||||
result = await asyncio.wait_for(
|
||||
response_queue.get(),
|
||||
timeout=self.valves.TOOL_TIMEOUT
|
||||
)
|
||||
|
||||
return {"success": True, "result": result}
|
||||
except asyncio.TimeoutError:
|
||||
return {"success": False, "error": "Function execution timeout"}
|
||||
except Exception as e:
|
||||
self._emit_debug_log_sync(
|
||||
f"Event emitter call failed: {e}",
|
||||
__event_call__
|
||||
)
|
||||
# 继续尝试其他方法
|
||||
|
||||
# 策略 3: 尝试调用 OpenWebUI internal API
|
||||
try:
|
||||
# 这需要研究 OpenWebUI 源码确定正确的调用方式
|
||||
from open_webui.apps.webui.models.functions import Functions
|
||||
|
||||
# 需要获取 function_id（这是关键问题）
|
||||
function_id = self._get_function_id_by_name(func_name)
|
||||
|
||||
if function_id:
|
||||
result = await Functions.execute(
|
||||
function_id=function_id,
|
||||
params=arguments
|
||||
)
|
||||
return {"success": True, "result": result}
|
||||
except ImportError:
|
||||
pass
|
||||
except Exception as e:
|
||||
self._emit_debug_log_sync(
|
||||
f"OpenWebUI API call failed: {e}",
|
||||
__event_call__
|
||||
)
|
||||
|
||||
# 策略 4: 返回"未实现"错误
|
||||
return {
|
||||
"success": False,
|
||||
"error": f"Function '{func_name}' is not implemented. "
|
||||
"Please implement it as a builtin method or ensure OpenWebUI API is available."
|
||||
}
|
||||
|
||||
return executor
|
||||
```
|
||||
|
||||
#### 4. Tool Handler 实现
|
||||
|
||||
```python
|
||||
def _create_tool_handler(self, tool_name: str, __event_call__=None):
|
||||
"""为 Copilot SDK 创建 Tool Handler"""
|
||||
|
||||
async def handler(invocation: dict) -> dict:
|
||||
"""
|
||||
Copilot Tool Handler
|
||||
|
||||
invocation: {
|
||||
"session_id": str,
|
||||
"tool_call_id": str,
|
||||
"tool_name": str,
|
||||
"arguments": dict
|
||||
}
|
||||
"""
|
||||
try:
|
||||
# 从 registry 获取 executor
|
||||
executor = self._function_registry.get(invocation["tool_name"])
|
||||
|
||||
if not executor:
|
||||
return {
|
||||
"textResultForLlm": f"Function '{invocation['tool_name']}' not found.",
|
||||
"resultType": "failure",
|
||||
"error": "function_not_found",
|
||||
"toolTelemetry": {}
|
||||
}
|
||||
|
||||
# 执行 function
|
||||
self._emit_debug_log_sync(
|
||||
f"Executing function: {invocation['tool_name']}({invocation['arguments']})",
|
||||
__event_call__
|
||||
)
|
||||
|
||||
exec_result = await executor(invocation["arguments"])
|
||||
|
||||
# 处理结果
|
||||
if exec_result.get("success"):
|
||||
result_text = str(exec_result.get("result", ""))
|
||||
return {
|
||||
"textResultForLlm": result_text,
|
||||
"resultType": "success",
|
||||
"error": None,
|
||||
"toolTelemetry": {}
|
||||
}
|
||||
else:
|
||||
error_msg = exec_result.get("error", "Unknown error")
|
||||
return {
|
||||
"textResultForLlm": f"Function execution failed: {error_msg}",
|
||||
"resultType": "failure",
|
||||
"error": error_msg,
|
||||
"toolTelemetry": {}
|
||||
}
|
||||
|
||||
except Exception as e:
|
||||
self._emit_debug_log_sync(
|
||||
f"Tool handler error: {e}",
|
||||
__event_call__
|
||||
)
|
||||
return {
|
||||
"textResultForLlm": "An unexpected error occurred during function execution.",
|
||||
"resultType": "failure",
|
||||
"error": str(e),
|
||||
"toolTelemetry": {}
|
||||
}
|
||||
|
||||
return handler
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🔌 内置 Functions 实现示例
|
||||
|
||||
### 示例 1: 获取当前时间
|
||||
|
||||
```python
|
||||
def _builtin_get_current_time(self, arguments: dict) -> str:
|
||||
"""内置实现：获取当前时间"""
|
||||
from datetime import datetime
|
||||
|
||||
timezone = arguments.get("timezone", "UTC")
|
||||
format_str = arguments.get("format", "%Y-%m-%d %H:%M:%S")
|
||||
|
||||
now = datetime.now()
|
||||
return now.strftime(format_str)
|
||||
```
|
||||
|
||||
### 示例 2: 简单计算器
|
||||
|
||||
```python
|
||||
def _builtin_calculate(self, arguments: dict) -> str:
|
||||
"""内置实现：数学计算"""
|
||||
expression = arguments.get("expression", "")
|
||||
|
||||
try:
|
||||
# 安全的数学计算（仅允许基本运算）
|
||||
allowed_chars = set("0123456789+-*/()., ")
|
||||
if not all(c in allowed_chars for c in expression):
|
||||
raise ValueError("Invalid characters in expression")
|
||||
|
||||
result = eval(expression, {"__builtins__": {}})
|
||||
return str(result)
|
||||
except Exception as e:
|
||||
raise ValueError(f"Calculation error: {e}")
|
||||
```
|
||||
|
||||
### 示例 3: 网页搜索（需要外部 API）
|
||||
|
||||
```python
|
||||
async def _builtin_search_web(self, arguments: dict) -> str:
|
||||
"""内置实现：网页搜索（使用 DuckDuckGo）"""
|
||||
query = arguments.get("query", "")
|
||||
max_results = arguments.get("max_results", 5)
|
||||
|
||||
try:
|
||||
# 使用 duckduckgo_search 库
|
||||
from duckduckgo_search import DDGS
|
||||
|
||||
results = []
|
||||
with DDGS() as ddgs:
|
||||
for r in ddgs.text(query, max_results=max_results):
|
||||
results.append({
|
||||
"title": r.get("title", ""),
|
||||
"url": r.get("href", ""),
|
||||
"snippet": r.get("body", "")
|
||||
})
|
||||
|
||||
# 格式化结果
|
||||
formatted = "\n\n".join([
|
||||
f"**{r['title']}**\n{r['url']}\n{r['snippet']}"
|
||||
for r in results
|
||||
])
|
||||
|
||||
return formatted
|
||||
except Exception as e:
|
||||
raise ValueError(f"Search failed: {e}")
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## 🚀 完整集成流程
|
||||
|
||||
### pipe() 方法中的集成
|
||||
|
||||
```python
|
||||
async def pipe(
|
||||
self,
|
||||
body: dict,
|
||||
__metadata__: Optional[dict] = None,
|
||||
__event_emitter__=None,
|
||||
__event_call__=None,
|
||||
) -> Union[str, AsyncGenerator]:
|
||||
# ... 现有代码 ...
|
||||
|
||||
# ✅ Step 1: 提取 OpenWebUI Functions
|
||||
owui_functions = body.get("tools", [])
|
||||
|
||||
# ✅ Step 2: 注册 Functions
|
||||
if self.valves.ENABLE_TOOLS and owui_functions:
|
||||
self._register_openwebui_functions(
|
||||
owui_functions,
|
||||
__event_emitter__,
|
||||
__event_call__
|
||||
)
|
||||
|
||||
# ✅ Step 3: 转换为 Copilot Tools
|
||||
copilot_tools = []
|
||||
for func_name in self._function_registry.keys():
|
||||
metadata = self._function_metadata[func_name]
|
||||
copilot_tools.append({
|
||||
"name": func_name,
|
||||
"description": metadata["description"],
|
||||
"parameters": metadata["parameters"],
|
||||
"handler": self._create_tool_handler(func_name, __event_call__)
|
||||
})
|
||||
|
||||
# ✅ Step 4: 创建 Session 并传递 Tools
|
||||
session_config = SessionConfig(
|
||||
model=real_model_id,
|
||||
tools=copilot_tools, # ✅ 关键
|
||||
...
|
||||
)
|
||||
|
||||
session = await client.create_session(config=session_config)
|
||||
|
||||
# ... 后续代码 ...
|
||||
```
|
||||
|
||||
---
|
||||
|
||||
## ⚠️ 待解决问题
|
||||
|
||||
### 1. Function ID 映射
|
||||
|
||||
**问题：** OpenWebUI Functions 通常通过 UUID 标识，但 body 中只有 name
|
||||
|
||||
**解决思路：**
|
||||
|
||||
- 在 OpenWebUI 启动时建立 name → id 映射表
|
||||
- 或者修改 OpenWebUI 在 body 中同时传递 id
|
||||
|
||||
### 2. Event Emitter 回调机制
|
||||
|
||||
**问题：** 不确定 **event_emitter** 是否支持 function_call 事件
|
||||
|
||||
**验证方法：**
|
||||
|
||||
```python
|
||||
# 测试代码
|
||||
await __event_emitter__({
|
||||
"type": "function_call",
|
||||
"data": {"name": "test_func", "arguments": {}}
|
||||
})
|
||||
```
|
||||
|
||||
### 3. 异步执行超时
|
||||
|
||||
**问题：** 某些 Functions 可能执行很慢
|
||||
|
||||
**解决方案：**
|
||||
|
||||
- 实现 timeout 机制（已在 executor 中实现）
|
||||
- 对于长时间运行的任务，考虑返回"processing"状态
|
||||
|
||||
---
|
||||
|
||||
## 📝 实现清单
|
||||
|
||||
- [ ] 实现 _function_registry 和 _function_metadata
|
||||
- [ ] 实现 _register_openwebui_functions()
|
||||
- [ ] 实现 _create_function_executor()
|
||||
- [ ] 实现 _create_tool_handler()
|
||||
- [ ] 实现 3-5 个常用内置 Functions
|
||||
- [ ] 测试 Function 注册和调用流程
|
||||
- [ ] 验证 **event_emitter** 机制
|
||||
- [ ] 研究 OpenWebUI Functions API
|
||||
- [ ] 添加错误处理和超时机制
|
||||
- [ ] 更新文档
|
||||
|
||||
---
|
||||
|
||||
**下一步行动：**
|
||||
|
||||
1. 实现基础的 Function Registry
|
||||
2. 添加 2-3 个简单的内置 Functions（如 get_time, calculate）
|
||||
3. 测试基本的 Tool Calling 流程
|
||||
4. 根据测试结果调整架构
|
||||
|
||||
**作者：** Fu-Jie
|
||||
**日期：** 2026-01-26
|
||||
Reference in New Issue
Block a user