125 lines
4.4 KiB
Python
125 lines
4.4 KiB
Python
import re
|
||
|
||
|
||
def calculate_rate(user_input: str) -> dict:
|
||
"""
|
||
从自然语言中提取两个数值并计算比例。
|
||
|
||
V1 目标不是构建复杂公式引擎，而是提供一个可演示的“业务工具”示例：
|
||
只要输入中出现两个数字，就将其解释为“已完成值 / 总数”。
|
||
"""
|
||
numbers = [float(item) for item in re.findall(r"\d+(?:\.\d+)?", user_input)]
|
||
if len(numbers) < 2:
|
||
return {
|
||
"success": False,
|
||
"rate": 0.0,
|
||
"numerator": 0.0,
|
||
"denominator": 0.0,
|
||
"note": "未能从输入中提取两个数字，无法计算比例。",
|
||
}
|
||
numerator, denominator = numbers[0], numbers[1]
|
||
if denominator == 0:
|
||
return {
|
||
"success": False,
|
||
"rate": 0.0,
|
||
"numerator": numerator,
|
||
"denominator": denominator,
|
||
"note": "分母为 0，无法计算比例。",
|
||
}
|
||
return {
|
||
"success": True,
|
||
"numerator": numerator,
|
||
"denominator": denominator,
|
||
"rate": round(numerator / denominator, 4),
|
||
"note": "已按输入中的前两个数字完成比例计算。",
|
||
}
|
||
|
||
|
||
def query_demo_records(user_input: str) -> dict:
|
||
"""
|
||
查询示例业务记录。
|
||
|
||
该工具依赖 Audit 模块中的 DemoBusinessRecord 演示表，用于证明
|
||
“场景 + 结构化数据 + 工具调用”可以组成更可信的业务 Agent。
|
||
"""
|
||
try:
|
||
from apps.audit.models import DemoBusinessRecord
|
||
except Exception as exc:
|
||
return {"records": [], "error": str(exc)}
|
||
|
||
queryset = DemoBusinessRecord.objects.all()
|
||
tokens = {token.strip().lower() for token in user_input.split() if token.strip()}
|
||
scenario_ids = set(queryset.values_list("scenario_id", flat=True))
|
||
record_types = set(queryset.values_list("record_type", flat=True))
|
||
matched_scenario_ids = scenario_ids & tokens
|
||
matched_record_types = record_types & tokens
|
||
if matched_scenario_ids:
|
||
queryset = queryset.filter(scenario_id__in=matched_scenario_ids)
|
||
if matched_record_types:
|
||
queryset = queryset.filter(record_type__in=matched_record_types)
|
||
records = [
|
||
{
|
||
"id": record.id,
|
||
"scenario_id": record.scenario_id,
|
||
"record_type": record.record_type,
|
||
"title": record.title,
|
||
"payload": record.payload,
|
||
}
|
||
for record in queryset[:20]
|
||
]
|
||
return {"records": records}
|
||
|
||
|
||
def check_required_fields(user_input: str) -> dict:
|
||
"""
|
||
检查输入中声明的必填项是否全部出现。
|
||
|
||
约定格式示例：
|
||
“请检查必填项：合同编号、供应商、金额。当前只提供了合同编号和金额。”
|
||
"""
|
||
required_match = re.search(r"必填项[:：](.+?)(?:。|\.)", user_input)
|
||
provided_match = re.search(r"(?:当前|已|仅)?提供了(.+?)(?:。|\.)", user_input)
|
||
required_fields = _split_cn_items(required_match.group(1) if required_match else "")
|
||
provided_fields = set(_split_cn_items(provided_match.group(1) if provided_match else ""))
|
||
missing_fields = [field for field in required_fields if field not in provided_fields]
|
||
return {
|
||
"required_fields": required_fields,
|
||
"provided_fields": list(provided_fields),
|
||
"missing_fields": missing_fields,
|
||
"note": "已根据输入中的“必填项/提供了”描述完成检查。",
|
||
}
|
||
|
||
|
||
def generate_action_items(user_input: str) -> dict:
|
||
"""
|
||
生成最小可执行行动项。
|
||
|
||
该工具主要用于演示“模型回答之外，还可以得到结构化待办建议”。
|
||
"""
|
||
return {
|
||
"items": [
|
||
"先确认问题背景和适用场景。",
|
||
f"围绕当前问题继续核实：{user_input}",
|
||
"根据知识库和审计结果安排下一步处理动作。",
|
||
]
|
||
}
|
||
|
||
|
||
def _split_cn_items(raw_text: str) -> list[str]:
|
||
"""将中文顿号、逗号和连接词分隔的字段串切分为列表。"""
|
||
normalized = (
|
||
raw_text.replace("和", "、")
|
||
.replace("以及", "、")
|
||
.replace(",", "、")
|
||
.replace("，", "、")
|
||
)
|
||
return [item.strip(" 。.") for item in normalized.split("、") if item.strip(" 。.")]
|
||
|
||
|
||
BUILTIN_TOOLS = {
|
||
"calculate_rate": calculate_rate,
|
||
"query_demo_records": query_demo_records,
|
||
"check_required_fields": check_required_fields,
|
||
"generate_action_items": generate_action_items,
|
||
}
|