LLM 应用开发系列

• Embedding 嵌入向量
• 向量数据库
• RAG 检索增强生成
• Prompt Engineering
• LangChain 框架
• Agent 智能体
• Function Calling
• 部署与优化
• 安全防护
• 对话记忆管理
• Streaming 流式输出
• LlamaIndex 框架
• 结构化输出
• 多模态 LLM 应用
• LLM 评估与测试
• 多模型路由
• 缓存策略
• 成本优化
• 监控与可观测性
• 高级 RAG 技术
• GraphRAG 知识图谱
• 长上下文处理
• LLM 微调实战
• 本地模型部署
• MCP 协议开发
• AI 搜索引擎
• 代码生成与执行
• 文档智能处理
• 语音交互系统
• 多租户架构
• Prompt 版本管理
• 混合检索策略
• 对话系统设计
• 内容审核与过滤
• 个性化推荐
• 知识库构建
• AI 写作助手
• 实时协作
• 边缘部署

prompt = """
你是一个高级 Python 工程师。请根据要求编写函数。

示例：
要求：计算列表的平均值，处理空列表。
代码：
def calculate_average(numbers: list[float]) -> float:
    if not numbers:
        return 0.0
    return sum(numbers) / len(numbers)

要求：{user_request}
代码：
"""

import re

def extract_code(text: str, language: str = "python") -> str:
    """从 Markdown 文本中提取代码块"""
    pattern = rf"```{language}\n(.*?)\n```"
    match = re.search(pattern, text, re.DOTALL)
    if match:
        return match.group(1).strip()
    
    # 如果没有 Markdown 标记，尝试直接返回
    if "def " in text or "import " in text:
        return text.strip()
    
    return ""

from e2b_code_interpreter import CodeInterpreter

def run_with_e2b(code: str):
    with CodeInterpreter() as sandbox:
        # 执行代码
        execution = sandbox.notebook.exec_cell(code)
        
        # 处理输出（文本、图表、错误）
        if execution.error:
            print(f"执行错误: {execution.error.name}: {execution.error.value}")
            return None, execution.error.value
            
        # 如果生成了图表，E2B 会返回 base64 编码的图片
        for result in execution.results:
            if result.png:
                save_chart(result.png)
                
        return execution.text, None

---

#### 核心技术：自我修复循环 (Self-Correction Loop)

这是 AI 程序员（如 Devin 或 OpenDevin）的核心逻辑。

#### 流程
1.  **生成**：LLM 生成代码。
2.  **执行**：在沙箱中运行代码。
3.  **反馈**：如果报错，将 **错误堆栈 (Traceback)** 和 **原始代码** 发回给 LLM。
4.  **修复**：LLM 分析错误并生成修复后的版本。
5.  **循环**：重复上述步骤，直到代码运行成功或达到最大重试次数。

```python
def autonomous_coder(task: str, max_retries: int = 3):
    code = generate_initial_code(task)
    
    for i in range(max_retries):
        output, error = run_in_sandbox(code)
        if not error:
            return code, output
            
        print(f"尝试 {i+1} 失败，正在修复...")
        code = generate_fix(code, error)
        
    raise Exception("无法修复代码")

def run_step(self, code: str):
    """执行代码并保持状态"""
    print(f"正在执行代码块...")
    execution = self.sandbox.notebook.exec_cell(code)
    
    if execution.error:
        return {"status": "error", "msg": execution.error.traceback}
    
    # 获取标准输出和结果
    return {
        "status": "success",
        "stdout": execution.logs.stdout,
        "results": [r.to_dict() for r in execution.results]
    }

def close(self):
    self.sandbox.close()

#### 2. 多文件项目管理

如果任务涉及多个文件（如 Web 爬虫或小型应用），我们可以在沙箱中动态创建文件结构。

```python
def setup_project(sandbox: CodeInterpreter):
    # 在沙箱中创建目录和文件
    sandbox.filesystem.make_dir("src")
    sandbox.filesystem.write("src/utils.py", "def add(a, b): return a + b")
    sandbox.filesystem.write("main.py", "from src.utils import add\nprint(add(1, 2))")
    
    # 执行主程序
    execution = sandbox.process.start("python3 main.py")
    return execution.wait().stdout

async def self_healing_code_gen(task: str, max_rounds: int = 3):
    code = await llm.generate(f"任务: {task}")
    
    for _ in range(max_rounds):
        result = sandbox.execute(code)
        if result.success:
            return result.output
        
        # 将错误信息喂回给 LLM
        code = await llm.generate(f"代码执行失败，错误如下：{result.error}\n请修复代码。")
    
    return "无法修复代码"

class AICodeAssistant:
    def __init__(self):
        self.llm = ChatOpenAI(model="gpt-4o")
        self.max_retries = 3

    def solve_task(self, task: str):
        current_task = task
        for i in range(self.max_retries):
            # 1. 生成代码
            code_response = self.llm.invoke(f"编写 Python 代码完成任务：{current_task}。只输出代码。")
            code = extract_code(code_response.content)
            
            print(f"尝试 {i+1}: 执行代码...")
            
            # 2. 执行代码
            result = execute_ai_code(code)
            
            if result["success"]:
                print("✅ 执行成功！")
                return result
            else:
                print(f"❌ 执行失败: {result['error']}")
                # 3. 自我修复：将错误反馈给 LLM
                current_task = f"""
                之前的代码执行失败了。
                任务：{task}
                错误类型：{result['error']}
                错误详情：{result['traceback']}
                请修复代码并重新提供。
                """
        
        return {"success": False, "message": "达到最大重试次数"}

def execute_and_get_charts(code: str):
    with CodeInterpreter() as sandbox:
        execution = sandbox.notebook.exec_cell(code)
        
        # E2B 会自动捕获 matplotlib/plotly 的输出
        charts = []
        for result in execution.results:
            if result.png:
                charts.append(result.png) # Base64 编码的图片
        
        return charts

def security_check(code: str) -> bool:
    forbidden_patterns = [
        r"import\s+os",
        r"import\s+subprocess",
        r"open\(",
        r"eval\(",
        r"exec\("
    ]
    for pattern in forbidden_patterns:
        if re.search(pattern, code):
            return False
    return True

test_prompt = f"""
针对以下函数编写 3 个 pytest 测试用例：
{generated_code}

要求：
1. 覆盖正常情况和边界情况。
2. 使用 assert 语句。
"""