这是一个基于 Kilo Code 应用的 Python 版本智能字符串替换系统，提供从基础文本替换到 LLM 集成的完整解决方案。
- 智能字符串替换: 支持精确匹配、正则表达式和模糊匹配
- ApplyDiff 格式支持: 完整兼容 Kilo Code 的 apply_diff 工具格式
- LLM 集成: 支持 Anthropic Claude、OpenAI GPT、OpenRouter 等多个 LLM 提供商
- 批量处理: 支持并发和顺序批量任务执行
- 流式处理: 实时处理 LLM 输出和响应
- 模糊匹配算法: 基于 Levenshtein 距离的智能文本匹配
graph TD
A[用户请求] --> B{请求类型}
B -->|基础替换| C[SearchReplaceEngine]
B -->|ApplyDiff格式| D[ApplyDiffParser]
B -->|LLM集成| E[Orchestrator]
C --> C1[参数验证]
C1 --> C2[文件读取]
C2 --> C3[文本处理]
C3 --> C4[结果生成]
C4 --> C5[文件保存]
D --> D1[格式解析]
D1 --> D2[模糊匹配]
D2 --> D3[批量替换]
D3 --> D4[差异报告]
E --> E1[初始化]
E1 --> E2[上下文准备]
E2 --> E3[消息增强]
E3 --> E4[任务循环]
E4 --> F[TaskLoop]
F --> F1[提示词工程]
F1 --> F2[LLM交互]
F2 --> F3[响应解析]
F3 --> F4[工具执行]
F4 --> F5{任务完成?}
F5 -->|否| F1
F5 -->|是| G[返回结果]
F4 --> H[ToolExecutor]
H --> H1{工具类型}
H1 -->|search_replace| I[SearchReplaceEngine]
H1 -->|apply_diff| J[ApplyDiffParser]
H1 -->|其他工具| K[其他执行器]
I --> L[文件操作]
J --> L
K --> L
L --> M[执行结果]
M --> F5
用户请求
↓
LLM集成编排器 (Orchestrator)
↓
任务循环引擎 (TaskLoop)
↓
工具执行器 (ToolExecutor)
↓
核心引擎 (SearchReplaceEngine/ApplyDiffParser)
↓
文件系统操作
python_replace_string/
├── README.md # 本文件
├── core/ # 核心实现
│ ├── __init__.py
│ ├── search_replace.py # 搜索替换核心逻辑
│ ├── diff_strategy.py # 差异策略实现
│ ├── fuzzy_match.py # 模糊匹配算法
│ └── text_utils.py # 文本处理工具
├── llm_integration/ # LLM集成层
│ ├── __init__.py
│ ├── api_handlers/ # API处理器
│ │ ├── __init__.py
│ │ ├── base.py # 基础API处理器
│ │ ├── anthropic.py # Anthropic API处理器
│ │ ├── openai.py # OpenAI API处理器
│ │ └── openrouter.py # OpenRouter API处理器
│ ├── orchestrator.py # LLM集成编排器
│ ├── task_loop.py # 任务循环引擎
│ ├── streaming.py # 流式响应处理
│ ├── prompt_engineer.py # 提示词工程
│ ├── message_enhancer.py # 消息增强器
│ └── context_manager.py # 上下文管理器
├── formats/ # 格式化相关
│ ├── __init__.py
│ ├── apply_diff_format.py # apply_diff格式处理
│ ├── code_generation.py # 代码生成格式要求
│ └── template_system.py # 模板系统
├── examples/ # 使用示例
│ ├── __init__.py
│ ├── basic_usage.py # 基本使用示例
│ ├── advanced_usage.py # 高级使用示例
│ └── llm_integration_examples.py # LLM集成示例
└── tests/ # 测试文件
├── __init__.py
├── test_search_replace.py
├── test_diff_strategy.py
├── test_fuzzy_match.py
└── test_llm_integration.py
核心特性：
- ✅ 精确文本匹配和正则表达式匹配
- ✅ 大小写敏感/不敏感搜索
- ✅ 指定行范围的替换
- ✅ 实时差异预览功能
- ✅ UTF-8 编码支持
工作流程：
参数验证 → 文件读取 → 文本处理 → 结果生成 → 文件保存
核心特性：
- ✅ 完整支持 Kilo Code 的 apply_diff 工具格式
- ✅ SEARCH/REPLACE 块格式解析
- ✅ 多文件批量操作
- ✅ 行号提示支持 (
:start_line:) - ✅ XML 和文本格式双重支持
格式示例：
<<<<<<< SEARCH
:start_line:10
-------
def old_function():
return "old"
=======
def new_function():
return "new"
>>>>>>> REPLACE
算法特性：
- 🧠 基于 Levenshtein 距离的相似度计算
- 🎯 中间向外搜索策略，提高匹配效率
- 📝 智能引号和特殊字符标准化
- ⚡ 可配置相似度阈值
匹配流程：
graph TD
A[输入搜索文本] --> B[文本标准化]
B --> C[中间向外搜索]
C --> D[计算Levenshtein距离]
D --> E[相似度评分]
E --> F{达到阈值?}
F -->|是| G[返回匹配结果]
F -->|否| H[继续搜索]
H --> C
支持的 LLM 提供商：
- 🔵 Anthropic Claude: Claude-3.5-Sonnet, Claude-3-Haiku
- 🟢 OpenAI GPT: GPT-4, GPT-3.5-Turbo
- 🟡 OpenRouter: 多模型聚合平台
核心组件：
- Orchestrator: 总体编排和协调
- TaskLoop: 智能任务循环引擎
- PromptEngineer: 自动提示词优化
- MessageEnhancer: 用户输入增强
- ContextManager: 上下文和历史管理
- StreamingHandler: 实时流式响应处理
LLM 集成详细流程：
sequenceDiagram
participant User as 用户
participant Orch as Orchestrator
participant TaskLoop as TaskLoop
participant API as LLM_API
participant TE as ToolExecutor
participant SR as SearchReplaceEngine
User->>Orch: execute_request()
Orch->>TaskLoop: execute_task_loop()
%% --- 任务循环的单次执行 ---
TaskLoop->>API: send_request()
API-->>TaskLoop: llm_response
TaskLoop->>TE: execute_tool()
TE->>SR: perform_replacement()
SR-->>TE: result
TE-->>TaskLoop: tool_result
%% --- 任务完成的表示 ---
note over TaskLoop: 任务完成，退出循环
TaskLoop-->>Orch: task_result
Orch-->>User: final_result
from python_replace_string.core.search_replace import SearchReplaceEngine
from python_replace_string.formats.apply_diff_format import ApplyDiffParser
# 基本搜索替换
engine = SearchReplaceEngine()
result = engine.replace_in_file(
file_path="example.py",
search="old_function",
replace="new_function",
use_regex=False
)
# 使用apply_diff格式
parser = ApplyDiffParser()
diff_content = '''
<<<<<<< SEARCH
:start_line:1
-------
def old_function():
return "old"
=======
def new_function():
return "new"
>>>>>>> REPLACE
'''
result = parser.apply_diff("example.py", diff_content)import asyncio
from python_replace_string.llm_integration import create_orchestrator
async def main():
# 创建LLM集成编排器
orchestrator = await create_orchestrator(
api_provider="anthropic", # 或 "openai", "openrouter"
api_key="your-api-key",
workspace_path="./my_project"
)
# 执行智能代码修改任务
result = await orchestrator.execute_request(
user_input="将main.py中的hello_world函数改名为greet_user，并添加name参数",
mode="code"
)
print(f"任务执行成功: {result['success']}")
print(f"执行了 {result['statistics']['total_tool_executions']} 个工具操作")
# 运行示例
asyncio.run(main())# 批量执行多个任务
requests = [
{"user_input": "为所有Python文件添加文档字符串", "mode": "code"},
{"user_input": "优化import语句", "mode": "code"},
{"user_input": "添加类型注解", "mode": "code"}
]
results = await orchestrator.execute_batch_requests(requests)engine = SearchReplaceEngine()
result = engine.replace_in_file(
file_path="example.py",
search="old_function",
replace="new_function"
)执行步骤：
- 参数验证: 检查搜索内容、替换内容和文件路径
- 文件读取: 使用 UTF-8 编码读取文件内容
- 文本处理: 根据参数创建搜索模式并执行替换
- 结果生成: 计算变化次数并生成差异报告
parser = ApplyDiffParser()
result = parser.apply_diff("example.py", diff_content)执行步骤：
- 格式解析: 识别 SEARCH/REPLACE 块格式和行号提示
- 模糊匹配: 使用 Levenshtein 距离算法和中间向外搜索策略
- 批量替换: 按顺序处理多个 SEARCH/REPLACE 块
- 差异报告: 生成详细的变更报告
- API 处理器初始化: 根据提供商创建对应的处理器
- 组件初始化: 初始化上下文管理器、消息增强器等
- 工作空间扫描: 扫描项目文件结构
- 上下文准备: 生成项目概览和上下文信息
- 消息增强: 改进用户输入质量（可选）
- 任务循环执行: 进入智能任务循环
while iteration < max_iterations and not task_complete:
1. 构建系统提示词
2. 发送请求到 LLM
3. 解析 LLM 响应
4. 执行工具调用
5. 检查任务完成状态
iteration += 1
- search_and_replace工具: 调用 SearchReplaceEngine 执行文本替换
- apply_diff工具: 调用 ApplyDiffParser 执行批量差异应用
- 其他工具: 根据工具类型调用相应的执行器
def find_best_match_middle_out(self, content: str, search_text: str):
"""中间向外搜索策略"""
lines = content.split('\n')
middle = len(lines) // 2
for offset in range(len(lines)):
# 向下搜索
if middle + offset < len(lines):
match = self._try_match_at_line(lines, middle + offset, search_text)
if match and match.similarity >= self.threshold:
return match
# 向上搜索
if middle - offset >= 0:
match = self._try_match_at_line(lines, middle - offset, search_text)
if match and match.similarity >= self.threshold:
return matchdef _perform_replacement(self, content: str, search: str, replace: str,
use_regex: bool, ignore_case: bool):
"""执行实际的替换操作"""
flags = re.IGNORECASE if ignore_case else 0
if use_regex:
pattern = re.compile(search, flags)
else:
pattern = re.compile(re.escape(search), flags)
return pattern.sub(replace, content)完整的格式要求和代码生成规范请参见：formats/code_generation.py
- 基础使用:
examples/basic_usage.py - 高级功能:
examples/advanced_usage.py - LLM集成:
examples/llm_integration_examples.py - API测试:
examples/api_test_example.py
我们提供了经过验证的 API 集成测试，包含完整的读取→交互→修改→保存流程：
- Base URL:
https://litellm.talesofai.cn/v1 - Model:
glm-4.5 - 测试结果: 4/4 项测试通过
# 安装依赖
pip install openai
# 运行API测试示例
cd python_replace_string/examples
python api_test_example.py- ✅ 基本连接: API 连通性验证
- ✅ 代码分析: 代码质量评估和建议
- ✅ 代码修改: 函数重构、类型注解、文档字符串
- ✅ 文件操作: 完整的文件读取、修改、保存流程
🚀 API 集成测试
📡 Base URL: https://litellm.talesofai.cn/v1
🤖 Model: glm-4.5
📝 创建测试文件: example.py
✅ 代码改进成功!
📄 改进后的代码:
----------------------------------------
def greet(name: str) -> None:
"""打印个性化问候语"""
print(f"Hello {name}")
def calculate(x: int, y: int) -> int:
"""计算两个整数的和"""
return x + y
----------------------------------------
🎉 API 集成测试成功!
系统使用基于 Levenshtein 距离的模糊匹配算法，结合中间向外搜索策略：
def find_best_match_middle_out(self, content: str, search_text: str):
"""中间向外搜索策略"""
lines = content.split('\n')
middle = len(lines) // 2
for offset in range(len(lines)):
# 向下搜索
if middle + offset < len(lines):
match = self._try_match_at_line(lines, middle + offset, search_text)
if match and match.similarity >= self.threshold:
return match
# 向上搜索
if middle - offset >= 0:
match = self._try_match_at_line(lines, middle - offset, search_text)
if match and match.similarity >= self.threshold:
return matchdef _perform_replacement(self, content: str, search: str, replace: str,
use_regex: bool, ignore_case: bool):
"""执行实际的替换操作"""
flags = re.IGNORECASE if ignore_case else 0
if use_regex:
pattern = re.compile(search, flags)
else:
pattern = re.compile(re.escape(search), flags)
return pattern.sub(replace, content)- 继承
BaseApiHandler类 - 实现必要的接口方法
- 注册到
ApiHandlerFactory
class CustomApiHandler(BaseApiHandler):
async def send_request(self, messages: List[Dict]) -> Dict:
# 实现自定义 API 调用逻辑
pass
async def initialize(self) -> None:
# 实现初始化逻辑
pass- 继承
ToolExecutor类 - 实现工具执行逻辑
- 添加到工具注册表
class CustomToolExecutor(ToolExecutor):
async def execute_custom_tool(self, tool_input: Dict) -> Dict:
# 实现自定义工具逻辑
return {"success": True, "result": "..."}- 内存优化: 大文件分块处理，避免内存溢出
- 并发处理: 支持异步 I/O 和并发任务执行
- 缓存机制: 智能缓存常用的正则表达式模式
- 错误恢复: 完善的错误处理和重试机制
- 流式处理: 实时处理 LLM 响应，提升用户体验
运行测试套件：
python -m pytest python_replace_string/tests/ -v测试覆盖：
- 基础字符串替换功能
- ApplyDiff 格式解析
- 模糊匹配算法
- LLM 集成组件
- 错误处理和边界情况
本项目基于 Kilo Code 应用开发，遵循相应的许可证条款。