[ PROMPT_NODE_24532 ]
custom_functions
[ SKILL_DOCUMENTATION ]
# 自定义函数参考
在 CocoIndex 中创建自定义函数的完整指南。
## 概述
自定义函数允许创建可在工作流中使用的转换逻辑。有两种方法：
1. **独立函数** - 简单，无需配置或设置逻辑
2. **函数规范 + 执行器** - 高级，具有配置和设置逻辑
## 独立函数
用于不需要配置或设置的简单转换。
### 基本示例
python
@cocoindex.op.function(behavior_version=1)
def compute_word_count(text: str) -> int:
"""计算文本中的单词数。"""
return len(text.split())
**要求：**
- 使用 `@cocoindex.op.function()` 装饰
- 所有参数和返回值必须有类型注解
- 支持基本类型、结构体、表和 numpy 数组
### 带可选参数
python
@cocoindex.op.function(behavior_version=1)
def extract_info(content: str, filename: str, max_length: int | None = None) -> dict:
"""
从内容中提取信息。
参数:
content: 文档内容
filename: 源文件名
max_length: 可选的截断最大长度
"""
info = {
"filename": filename,
"length": len(content),
"word_count": len(content.split())
}
if max_length and len(content) > max_length:
info["truncated"] = True
return info
### 在工作流中使用
python
@cocoindex.flow_def(name="MyFlow")
def my_flow(flow_builder: cocoindex.FlowBuilder, data_scope: cocoindex.DataScope):
data_scope["documents"] = flow_builder.add_source(
cocoindex.sources.LocalFile(path="documents")
)
collector = data_scope.add_collector()
with data_scope["documents"].row() as doc:
# 使用独立函数
doc["word_count"] = doc["content"].transform(compute_word_count)
# 带附加参数
doc["info"] = doc["content"].transform(
extract_info,
filename=doc["filename"],
max_length=1000
)
collector.collect(
filename=doc["filename"],
word_count=doc["word_count"],
info=doc["info"]
)
collector.export("documents", cocoindex.targets.Postgres(), primary_key_fields=["filename"])
## 函数规范 + 执行器
用于需要配置或设置逻辑（例如加载模型）的函数。
### 基本结构
python
# 1. 定义函数规范 (configu
粤公网安备44030002003366号