201 lines
11 KiB
Python
201 lines
11 KiB
Python
from core.task_executor import TaskExecutor, TaskStatus
|
||
from typing import Dict, Any
|
||
import time
|
||
import json
|
||
import re
|
||
|
||
class TextAnalysisExecutor(TaskExecutor):
|
||
def __init__(self):
|
||
super().__init__(llm_model="deepseek-chat")
|
||
self.task_steps = [
|
||
{
|
||
"id": "input_validation",
|
||
"name": "Validate Input",
|
||
"required_info": ["text"],
|
||
"instruction": "Validate if the input text is not empty and contains valid Chinese characters. Check for proper UTF-8 encoding."
|
||
},
|
||
{
|
||
"id": "text_preprocessing",
|
||
"name": "Preprocess Text",
|
||
"required_info": ["text"],
|
||
"instruction": "Clean and preprocess the text by: 1) Normalizing Chinese punctuation, 2) Removing unnecessary whitespace while preserving sentence structure, 3) Standardizing traditional/simplified characters if needed."
|
||
},
|
||
{
|
||
"id": "generate_summary",
|
||
"name": "Generate Summary",
|
||
"required_info": ["preprocessed_text"],
|
||
"instruction": "Generate a concise summary in Chinese that captures the main points. Maintain the original language style and terminology."
|
||
},
|
||
{
|
||
"id": "extract_keywords",
|
||
"name": "Extract Keywords",
|
||
"required_info": ["preprocessed_text"],
|
||
"instruction": "Extract the most important Chinese keywords and key phrases from the text. Include both technical terms and contextual phrases."
|
||
},
|
||
{
|
||
"id": "final_analysis",
|
||
"name": "Final Analysis",
|
||
"required_info": ["summary", "keywords"],
|
||
"instruction": "Combine the summary and keywords into a comprehensive analysis report in Chinese. Structure the report with clear sections for summary, key points, and insights."
|
||
}
|
||
]
|
||
|
||
def validate_input(self, input_data: Dict[str, Any]) -> bool:
|
||
"""Validate specific input requirements for the text analysis task."""
|
||
if "text" not in input_data:
|
||
return False
|
||
text = input_data.get("text", "")
|
||
return isinstance(text, str) and len(text.strip()) > 0
|
||
|
||
async def execute_step(self, step_id: str, step_data: Dict[str, Any]) -> bool:
|
||
"""Execute a specific step of the text analysis task."""
|
||
try:
|
||
self.current_step = {
|
||
"id": step_id,
|
||
"name": step_data["name"],
|
||
"status": TaskStatus.IN_PROGRESS.value,
|
||
"progress": 0
|
||
}
|
||
|
||
# Get step instruction and input
|
||
instruction = step_data.get("instruction", "")
|
||
step_input = {}
|
||
|
||
# Prepare step-specific input
|
||
if step_id == "input_validation":
|
||
text = self.task_input.get("text", "")
|
||
# Validate Chinese text encoding
|
||
try:
|
||
text.encode('utf-8').decode('utf-8')
|
||
except UnicodeError:
|
||
return {"error": "Invalid text encoding. Please ensure the text is properly encoded in UTF-8."}
|
||
step_input = {"text": text}
|
||
|
||
elif step_id == "text_preprocessing":
|
||
text = self.task_input.get("text", "")
|
||
# Basic Chinese text preprocessing
|
||
# Normalize whitespace while preserving Chinese text structure
|
||
text = re.sub(r'\s+', ' ', text).strip()
|
||
# Normalize Chinese punctuation (simple example)
|
||
text = text.replace('，', ',').replace('。', '.').replace('：', ':')
|
||
step_input = {"text": text}
|
||
elif step_id == "generate_summary":
|
||
# Get the preprocessed text from previous step
|
||
prev_result = next(
|
||
(step["result"] for step in self.execution_path if step["step_id"] == "text_preprocessing"),
|
||
{}
|
||
)
|
||
step_input = {"text": prev_result.get("preprocessed_text", self.task_input.get("text", ""))}
|
||
elif step_id == "extract_keywords":
|
||
# Get the preprocessed text from previous step
|
||
prev_result = next(
|
||
(step["result"] for step in self.execution_path if step["step_id"] == "text_preprocessing"),
|
||
{}
|
||
)
|
||
step_input = {"text": prev_result.get("preprocessed_text", self.task_input.get("text", ""))}
|
||
elif step_id == "final_analysis":
|
||
# Get summary and keywords from previous steps
|
||
summary_result = next(
|
||
(step["result"] for step in self.execution_path if step["step_id"] == "generate_summary"),
|
||
{}
|
||
)
|
||
keywords_result = next(
|
||
(step["result"] for step in self.execution_path if step["step_id"] == "extract_keywords"),
|
||
{}
|
||
)
|
||
step_input = {
|
||
"summary": summary_result.get("summary", ""),
|
||
"keywords": keywords_result.get("keywords", [])
|
||
}
|
||
|
||
# Execute step using LLM
|
||
step_result = await self.llm_executor.execute_step(
|
||
step_instruction=instruction,
|
||
step_input=step_input
|
||
)
|
||
|
||
if not step_result.get("success", False):
|
||
raise Exception(f"Step failed: {step_result.get('error', 'Unknown error')}")
|
||
|
||
# Update execution path
|
||
self.execution_path.append({
|
||
"step_id": step_id,
|
||
"result": step_result.get("output", {})
|
||
})
|
||
|
||
# Update step status
|
||
self.current_step["status"] = TaskStatus.COMPLETED.value
|
||
self.current_step["progress"] = 100
|
||
|
||
# Create checkpoint if needed
|
||
if len(self.execution_path) % self.CHECKPOINT_INTERVAL == 0:
|
||
self.create_checkpoint()
|
||
|
||
return True
|
||
|
||
except Exception as e:
|
||
self.logger.error(f"Step {step_id} failed: {str(e)}")
|
||
self.current_step["status"] = TaskStatus.FAILED.value
|
||
return False
|
||
|
||
def main():
|
||
# Set console encoding to UTF-8
|
||
import sys, io
|
||
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
|
||
|
||
# Create the executor
|
||
executor = TextAnalysisExecutor()
|
||
|
||
# Sample Chinese text for analysis
|
||
sample_text = """
|
||
从 ChatGPT 到 Devin：AI 编程的四个发展阶段与范式转变 Koji：我们再聊一聊 AI 编程。编程领域今年取得了非常令人兴奋的进展。雨森一直有很强的框架归纳和总结能力。前不久你跟我分享过你提炼出来的 AI 编程发展四段论，要不要在播客里和大家分享一下？ 雨森：这其实是和很多朋友一起探讨得出的结果，是大家智慧的结晶。AI 编程从 ChatGPT 出现到现在也就两年出头的时间，但已经经历了四个阶段。 第一个阶段是让 AI 直接写代码，典型代表是早期的 ChatGPT、Claude。我们给它一个需求，比如「帮我写个贪吃蛇」，它就给出一段代码。在这个过程中，它既不知道我为什么要写贪吃蛇，也不知道代码运行情况如何。可能要我去本地编译运行后发现报错，再把错误告诉它，它才能给出调试后的结果。这时的 AI 完全就像一个只能通过邮件交流的笔友，是简单的问答模式。 第二阶段是以 GitHub Copilot 为代表，AI 开始拥有上下文，它可以把整个组织的代码库作为 context。这样 AI 就获得了大量新的背景信息。但这时用户还是需要手动把代码贴到 IDE 里面进行调试。我觉得这是 2.0 阶段，就是我们让 AI 拥有了 codebase 作为上下文。 2024 年一个非常大的进步是以 Cursor 为代表的编程 Copilot 的出现。它的核心理念是预测用户未来要写什么代码。根据你的代码库以及刚才写的代码，它预测你接下来要写什么代码、创建什么文件、做什么操作。这里面对于生成代码的质量和数量，以及文件的创建和修改都有很大提升。后来 Windsurf 还加入了对命令行操作的自动化，这样 AI 就能很好地使用我的电脑。原来的 AI 是在一张纸上写代码，我把代码抄走运行；现在 AI 可以在我的电脑上创建文件、执行命令行操作，进入到「我为你写」的阶段。 当我们觉得这已经很令人兴奋时，Devin 的出现带来了几个重要突破：首先，它可以异步工作。Cursor、Windsurf 这些工具虽然一步操作做的事情比较多，但仍然需要持续的注意力，即「我说一步它做一步」。而 Devin 可以持续工作，把用户的注意力释放出来。这是因为它多了一个 Planner，可以规划任务。 其次，它可以通过虚拟机执行更多操作，做更多调试工作。比如你写个网站，它可以自己用虚拟机去访问这个网站，检查前端后端的业务逻辑是否正确，并且可以随时打断和调整。大家用 Cursor 或者 ChatGPT 都知道，你无法在它输出的中间做调整，必须等它输出完后才能修改。但 Devin 就像真人一样，你可以在它完成任务时给出新指令，它会把这个结合到已有的 Planner 里调整计划。这就从「为你写」进化到了「为你做」。 总结一下这四个阶段：第一阶段是让 AI 写代码，代表是 ChatGPT；第二阶段是 AI 开放代码库，代表是 GitHub Copilot；第三阶段是 AI 可以自动写代码并执行，代表是 Cursor 和 Windsurf；第四阶段是 AI 虚拟员工，Devin 开创了一个很好的先例。
|
||
"""
|
||
|
||
# Prepare input
|
||
task_input = {
|
||
"text": sample_text
|
||
}
|
||
|
||
# Execute the task
|
||
import asyncio
|
||
result = asyncio.run(executor.execute(task_input))
|
||
|
||
# Print results with proper formatting
|
||
print("\n 文本分析结果:")
|
||
print("=" * 50)
|
||
|
||
# Print each step's result with proper formatting
|
||
for step in result.get("execution_path", []):
|
||
step_id = step["step_id"]
|
||
result_data = step["result"]
|
||
|
||
print(f"\n {step_id.upper()}:")
|
||
print("-" * 30)
|
||
|
||
if step_id == "input_validation":
|
||
print(" 输入验证完成")
|
||
elif step_id == "text_preprocessing":
|
||
print(" 文本预处理完成")
|
||
if "preprocessed_text" in result_data:
|
||
print("\n处理后的文本:")
|
||
print(result_data["preprocessed_text"])
|
||
elif step_id == "generate_summary":
|
||
print("\n 文本摘要:")
|
||
if "summary" in result_data:
|
||
print(result_data["summary"])
|
||
elif step_id == "extract_keywords":
|
||
print("\n 关键词:")
|
||
if "keywords" in result_data:
|
||
keywords = result_data["keywords"]
|
||
if isinstance(keywords, list):
|
||
print("、".join(keywords))
|
||
else:
|
||
print(keywords)
|
||
elif step_id == "final_analysis":
|
||
print("\n 最终分析:")
|
||
if "analysis" in result_data:
|
||
print(result_data["analysis"])
|
||
|
||
if __name__ == "__main__":
|
||
main()
|