第2315篇：Program of Thought——让AI生成可执行代码来辅助推理
2026/4/30大约 6 分钟
第2315篇：Program of Thought——让AI生成可执行代码来辅助推理
适读人群：AI应用工程师、数据分析系统开发者 | 阅读时长：约17分钟 | 核心价值：理解PoT的工程架构，构建以代码执行为推理核心的可靠AI分析系统
有一次，我给我们的AI数据分析助手问了一个问题："过去6个月里，哪个季度的销售增长率最高？"
AI洋洋洒洒地给出了一段分析，说Q2增长率最高，达到了23.5%。我去核实了一下，实际上Q3才是最高的，增长率是31.2%，AI完全算错了。
问题出在哪？LLM在做数值计算的时候，本质上是在做"文字游戏"——它会根据数字之间的语言模式来生成看起来合理的结果，而不是真正做数学运算。LLM不是计算器。
Program of Thought（PoT）的核心思想就是：让LLM做它擅长的事（自然语言理解、逻辑规划、代码生成），把它不擅长的事（精确计算、数据处理）交给真正的执行器（Python/Java代码）。
PoT vs CoT：核心区别
Chain of Thought：让LLM用自然语言一步步推理，包括数值计算。 Program of Thought：让LLM生成代码，在代码里做计算，然后把执行结果作为推理的一部分。
关键优势：
- 计算精确：代码是确定性的，23.5%就是23.5%
- 可审计：生成的代码本身就是推理过程的记录
- 可复现：同样的代码再执行一次结果相同
安全的代码执行沙箱
PoT最大的工程挑战是安全执行LLM生成的代码——你不能让LLM随便执行任意代码：
@Component
public class SecureCodeExecutor {
private static final long EXECUTION_TIMEOUT_MS = 5000; // 5秒超时
private static final long MAX_MEMORY_BYTES = 128 * 1024 * 1024; // 128MB内存限制
/**
* 在沙箱中执行Python代码
* 使用Docker容器隔离，确保安全
*/
public CodeExecutionResult executePython(String code, Map<String, Object> inputData) {
// 将输入数据注入代码的数据上下文
String injectedCode = injectInputData(code, inputData);
// 静态安全检查（黑名单过滤）
SecurityCheckResult secCheck = staticSecurityCheck(injectedCode);
if (!secCheck.isSafe()) {
return CodeExecutionResult.blocked("代码包含不安全操作：" + secCheck.reason());
}
// 在Docker沙箱中执行
return executeInDocker(injectedCode);
}
private SecurityCheckResult staticSecurityCheck(String code) {
// 黑名单：绝对不允许的操作
List<String> blacklistPatterns = List.of(
"import os", // 操作系统调用
"import sys", // 系统级操作
"import subprocess", // 子进程
"import socket", // 网络操作
"open(", // 文件操作（需特别授权）
"__import__", // 动态导入
"eval(", // 动态执行
"exec(", // 动态执行
"globals()", // 全局变量访问
"locals()" // 本地变量访问
);
for (String pattern : blacklistPatterns) {
if (code.contains(pattern)) {
return SecurityCheckResult.unsafe("包含禁止操作：" + pattern);
}
}
return SecurityCheckResult.safe();
}
private CodeExecutionResult executeInDocker(String code) {
// 构建Docker执行命令
List<String> command = List.of(
"docker", "run", "--rm",
"--network=none", // 禁止网络访问
"--memory=" + MAX_MEMORY_BYTES,
"--cpus=0.5", // 限制CPU
"--read-only", // 只读文件系统
"python:3.11-slim",
"python", "-c", code
);
ProcessBuilder pb = new ProcessBuilder(command);
pb.redirectErrorStream(true);
try {
Process process = pb.start();
// 等待执行，带超时
boolean completed = process.waitFor(EXECUTION_TIMEOUT_MS, TimeUnit.MILLISECONDS);
if (!completed) {
process.destroyForcibly();
return CodeExecutionResult.timeout("代码执行超时（" + EXECUTION_TIMEOUT_MS + "ms）");
}
int exitCode = process.exitValue();
String output = new String(process.getInputStream().readAllBytes());
if (exitCode != 0) {
return CodeExecutionResult.error("代码执行错误（exitCode=" + exitCode + "）：" + output);
}
return CodeExecutionResult.success(output.trim());
} catch (Exception e) {
return CodeExecutionResult.error("执行异常：" + e.getMessage());
}
}
/**
* 将Java对象数据注入到Python代码的变量中
*/
private String injectInputData(String code, Map<String, Object> inputData) {
if (inputData == null || inputData.isEmpty()) {
return code;
}
StringBuilder injection = new StringBuilder();
injection.append("import json\n");
for (Map.Entry<String, Object> entry : inputData.entrySet()) {
String jsonValue = new ObjectMapper().writeValueAsString(entry.getValue());
injection.append(entry.getKey()).append(" = ").append(jsonValue).append("\n");
}
return injection + "\n" + code;
}
}PoT代码生成的Prompt设计
@Component
public class PoTCodeGenerator {
private final ChatClient codeGenerationClient;
private static final String CODE_GENERATION_PROMPT = """
你需要生成Python代码来解答问题。
重要规则：
1. 只使用以下允许的库：pandas, numpy, statistics, datetime, json, math, collections
2. 不能使用网络、文件系统、系统调用
3. 所有输入数据已通过变量注入，直接使用变量名
4. 最后必须 print() 输出答案（支持多行print）
5. 代码要简洁，避免不必要的复杂性
6. 如果问题需要多个步骤，用注释说明每步的目的
可用的输入数据变量：
%s
生成代码，不要加任何解释，只输出纯Python代码块。
""";
public String generateCode(String question, Map<String, String> availableVariables) {
String variableDesc = availableVariables.entrySet().stream()
.map(e -> "- %s: %s".formatted(e.getKey(), e.getValue()))
.collect(Collectors.joining("\n"));
String response = codeGenerationClient.prompt()
.system(CODE_GENERATION_PROMPT.formatted(variableDesc))
.user("问题：" + question)
.call()
.content();
return extractCodeBlock(response);
}
private String extractCodeBlock(String response) {
// 提取```python ... ``` 中的代码
Pattern pattern = Pattern.compile("```python\\s*\\n([\\s\\S]*?)\\n```");
Matcher matcher = pattern.matcher(response);
if (matcher.find()) {
return matcher.group(1);
}
// 如果没有代码块标记，直接返回（假设整个响应就是代码）
return response.trim();
}
}完整的PoT执行流水线
@Service
public class ProgramOfThoughtPipeline {
private final PoTCodeGenerator codeGenerator;
private final SecureCodeExecutor codeExecutor;
private final ChatClient interpretationClient;
private static final int MAX_RETRY_ATTEMPTS = 2;
/**
* PoT完整流水线：生成代码 → 执行 → 解读结果 → 综合回答
*/
public PoTResult solve(String question, Map<String, Object> data) {
// 准备变量描述
Map<String, String> variableDesc = describeVariables(data);
String generatedCode = null;
CodeExecutionResult executionResult = null;
// 生成-执行-重试循环
for (int attempt = 0; attempt < MAX_RETRY_ATTEMPTS; attempt++) {
// 生成代码
if (attempt == 0) {
generatedCode = codeGenerator.generateCode(question, variableDesc);
} else {
// 重试：把错误信息反馈给LLM，让它修正代码
generatedCode = codeGenerator.regenerateWithError(
question, variableDesc, generatedCode, executionResult.errorMessage()
);
}
log.info("第{}次尝试，生成的代码：\n{}", attempt + 1, generatedCode);
// 执行代码
executionResult = codeExecutor.executePython(generatedCode, data);
if (executionResult.success()) {
break;
}
log.warn("代码执行失败（第{}次）：{}", attempt + 1, executionResult.errorMessage());
}
if (!executionResult.success()) {
// 代码方式失败，降级到纯LLM回答（带标注）
return degradeFallback(question, data, executionResult.errorMessage());
}
// 让LLM解读代码执行结果
String finalAnswer = interpretResult(question, generatedCode, executionResult.output());
return PoTResult.success(finalAnswer, generatedCode, executionResult.output());
}
private String interpretResult(String question, String code, String executionOutput) {
return interpretationClient.prompt()
.system("""
用户提出了一个问题，你通过执行代码得到了结果。
请基于代码执行结果，给出清晰、准确的回答。
不要重复解释代码，直接说结论。
""")
.user("""
原始问题：%s
执行的代码：
```python
%s
```
代码执行输出：
%s
请基于以上结果回答问题。
""".formatted(question, code, executionOutput))
.call()
.content();
}
private PoTResult degradeFallback(String question, Map<String, Object> data,
String errorReason) {
log.warn("PoT代码执行失败，降级到直接LLM回答。原因：{}", errorReason);
String fallbackAnswer = interpretationClient.prompt()
.system("请直接回答问题，但注意：你的数值计算可能有误差，请尽量保守估计。")
.user(question)
.call()
.content();
return PoTResult.degraded(
fallbackAnswer + "\n\n⚠️ 注意：此回答由于代码执行失败，数值计算部分可能不精确。",
errorReason
);
}
private Map<String, String> describeVariables(Map<String, Object> data) {
return data.entrySet().stream()
.collect(Collectors.toMap(
Map.Entry::getKey,
e -> describeVariable(e.getValue())
));
}
private String describeVariable(Object value) {
if (value instanceof List<?> list) {
return "List，包含%d个元素".formatted(list.size());
} else if (value instanceof Map<?, ?> map) {
return "Dict，包含键：" + map.keySet();
} else {
return value.getClass().getSimpleName() + "类型";
}
}
}实际业务场景：销售数据分析
@Service
public class SalesAnalysisPoT {
private final ProgramOfThoughtPipeline potPipeline;
public String analyzeSalesQuestion(String question, SalesData salesData) {
// 准备数据
Map<String, Object> data = new HashMap<>();
data.put("monthly_sales", salesData.getMonthlySalesList()); // List<Map>
data.put("product_categories", salesData.getCategories()); // List<String>
data.put("region_data", salesData.getRegionMap()); // Map<String, List>
PoTResult result = potPipeline.solve(question, data);
if (result.isSuccess()) {
// 附上代码供审计
log.info("PoT计算代码：\n{}", result.generatedCode());
}
return result.answer();
}
}PoT最大的价值不只是准确性，而是可审计性。当业务用户质疑AI给出的数字时，你可以把生成的代码给他们看——这段代码是确定性的，他们可以自己运行来验证，而不是面对一个"黑盒"。这对数据分析、财务报告等高可信度要求的场景尤为重要。