当你们的工具能追踪自己的时候,你们可以自己进进进进取,而自己选择;
注: 这是语义情报系列第8部分,第7部分涵盖了国防和安全部队的整体结构。 工具本身如何运作,如何跟踪使用,如何演变,如何随着时间推移变得更聪明。
注: 如果你认为第7部分的工作流程演变是疯狂的, 等到你看到当 每个工具 拥有同样的能力。
在第七部分,我向大家展示了导导合成进化:计划、生成、执行、评价和改进的工作流程。我多次提到“工具”。
这就是我没有解释的:
这些工具不是静态的 它们不是没有变化的配置文件
他们是活生生的文物
换言之: 工具是节点。 节点是工具。 一切都在演变中 。
越来越怪异了
让我向大家展示一下这个系统实际上有什么:
$ ls -la tools/
drwxr-xr-x llm/ # LLM-based tools (27 specialists)
drwxr-xr-x executable/ # Executable validators/generators
drwxr-xr-x openapi/ # External API integrations
drwxr-xr-x custom/ # User-defined tools
-rw-r--r-- index.json # 5,464 lines of tool metadata
这: index.json? 5 464条线路 工具定义、使用数据、版本历史、健身分数和线条跟踪。
里面的每一个工具:
注意: 系统实际上运行时没有任何工具。 效率更低; 更笨。 没有这些工具, 工具就会生成成为工作流程分解的正常部分。 它仍然会缓慢地适应, 但需要更多符号 。
让我们看看工具实际上是什么。
以下是系统上真正的工具定义:
tools/llm/long_form_writer.yaml
name: "Long-Form Content Writer"
type: "llm"
description: "Specialized for writing long-form content (novels, books, long articles) using mistral-nemo's massive 128K context window."
cost_tier: "high"
speed_tier: "slow"
quality_tier: "excellent"
max_output_length: "very-long"
llm:
model: "mistral-nemo"
endpoint: null
system_prompt: "You are a creative writer specializing in long-form content. You have a massive 128K token context window..."
prompt_template: "{prompt}\n\nPrevious context:\n{context}\n\nGenerate the next section maintaining consistency."
tags: ["creative-writing", "novel", "story", "long-form", "article", "book", "large-context"]
注意那里有什么:
但这里是 否 载在叶尔孤白中的,
# Auto-generated at runtime:
tool.usage_count = 47 # How many times used
tool.version = "1.2.0" # Semantic versioning
tool.definition_hash = "a3f5..."# Change detection
tool.quality_score = 0.89 # From evaluations
tool.avg_latency_ms = 12_400 # Performance tracking
tool.last_updated = "2025-11-15"
系统系统 增强 具有运行时间学习的静态定义。
当你使用一个工具时会发生什么:
# User request
result = tools_manager.invoke_llm_tool(
tool_id="long_form_writer",
prompt="Write a romance novel chapter"
)
# Behind the scenes:
sequenceDiagram
participant U as User
participant TM as ToolsManager
participant RAG as RAG Memory
participant LLM as Long Form Writer
participant Metrics as Metrics Tracker
U->>TM: invoke_llm_tool("long_form_writer", prompt)
TM->>RAG: Check cache (tool + prompt hash)
alt Cache Hit
RAG-->>TM: Cached response (v1.2.0, fitness: 0.89)
TM->>Metrics: Increment cache_hits
TM-->>U: Return cached result ✓
else Cache Miss
TM->>Metrics: Start timer
TM->>LLM: Generate response
LLM-->>TM: Response
TM->>Metrics: Record latency, quality
TM->>RAG: Store invocation with metadata
TM->>TM: Update tool.usage_count++
TM-->>U: Return result
end
TM->>Metrics: Update adaptive timeout stats
TM->>RAG: Update tool fitness score
追踪到的是什么 :
让我们来看看 缓存机制。
最聪明的一点: 系统缓存工具多层引用 。
级别 1: 精确匹配抓取
def invoke_llm_tool(self, tool_id: str, prompt: str) -> str:
"""Invoke LLM tool with hierarchical caching."""
# Normalize prompt for exact matching
normalized_prompt = prompt.lower().strip()
# Search RAG for previous invocations
tool_invocations = self.rag_memory.find_by_tags(
["tool_invocation", tool_id],
limit=100
)
# Find ALL exact matches for this tool + prompt
matches = []
for artifact in tool_invocations:
cached_prompt = artifact.metadata.get("user_prompt", "").lower().strip()
if cached_prompt == normalized_prompt:
# Collect fitness and version
matches.append({
"artifact": artifact,
"fitness": artifact.metadata.get("fitness_score", 0.0),
"version": artifact.metadata.get("version", "1.0.0"),
"timestamp": artifact.metadata.get("timestamp", 0)
})
if matches:
# Select LATEST, HIGHEST FITNESS version
best_match = sorted(
matches,
key=lambda m: (m["fitness"], m["timestamp"]),
reverse=True
)[0]
logger.info(
f"✓ CACHE HIT: Reusing result for '{tool.name}' "
f"(version {best_match['version']}, fitness {best_match['fitness']:.2f})"
)
# Increment usage counters
self.increment_usage(tool_id)
self.rag_memory.increment_usage(artifact.artifact_id)
return best_match["artifact"].content
为什么这很重要:
如果你要求系统“写一则代号”两次, 第二次是 即时。 LLM 不运行。 RAG 内存返回缓存结果 。
但以下是聪明的一点: 返回 BEST 版本 如果存在多个 。
示例:
Invocation 1: "write a haiku about code"
→ Generated with tool v1.0.0
→ Fitness: 0.75
→ Stored in RAG
Invocation 2: "write a haiku about code" (exact match!)
→ Tool evolved to v1.1.0
→ Fitness: 0.92 (better!)
→ Stored in RAG
Invocation 3: "write a haiku about code"
→ Finds BOTH cached versions
→ Selects v1.1.0 (higher fitness + later timestamp)
→ Returns best result instantly
系统系统 自动选择最高质量的缓存结果.
一种微妙的特征是:系统了解每个模型的反应需要多长时间。
问题:
不同模型的反应时间大不相同:
tinyllama (2B):~3秒llama3 (8B):~10秒qwen2.5-coder (14B):~25秒deepseek-coder-v2 (16B):~60秒如果你设置一个全球超时(例如30秒),你就浪费27秒等待 tinyllama,而你杀死 deepseek 在它完成之前。
解决方案:适应性学习
def _update_adaptive_timeout(
self,
model: str,
tool_id: str,
response_time: float,
timed_out: bool,
prompt_length: int
):
"""Learn optimal timeout from actual performance."""
# Get existing stats
stats_id = f"timeout_stats_{model.replace(':', '_')}"
existing = self.rag_memory.get_artifact(stats_id)
if existing:
response_times = existing.metadata.get("response_times", [])
timeout_count = existing.metadata.get("timeout_count", 0)
success_count = existing.metadata.get("success_count", 0)
else:
response_times = []
timeout_count = 0
success_count = 0
# Update stats
if timed_out:
timeout_count += 1
else:
success_count += 1
response_times.append(response_time)
response_times = response_times[-50:] # Keep last 50
# Calculate recommended timeout (95th percentile + 20% buffer)
if response_times:
sorted_times = sorted(response_times)
p95_index = int(len(sorted_times) * 0.95)
p95_time = sorted_times[min(p95_index, len(sorted_times) - 1)]
recommended_timeout = int(p95_time * 1.2)
logger.info(
f"Adaptive timeout for {model}: {recommended_timeout}s "
f"(based on {len(response_times)} samples)"
)
如何运作:
结果:
Model: tinyllama
Samples: 50
95th percentile: 3.2s
Recommended timeout: 4s (3.2 * 1.2)
Model: qwen2.5-coder:14b
Samples: 50
95th percentile: 28.5s
Recommended timeout: 34s (28.5 * 1.2)
系统系统 学习每种模型的正确超时 而不是使用一个全球值 。
当你要求系统做一些事情时, 它不只是选择第一个匹配工具。 它运行一个 健身功能 跨越多个维度。
健身性计算:
def calculate_fitness(tool, similarity_score):
"""
Calculate overall fitness score (0-100+).
Factors:
- Semantic similarity (how well it matches the task)
- Speed (fast tools get bonus)
- Cost (cheap tools get bonus)
- Quality (high-quality tools get bonus)
- Historical success rate~~~~
- Latency metrics
- Reuse potential
"""
fitness = similarity_score * 100 # Base: 0-100
metadata = tool.metadata or {}
# Speed bonus/penalty
speed_tier = metadata.get('speed_tier', 'medium')
if speed_tier == 'very-fast':
fitness += 20
elif speed_tier == 'fast':
fitness += 10
elif speed_tier == 'slow':
fitness -= 10
elif speed_tier == 'very-slow':
fitness -= 20
# Cost bonus (cheaper = better for most tasks)
cost_tier = metadata.get('cost_tier', 'medium')
if cost_tier == 'free':
fitness += 15
elif cost_tier == 'low':
fitness += 10
elif cost_tier == 'high':
fitness -= 10
elif cost_tier == 'very-high':
fitness -= 15
# Quality bonus
quality_tier = metadata.get('quality_tier', 'good')
if quality_tier == 'excellent':
fitness += 15
elif quality_tier == 'very-good':
fitness += 10
elif quality_tier == 'poor':
fitness -= 15
# Success rate from history
quality_score = metadata.get('quality_score', 0)
if quality_score > 0:
fitness += quality_score * 10 # 0-10 bonus
# Latency metrics
latency_ms = metadata.get('latency_ms', 0)
if latency_ms > 0:
if latency_ms < 100:
fitness += 15 # Very fast
elif latency_ms < 500:
fitness += 10
elif latency_ms > 5000:
fitness -= 10 # Too slow
# Reuse bonus: existing workflow = less effort
if tool.tool_type == ToolType.WORKFLOW:
if similarity >= 0.90:
fitness += 30 # Exact match!
elif similarity >= 0.70:
fitness += 15 # Template reuse
return fitness
真实示例:
Task: "Quickly validate this email address"
Tools found:
1. email_validator_workflow (similarity: 0.95)
- Speed: very-fast (+20)
- Cost: free (+15)
- Quality: excellent (+15)
- Latency: 45ms (+15)
- Reuse: exact match (+30)
→ FINAL FITNESS: 190
2. general_validator (similarity: 0.70)
- Speed: medium (+0)
- Cost: free (+15)
- Quality: good (+10)
- Latency: 850ms (+0)
- Reuse: none (+0)
→ FINAL FITNESS: 95
3. llm_based_validator (similarity: 0.65)
- Speed: slow (-10)
- Cost: high (-10)
- Quality: excellent (+15)
- Latency: 8200ms (-10)
- Reuse: none (+0)
→ FINAL FITNESS: 50
选中 : email_validator_workflow (适用性:190)
系统选择 快速、免费、高质量、经过证明 答案不是最相似的词义,也不是最强大的
最能满足多种制约的。
工具不是静止的,而是进化的
版本和更改检测 :
每个工具都有 定义 shash 根据YAML计算:
def calculate_tool_hash(tool_def: Dict[str, Any]) -> str:
"""SHA256 hash of tool definition for change detection."""
stable_json = json.dumps(tool_def, sort_keys=True)
return hashlib.sha256(stable_json.encode('utf-8')).hexdigest()
当您编辑工具的 YAML 时:
# BEFORE (v1.0.0)
name: "Email Validator"
tags: ["email", "validation"]
# AFTER (edit the YAML)
name: "Email Validator"
tags: ["email", "validation", "dns-check"] # Added DNS checking!
下次装入时 :
# System detects change
new_hash = calculate_tool_hash(tool_def) # Different!
old_hash = existing_tool.definition_hash
if old_hash != new_hash:
# Determine change type
change_type = tool_def.get("change_type", "patch") # minor, major, patch
# Bump version
old_version = "1.0.0"
new_version = bump_version(old_version, change_type)
# new_version = "1.1.0" (minor change)
console.print(
f"[yellow]↻ Updated email_validator "
f"v{old_version} → v{new_version} ({change_type})[/yellow]"
)
语义版本 :
def bump_version(current_version: str, change_type: str) -> str:
"""Bump semver based on change type."""
major, minor, patch = map(int, current_version.split('.'))
if change_type == "major":
return f"{major + 1}.0.0" # Breaking changes
elif change_type == "minor":
return f"{major}.{minor + 1}.0" # New features
else: # patch
return f"{major}.{minor}.{patch + 1}" # Bug fixes
断开更改 :
name: "Email Validator"
version: "2.0.0"
change_type: "major"
breaking_changes:
- "Changed return format from boolean to object"
- "Removed deprecated 'simple_check' parameter"
- "Now requires 'domain' to be specified"
装载时 :
[yellow]↻ Updated email_validator v1.3.2 → v2.0.0 (major)[/yellow]
[red]! Breaking changes:[/red]
- Changed return format from boolean to object
- Removed deprecated 'simple_check' parameter
- Now requires 'domain' to be specified
系统系统 警告您要打破更改 并保存版本历史。
每个工具都会在 RAG 内存中索引, 用于语义搜索 :
装入时间 :
def _store_yaml_tool_in_rag(self, tool: Tool, tool_def: dict, yaml_path: str):
"""Store YAML tool in RAG for semantic search."""
# Build comprehensive content for embedding
content_parts = [
f"Tool: {tool.name}",
f"ID: {tool.tool_id}",
f"Type: {tool.tool_type.value}",
f"Description: {tool.description}",
f"Tags: {', '.join(tool.tags)}",
""
]
# Add input/output schemas
if tool_def.get("input_schema"):
content_parts.append("Input Parameters:")
for param, desc in tool_def["input_schema"].items():
content_parts.append(f" - {param}: {desc}")
# Add examples
if tool_def.get("examples"):
content_parts.append("Examples:")
for example in tool_def["examples"]:
content_parts.append(f" {example}")
# Add performance tiers
content_parts.append("Performance:")
content_parts.append(f" Cost: {tool_def['cost_tier']}")
content_parts.append(f" Speed: {tool_def['speed_tier']}")
content_parts.append(f" Quality: {tool_def['quality_tier']}")
# Add full YAML
import yaml
content_parts.append("Full Definition:")
content_parts.append(yaml.dump(tool_def))
tool_content = "\n".join(content_parts)
# Store in RAG with metadata
self.rag_memory.store_artifact(
artifact_id=f"tool_{tool.tool_id}",
artifact_type=ArtifactType.PATTERN,
name=tool.name,
description=tool.description,
content=tool_content,
tags=["tool", "yaml-defined", tool.tool_type.value] + tool.tags,
metadata={
"tool_id": tool.tool_id,
"tool_type": tool.tool_type.value,
"is_tool": True,
"version": tool_def.get("version", "1.0.0"),
"cost_tier": tool_def.get("cost_tier"),
"speed_tier": tool_def.get("speed_tier"),
"quality_tier": tool_def.get("quality_tier")
},
auto_embed=True # Generate embedding!
)
现在搜索时:
# Semantic tool search
results = tools_manager.search("email validation", top_k=5)
# Results (ranked by fitness, not just similarity):
[
Tool(id="email_validator", fitness=190, similarity=0.95),
Tool(id="domain_checker", fitness=140, similarity=0.82),
Tool(id="regex_validator", fitness=110, similarity=0.78),
Tool(id="general_validator", fitness=95, similarity=0.70),
Tool(id="string_validator", fitness=60, similarity=0.65)
]
系统使用 RAG 嵌入嵌入 寻找相关工具,然后按 多维健身.
在我们的系统中,我们保存向量 嵌入 解冻 这是一个矢量数据库, 它为我们提供了一个清晰的方法 来查看我们的工具空间。 这是我们的工作流程建设系统的记忆系统 。
被分割到原始模板( 工具目录中的 Yaml 文件) 和为解析任务生成的代码的每一个元素( 以及 llms 等的配置 ) 。 这些装饰成装配工作服的工具包。 预建的块块范围不一而足, 范围不一;
如此一来,所有工作站都是可作成和可测试的,因为每个Python元素都有一套测试、BDD特征、用于核查正确性的静态工具以及多个LLM评价员,以确保工作正常。
副作用是每一个在工作流程中运行的代码 都在那里, 准备检查,因为每个“工具”的创建 都会导致可检查的 Python Scripes 。
你可以看到那些太过热 和每一个小肿块一起 成为一套模擬的 连结工具,比如:
自然而然地以系统的性质 来专门研究这个系统
在未来,我们可能想要优化这些组群 将代码库缩小到一个更小,更紧,更优化的系统。
我们可以追踪版本、用法、变化和遗留物, 我们可以有选择地优化使用, 并且“集束脱裂”是最常用的, 也是大多数性能至关重要的共通点, 作为系统内在结构的一部分。
让我们看看现在系统里到底存在什么
法学硕士工具(27名专家):
$ ls tools/llm/
article_analyzer.yaml # Analyzes articles for structure/quality
code_explainer.yaml # Explains code in natural language
code_optimizer.yaml # Hierarchical optimization (local/cloud/deep)
code_reviewer.yaml # Reviews code for quality/security
content_generator.yaml # General content generation
doc_generator.yaml # Generates documentation
fast_code_generator.yaml # Quick code generation (small models)
general.yaml # General-purpose fallback
long_form_writer.yaml # Novels, books (128K context!)
model_selector.yaml # Selects best backend/model
performance_profiler.yaml # Profiles code performance
quick_feedback.yaml # Fast triage/feedback
quick_translator.yaml # Fast translation
security_auditor.yaml # Security vulnerability scanning
signalr_connection_parser.yaml # Parses SignalR connections
signalr_llmapi_management.yaml # Manages SignalR LLM API
summarizer.yaml # Summarizes long content
task_to_workflow_router.yaml # Routes tasks to workflows
technical_writer.yaml # Technical documentation
translation_quality_checker.yaml # Validates translations
workflow_documenter.yaml # Auto-generates workflow docs
可执行工具 :
$ ls tools/executable/
call_tool_validator.yaml # Validates call_tool() usage
connect_signalr.yaml # SignalR connection tool
document_workflow.yaml # Workflow documentation generator
mypy_type_checker.yaml # Static type checking
python_syntax_validator.yaml # Syntax validation
run_static_analysis.yaml # Static analysis runner
save_to_disk.yaml # Disk persistence
signalr_hub_connector.yaml # Hub connection
signalr_websocket_stream.yaml # WebSocket streaming
unit_converter.yaml # Unit conversion utilities
OpenAPI 工具 :
$ ls tools/openapi/
nmt_translator.yaml # Neural machine translation API
工具总计 : 50+
元数据项目总额: 5,464条线英寸 index.json
让我向你们展示这个系统中最先进的工具: 代码优化.
定义: tools/llm/code_optimizer.yaml (317条线! )
它会做什么:
等级优化:
optimization_levels:
- name: "local"
model_key: "escalation" # qwen2.5-coder:14b
cost_usd: 0.0
expected_improvement: 0.10 # 10%
triggers:
- "Default for all optimizations"
- "Quick wins, obvious inefficiencies"
- name: "cloud"
model_key: "cloud_optimizer" # GPT-4/Claude
cost_usd: 0.50
expected_improvement: 0.30 # 30%
triggers:
- "Local improvement < 15%"
- "Code is critical path"
- "User explicitly requests it"
- name: "deep"
model_key: "deep_analyzer"
cost_usd: 5.0
expected_improvement: 0.50 # 50%
triggers:
- "Workflow/system-level optimization"
- "Cloud improvement < 25%"
- "Architectural changes needed"
费用管理:
cost_management:
max_daily_budget: 50.0 # USD
fallback_on_budget_exceeded: "local"
optimization_strategy: |
1. Always try LOCAL first (free)
2. Escalate to CLOUD if:
- Local improvement < 15%
- Reuse count > 100
3. Escalate to DEEP if:
- Cloud improvement < 25%
- System-level changes needed
测试集成 :
test_integration:
auto_update: true
test_discovery:
- "Find test_*.py in tests/"
- "Identify tests for specific functions"
test_generation:
- "Generate missing tests"
- "Add performance assertions"
- "Create regression tests"
版本管理 :
version_management:
semver: true
breaking_change_detection:
- "Function signature changed"
- "Return type changed"
- "Dependencies added/removed"
auto_migration:
enabled: true
conditions:
- "No breaking changes"
- "All tests pass"
- "Improvement >= 10%"
这个 单一工具 弦乐 :
和它,它 只有一个工具 在一个系统中 50+工具.
最现代的工具之一: 模型选择器.
自然语言选择:
# User says: "using the most powerful code llm review this code"
selection = tools_manager.invoke_llm_tool(
tool_id="model_selector",
prompt="using the most powerful code llm review this code"
)
# Result:
{
"backend": "anthropic",
"model": "claude-3-opus-20240229",
"reasoning": "Request specifies 'most powerful'. Claude Opus is the highest-quality code model.",
"confidence": 0.95,
"cost_tier": "very-high",
"speed_tier": "slow",
"quality_tier": "exceptional"
}
如何运作:
def select_model(
self,
task_description: str,
constraints: Optional[Dict[str, Any]] = None
) -> List[Dict[str, Any]]:
"""Select best model for task."""
task_lower = task_description.lower()
# Parse natural language preferences
backend_preference = None
if any(kw in task_lower for kw in ["openai", "gpt"]):
backend_preference = "openai"
elif any(kw in task_lower for kw in ["anthropic", "claude"]):
backend_preference = "anthropic"
# Parse model preference
model_preference = None
if "gpt-4o" in task_lower:
model_preference = "gpt-4o"
elif "opus" in task_lower:
model_preference = "opus"
# Analyze task characteristics
needs_long_context = any(w in task_lower for w in
["book", "novel", "document", "large", "long"])
needs_coding = any(w in task_lower for w in
["code", "function", "script", "program"])
needs_speed = any(w in task_lower for w in
["quick", "fast", "immediate"])
needs_quality = any(w in task_lower for w in
["complex", "analysis", "reasoning"])
# Score each model
scores = {}
for backend_model_id, info in self.backends.items():
score = 50.0 # Base
# Backend preference
if backend_preference and info["backend"] == backend_preference:
score += 50
# Model preference
if model_preference and model_preference in info["model"].lower():
score += 100 # Strong boost
# Context window
if needs_long_context:
context = info.get("context_window", 8192)
if context >= 100000:
score += 40
# Speed
if needs_speed:
if info["speed"] == "very-fast":
score += 30
# Quality
if needs_quality:
if info["quality"] == "excellent":
score += 30
# Specialization
if needs_coding:
if "code" in info.get("best_for", []):
score += 35
scores[backend_model_id] = score
# Return top-ranked models
ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True)
return [self.backends[bid] for bid, score in ranked[:3]]
系统分析自然语言 要选择模型。您可以说:
它聪明地向右后端前进
该系统对外部API的处理与对内部工具的处理相同。
实例:NMT笔译员
name: "NMT Translation Service"
type: "openapi"
description: "Neural machine translation API. VERY FAST but needs validation."
cost_tier: "low"
speed_tier: "very-fast"
quality_tier: "good"
openapi:
spec_url: "http://localhost:8000/openapi.json"
base_url: "http://localhost:8000"
code_template: |
import requests
def translate_text(text, source_lang="en", target_lang="de"):
url = "http://localhost:8000/translate"
params = {
"text": text,
"source_lang": source_lang,
"target_lang": target_lang
}
response = requests.get(url, params=params)
return response.json()["translations"][0]
tags: ["translation", "nmt", "api", "external"]
运行时 :
# System loads OpenAPI spec
openapi_tool = OpenAPITool(
tool_id="nmt_translator",
spec_url="http://localhost:8000/openapi.json"
)
# Parses operations
operations = openapi_tool.list_operations()
# [
# {"operation_id": "translate", "method": "GET", "path": "/translate"},
# {"operation_id": "get_languages", "method": "GET", "path": "/languages"}
# ]
# Invoke
result = tools_manager.invoke_openapi_tool(
"nmt_translator",
"translate",
parameters={"text": "hello", "source_lang": "en", "target_lang": "de"}
)
# Result: {"success": True, "data": {"translations": ["Hallo"]}}
追踪到的是什么 :
# Stored in RAG:
{
"artifact_type": "API_INVOCATION",
"tool_id": "nmt_translator",
"operation_id": "translate",
"status_code": 200,
"success": True,
"latency_ms": 124,
"parameters": {"text": "hello", "source_lang": "en", "target_lang": "de"},
"response": {"translations": ["Hallo"]}
}
外部广告商获得同样的待遇:
最野生的工具之一: 工作流程文件员_文件员.
它会做什么:
工作流程(a) main.py 文件)和 自动自动生成全面文档 以:
README.txt所有的自动。
定义: tools/llm/workflow_documenter.yaml (11,803个字符! )
输入 :
{
"workflow_path": "nodes/email_validator/main.py"
}
产出:
## Overview
Validates email addresses and optionally checks domain matching.
## What It Does
This workflow checks if an email address is valid using regex.
If you provide a domain, it checks if the email belongs to that domain.
## Required Inputs
- **email** (string, required)
- The email address to validate
- Example: "[email protected]"
- **domain** (string, optional)
- The domain to check against
- Example: "example.com"
## Process Flow
```mermaid
flowchart TD
A[Start: Receive Input] --> B[Extract email and domain]
B --> C{Email provided?}
C -->|No| D[Error: email required]
C -->|Yes| E[Validate email format]
E --> F{Valid format?}
F -->|No| G[Return: invalid]
F -->|Yes| H{Domain provided?}
H -->|No| I[Return: valid]
H -->|Yes| J[Extract email domain]
J --> K{Domains match?}
K -->|Yes| I
K -->|No| L[Return: domain_mismatch]
curl -X POST http://localhost:8080/execute/email_validator \
-H "Content-Type: application/json" \
-d '{"email": "[email protected]", "domain": "example.com"}'
result = call_tool("email_validator", {
"email": "[email protected]",
"domain": "example.com"
})
问题 : 这能否验证是否有电子邮件 ? A: 否, 仅此验证格式。 使用 DNS/ SMTP 检查是否存在 。
问题:它是否支持国际领域? A:是的,但可能需要双码转换。
**Saved to:** `nodes/email_validator/README.txt`
**The tool GENERATES ALL OF THIS** by analyzing the code.
## The Self-Expanding Toolkit
Here's where it gets wild: **tools generate tools**.
**Example Flow:**
用户:"我需要一个转换温度的工具"
系统 :
新创建的工具: 温度_ 转换器. yaml
**The system grows its own toolkit.**
## Tool Statistics: What The System Knows
```python
stats = tools_manager.get_statistics()
# Result:
{
"total_tools": 53,
"by_type": {
"llm": 27,
"executable": 19,
"openapi": 3,
"workflow": 2,
"custom": 2
},
"tag_distribution": {
"code": 15,
"validation": 12,
"translation": 8,
"optimization": 5,
"documentation": 4,
...
},
"most_used": [
{"id": "general", "name": "General Purpose LLM", "usage": 1247},
{"id": "code_optimizer", "name": "Code Optimizer", "usage": 89},
{"id": "nmt_translator", "name": "NMT Translator", "usage": 67},
{"id": "email_validator", "name": "Email Validator", "usage": 45},
{"id": "long_form_writer", "name": "Long-Form Writer", "usage": 23}
]
}
系统系统 已经知道:
和它 和它 使用此数据 至:
让我们退后想想我们实际上已经建造了些什么:
系统,其中:
我们创造了一个自我优化工具箱:
这不是配置管理
这是新兴工具生态学。
工具不是静态资源 进化系统中活的文物.
设想1:自我改进关键道路
System detects: email_validator used 500 times, fitness: 0.75
Action: Trigger code_optimizer with level=cloud (high reuse count)
Result: email_validator v2.0.0, fitness: 0.92
Migration: Auto-update all 15 workflows using v1.x to v2.0.0
Validation: Re-run all tests, all pass
Outcome: 23% performance improvement, no breaking changes
该系统在没有人类干预的情况下优化了其自身的关键道路。
设想情景2:适应性专业化
Pattern detected: "translate article" requested 20 times
Analysis: Using nmt_translator + translation_quality_checker every time
Decision: Generate specialized "article_translator" tool
Implementation:
- Combines both tools into one
- Adds caching for common phrases
- Optimizes for article-length text
- Auto-generates documentation
Registration: article_translator v1.0.0 added to registry
Fitness: 0.89 (vs 0.73 for manual combination)
Usage: Immediately used for next translation request
该系统确定了一种模式,并创建了一个专门工具。
设想方案3:成本-软件升级
Request: "optimize this function"
Level 1 (LOCAL): qwen2.5-coder:14b (free)
- Improvement: 8% (below 10% threshold)
- Decision: Escalate to CLOUD
Level 2 (CLOUD): claude-3-5-sonnet ($0.50)
- Improvement: 28% (good!)
- Cost: $0.50 (within budget)
- Decision: Accept
Result: Function optimized 28%, cost $0.50
Update: Store both versions in RAG
Mark v1 as "suboptimal", v2 as "optimized"
Future: Always use v2 for this function
该系统明智地花费了资金,以取得更好的成果。
让我把现在实际存在的东西编成目录
守则专家
code_explainer - 解释自然语言的代码code_optimizer - 等级优化(地方/广度/深度)code_reviewer - 质量和安全审查fast_code_generator - 具有小型型号的快速发电security_auditor - 脆弱性扫描performance_profiler - 法规特征分析和分析内容专家:
long_form_writer - 新书、新书(128K上下文)content_generator - 一般内容article_analyzer - 条款结构/质量summarizer - 概述长篇内容proofreader - 语法和风格seo_optimizer - SEO优化outline_generator - 内容大纲翻译:
quick_translator - 快速翻译(小型模式)translation_quality_checker - 验证翻译文 件:
doc_generator - 编码文件technical_writer - 技术文件workflow_documenter - 自动生成工作流程文档系统工具 :
general - 普通用途后退model_selector - 选择最佳后端/模型task_to_workflow_router - 工作流程路线任务quick_feedback - 快速分流signalr_connection_parser - Parseses 信号连接signalr_llmapi_management - 管理信号R LLM API验证 :
call_tool_validator - 校验调用工具()python_syntax_validator - 语法检查mypy_type_checker - 静态类型检查json_output_validator - JSON格式验证stdin_usage_validator - 验证标准使用量main_function_checker - 检查主函数()node_runtime_import_validator - 验证进口分析:
run_static_analysis - 运行静态分析工具performance_profiler - 描述代码性能公用事业:
save_to_disk - 磁盘持久性unit_converter - 单位改划random_data_generator - 测试数据生成buffer - 缓冲管理workflow_datastore - 工作流程数据储存stream_processor - 流流处理sse_stream - 服务器已存在事件整合:
connect_signalr - 信号连接signalr_hub_connector - 枢纽连接signalr_websocket_stream - 网络套件流文 件:
document_workflow - 工作流程文件生成器nmt_translator - 神经机器翻译 API共计:53个工具(和增长)
有趣的是: 工具组成其他工具.
当一种装配的工具演变的时候, 每个使用它的工作流程都自动改进.
让我们看看系统上的一个实际综合工具:
任务 : “将本条译成西班牙文并验证质量”
传统办法:
# Manual composition (brittle, no learning)
translated = nmt_translator.translate(text, "en", "es")
quality = translation_quality_checker.check(translated)
if quality.score < 0.7:
# Retry or error
DSE 方法:
该系统发现这种模式经常使用, 自动创建复合工具:
# tools/llm/validated_translator.yaml (auto-generated!)
name: "Validated Translator"
type: "composite"
description: "Translates text and validates quality automatically. Created from usage pattern analysis."
workflow:
steps:
- id: "translate"
tool: "nmt_translator"
parallel: false
- id: "validate"
tool: "translation_quality_checker"
parallel: false
depends_on: ["translate"]
- id: "retry"
tool: "nmt_translator"
condition: "quality_score < 0.7"
params:
beam_size: 10 # Higher quality on retry
depends_on: ["validate"]
version: "1.0.0"
created_from: "usage_pattern_analysis"
parent_tools: ["nmt_translator", "translation_quality_checker"]
usage_count: 0 # Just created!
什么是疯狂的这:
何时 nmt_translator 演变为 v2. 0.0 (可能更快20%), 复合工具 自动使用新版本无需修改代码。
结果: 使用的每一工作流程 validated_translator 得到20%更快 20%更快 未作任何修改.
以下是更酷的例子: 平行工具构成.
任务 : "彻底审查这个代码"
渐进方法:
# Sequential (SLOW)
security_check = security_auditor.review(code) # 8 seconds
style_check = code_reviewer.review(code) # 12 seconds
performance_check = performance_profiler.analyze(code) # 15 seconds
# TOTAL: 35 seconds
DSE 平行组成 :
# tools/llm/code_review_committee.yaml
name: "Code Review Committee"
type: "composite"
description: "Parallel code review using multiple specialist tools"
workflow:
steps:
# All three run IN PARALLEL
- id: "security"
tool: "security_auditor"
parallel: true
- id: "style"
tool: "code_reviewer"
parallel: true
- id: "performance"
tool: "performance_profiler"
parallel: true
# Aggregate results (runs after all complete)
- id: "aggregate"
tool: "general" # Use general LLM to synthesize
depends_on: ["security", "style", "performance"]
prompt: |
Synthesize these reviews into a cohesive report:
Security: {security.result}
Style: {style.result}
Performance: {performance.result}
Create a prioritized action list.
execution:
max_parallel: 3
timeout_per_tool: 20s
aggregate_timeout: 10s
执行 :
gantt
title Code Review Committee (Parallel Execution)
dateFormat s
axisFormat %S
section Sequential (Old)
Security Check :0, 8s
Style Check :8, 12s
Performance Check :20, 15s
Total: 35s :35, 1s
section Parallel (New)
Security Check :0, 8s
Style Check :0, 12s
Performance Check :0, 15s
Aggregate Results :15, 5s
Total: 20s :20, 1s
结果: 35秒 20秒(更快43%)
和何时 security_auditor 进化为 v3. 0.0 (例如, 30% 更快) , 整个委员会会自动更快 。
这就是真正狂野的部分: 工具就像基因一样.
观察: 当工具证明有用时,它 横跨整个系统.
真实示例: 质量检查器模式
Day 1: translation_quality_checker created
- Usage: 1 (manual test)
- Workflows using it: 0
Day 3: First workflow uses it (article_translator)
- Usage: 15
- Workflows: 1
- Fitness: 0.78
Day 7: Quality checker "gene" spreads
- Usage: 127
- Workflows using it: 7
1. article_translator
2. validated_translator (composite)
3. batch_translator
4. multilingual_content_generator
5. documentation_localizer
6. seo_multilingual_optimizer
7. chat_translator
- Fitness: 0.91 (improved through evolution!)
Day 14: Mutation detected
- translation_quality_checker v2.0.0
- Change: Added context-aware validation
- Breaking change: Output format different
- All 7 workflows auto-migrate
- New fitness: 0.94
Day 30: Specialization emerges
- Original tool spawns specialist: article_quality_checker
- Optimized specifically for article-length text
- 40% faster than general checker
- article_translator auto-switches to specialist
- General checker still used by other 6 workflows
这是字面上的遗传传播:
代码路径为 ALWAYS 最佳, 因为 :
这是非常聪明的一点: 您可以同时改进所有工具.
假设情景 : 您有 20 个工作流程, 每个工作流程使用 5- 10 个工具。 总计 :~ 100 个工具职业 。
传统制度:
Workflow 1 uses Tool A v1.0 (fitness: 0.70)
Workflow 2 uses Tool A v1.0 (fitness: 0.70)
...
Workflow 20 uses Tool A v1.0 (fitness: 0.70)
To improve: Manually edit Tool A, test on each workflow (20 tests!)
Risk: Breaking changes affect all 20 workflows
DSE 系统 :
# Trigger evolution for Tool A
evolve_tool("translation_quality_checker")
# System automatically:
# 1. Analyzes usage patterns across all 20 workflows
# 2. Identifies common failure modes
# 3. Generates improved version (v2.0)
# 4. A/B tests v1.0 vs v2.0 on EACH workflow
# 5. Calculates fitness improvement per workflow
# 6. Auto-migrates workflows where v2.0 is better
# 7. Keeps v1.0 for workflows where v2.0 regresses
结果:
Workflow 1: Tool A v2.0 (fitness: 0.85) ✓ Migrated
Workflow 2: Tool A v1.0 (fitness: 0.72) ✗ Kept old (v2 was worse)
Workflow 3: Tool A v2.0 (fitness: 0.89) ✓ Migrated
...
Workflow 20: Tool A v2.0 (fitness: 0.91) ✓ Migrated
Total migrated: 18/20 workflows (90%)
Average fitness improvement: +15%
你同时训练了一个工具 改进了EightEen工作流程
真正狂野的部分: 通过依赖图显示的进化级联.
示例:
Tool: nmt_translator v1.0 (fitness: 0.73)
Used by:
- validated_translator (composite)
- article_translator
- batch_translator
- chat_translator
Evolution triggered: nmt_translator v1.0 → v2.0
Improvement: 25% faster, 10% better quality
Fitness: 0.73 → 0.88
Cascade effect:
1. validated_translator FITNESS: 0.82 → 0.91 (automatic!)
2. article_translator FITNESS: 0.79 → 0.87 (automatic!)
3. batch_translator FITNESS: 0.75 → 0.83 (automatic!)
4. chat_translator FITNESS: 0.71 → 0.78 (automatic!)
Tools using those tools ALSO improve:
- multilingual_content_generator: 0.76 → 0.84
- documentation_localizer: 0.81 → 0.88
- seo_multilingual_optimizer: 0.69 → 0.77
Total workflows improved: 11
Total time spent: 0 (automatic propagation!)
Total code changes: 0
一个演进事件改进了ELEVE工作流程,没有任何人工干预。
由于工具跟踪健身能力、缓存结果和自动演进,系统 总是运行现有最佳执行工具.
实例执行:
User: "Translate this article to Spanish"
System thinks:
1. Search RAG for "translation" tools
→ Found: nmt_translator, validated_translator, quick_translator
2. Calculate fitness for this specific task:
- nmt_translator: 0.88 (fast, good quality)
- validated_translator: 0.91 (slower, validated)
- quick_translator: 0.76 (very fast, lower quality)
3. Task analysis:
- Input length: 2,500 words (long)
- Quality requirement: high (article)
- Speed requirement: medium (no rush)
4. Decision: Use validated_translator (highest fitness + quality match)
5. Check cache:
- Cache key: hash(tool_id + normalized_prompt)
- Found: 3 cached results
- v1.0 (fitness: 0.82, age: 5 days)
- v1.1 (fitness: 0.89, age: 2 days)
- v2.0 (fitness: 0.91, age: 1 hour)
- Select: v2.0 (highest fitness, most recent)
6. Execute: Return cached v2.0 result (INSTANT)
7. Update metrics:
- validated_translator.usage_count++
- validated_translator.cache_hits++
- validated_translator.avg_latency_ms (no change, cache hit)
系统:
每一步都优化代码路径 :
让我们确切地解释一下为什么 定向合成进化 并不只是“紧跟版本 ” :
作为 Genes 的工具 :
class Tool:
"""A tool is a genetic unit that:
- Replicates (used by multiple workflows)
- Mutates (evolves to new versions)
- Competes (fitness-based selection)
- Specializes (variants emerge)
- Dies (low-fitness tools pruned)
"""
# Genetic material
definition_hash: str # "DNA"
version: str # Generational marker
lineage: List[str] # Ancestry
# Replication rate
usage_count: int # How many "offspring"
workflows_using: int # Spread through ecosystem
# Fitness
quality_score: float # Survival metric
performance_metrics: Dict # Selection pressure
# Mutation
breaking_changes: List # Genetic incompatibility
evolution_history: List # Mutation record
直接演变:
# Unlike natural selection (random mutations),
# DSE uses DIRECTED mutations based on data:
def evolve_tool(tool_id: str):
"""Directed evolution with learning."""
# Analyze failure modes across ALL usage
failures = analyze_tool_failures(tool_id)
# "This tool fails when input > 5000 tokens"
# Generate targeted improvement
improvement_spec = create_improvement_plan(failures)
# "Add chunking for inputs > 5000 tokens"
# Mutate with purpose
new_version = apply_directed_mutation(tool_id, improvement_spec)
# Test fitness
fitness_improvement = a_b_test(old_version, new_version)
# Selection
if fitness_improvement > threshold:
promote_version(new_version) # Survives
else:
discard_version(new_version) # Dies
"黄金池":
53 tools in registry (current generation)
├── 27 LLM tools (specialist genes)
├── 19 executable tools (utility genes)
├── 3 OpenAPI tools (external interface genes)
├── 4 composite tools (multi-gene complexes)
Total genetic variations across versions: ~200+
Active in current generation: 53
Archived (evolutionary dead-ends): ~150
遗传传播可视化:
graph TB
T1["nmt_translator v1.0<br/>Fitness: 0.73<br/>Usage: 5"] --> T2["nmt_translator v2.0<br/>Fitness: 0.88<br/>Usage: 127"]
T2 --> W1["validated_translator<br/>Composite: nmt + quality<br/>Fitness: 0.91"]
T2 --> W2["article_translator<br/>Uses: nmt<br/>Fitness: 0.87"]
T2 --> W3["batch_translator<br/>Uses: nmt<br/>Fitness: 0.83"]
W1 --> U1["multilingual_content<br/>Uses: validated<br/>Fitness: 0.84"]
W1 --> U2["doc_localizer<br/>Uses: validated<br/>Fitness: 0.88"]
T2 -.->|Mutation| T3["nmt_translator v3.0<br/>Specialization: articles<br/>Fitness: 0.94"]
T3 --> W2
style T1 fill:#ffcccc
style T2 fill:#ccffcc
style T3 fill:#ccccff
style W1 fill:#ffffcc
style W2 fill:#ffffcc
style W3 fill:#ffffcc
style U1 fill:#ffeecc
style U2 fill:#ffeecc
遗传继承:
# Child tool inherits from parent
article_quality_checker:
parent: translation_quality_checker
inherited_attributes:
- quality_metrics
- validation_patterns
- error_detection
mutations:
- "Specialized for article-length text"
- "Added domain-specific checks"
- "40% faster (optimized for articles)"
fitness_inheritance:
parent_fitness: 0.91
child_fitness: 0.94 # Improvement!
selection_advantage:
- Chosen over parent for article tasks
- Parent still used for general translation
不错吧?
是啊,这真的太野了
我们建立了一个系统,其中:
这不是比喻
这是真正的定向合成进化
使用此系统数周后:
如果工具能够:
接下来呢?
以下是第七部分没有充分解释的:
进化的工作流程? 构成工作流程的工具? 它们也会演变。 管理进化的系统? 跟踪健身的测量标准?
这是工具 一路下来。
和每一个:
我们没有造代号发电机
我们建立了一个自我扩展、自我优化、自我文件工具箱, 碰巧可以生成代码。
区别问题。
因为当工具变成进化单元时 当它们追踪自己的身体状况时 当它们繁殖、变异和竞争时...
你没有工具箱
你有生态学
生态进化。
但是当进化破坏事物时会发生什么呢? 当工具突变引入关键错误时? 当优化使工具变得更差而不是更好时?
这就是第九部分的来处,我们探索 通过直系和觉知的补丁自愈工具不光是进化的系统, 它们记住每一个失败, 使用失败的分支, 并传播这些知识, 以防止整个生态系统发生类似的错误。
当你的工具可以打破自己, 你的系统应该记住为什么, 并且永远不要重犯错误。
仓库 : 大多为luccid.dse
密钥文件 :
src/tools_manager.py (2 293行) - 核心工具管理src/rag_integrated_tools.py (562行) - RAG整合src/openapi_tool.py (313行) - OpenAPI 支持src/model_selector_tool.py (460行) - 示范甄选tools/index.json (5,464行) - 工具登记tools/llm/*.yaml (27个工具) - 法学硕士专家定义tools/executable/*.yaml (19个工具) - 可操作工具tools/openapi/*.yaml (3个工具) - API整合文 件:
LLMS_AS_TOOLS.md - LLM甄选制度WORKFLOW_DOCUMENTATION_TOOL.md - 自动文档CHAT_TOOLS_GUIDE.md - 工具使用指南TOOL_PACKAGING.md - 工具开发指南系列导航:
这是语义情报系列的第八部分。第七部分展示了DSE的总体结构。这篇文章揭示了隐藏的复杂性:系统中的每一个工具都跟踪了使用情况、演变了实施过程、缓存结果并参与了以健身为基础的选择。工具箱不仅仅是一种资源 — — 它是一种进化的生态,可以扩展、优化和记录本身。工具产生工具。工具可以改进工具。整个系统随着时间的流逝变得更为智能。
代码是真实的,在Ollama本地运行, 真实的跟踪度量, 并且实际上在演化。它是实验性的,偶尔的不稳定, 并且绝对是“虚拟编码 ” 。 但是工具,跟踪和进化是有效的。工具本身会成长。
这些探索与关于新兴AI和自我优化系统的影响的Sci-fi小说“Michael”相连接。这里描述的工具是真正的实施过程,展示进化压力是如何创造专业化的,健身功能如何指导选择,自我改进系统如何自然地发展与生态相似的特性。这是否导致第六部分的行星级工具网络或完全出乎意料的事物,还有待观察。这就是为什么它是一个实验。
标记 : #AI #Tools #RAG #UsageTracking #Evolution #Fitness #Caching #Versioning #Ollama #Python #EmergentIntelligence #SelfOptimization #ToolEcology
© 2026 Scott Galloway — Unlicense — All content and source code on this site is free to use, copy, modify, and sell.